nsq/nsq-bundle
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

Add delay support

Browse Source
This commit is contained in:
Konstantin Grachev
2021-02-15 22:40:45 +03:00
parent bbbc9da3e0
commit e824d643c9
1 changed files with 10 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files

11
src/Messenger/NsqTransport.php
View File

@@ -12,6 +12,7 @@ use Nsq\Subscriber;
use Psr\Log\LoggerInterface; use Psr\Log\LoggerInterface;
use Symfony\Component\Messenger\Envelope; use Symfony\Component\Messenger\Envelope;
use Symfony\Component\Messenger\Exception\MessageDecodingFailedException; use Symfony\Component\Messenger\Exception\MessageDecodingFailedException;
use Symfony\Component\Messenger\Stamp\DelayStamp;
use Symfony\Component\Messenger\Stamp\TransportMessageIdStamp; use Symfony\Component\Messenger\Stamp\TransportMessageIdStamp;
use Symfony\Component\Messenger\Transport\Serialization\PhpSerializer; use Symfony\Component\Messenger\Transport\Serialization\PhpSerializer;
use Symfony\Component\Messenger\Transport\Serialization\SerializerInterface; use Symfony\Component\Messenger\Transport\Serialization\SerializerInterface;
@@ -48,7 +49,15 @@ final class NsqTransport implements TransportInterface
{ {
$encodedMessage = $this->serializer->encode($envelope->withoutAll(NsqReceivedStamp::class)); $encodedMessage = $this->serializer->encode($envelope->withoutAll(NsqReceivedStamp::class));
$this->producer->pub($this->topic, json_encode($encodedMessage, JSON_THROW_ON_ERROR)); /** @var DelayStamp|null $delayStamp */
$delayStamp = $envelope->last(DelayStamp::class);
$delay = null !== $delayStamp ? $delayStamp->getDelay() : null;
if (null === $delay) {
$this->producer->pub($this->topic, $encodedMessage['body']);
} else {
$this->producer->dpub($this->topic, $encodedMessage['body'], $delay);
}
return $envelope; return $envelope;
} }