Non retry messages must be acknowledged

composer.json

@@ -13,7 +13,7 @@
     "require": {
         "php": "^8.0.1",
         "ext-json": "*",
-        "nsq/nsq": "^0.5.1",
+        "nsq/nsq": "^0.6.2",
         "symfony/framework-bundle": "^5.0",
         "symfony/messenger": "^5.0"
     },

src/Messenger/AckUnrecoverableMessageListener.php

@@ -0,0 +1,37 @@
+<?php
+declare(strict_types=1);
+
+namespace Nsq\NsqBundle\Messenger;
+
+use Symfony\Component\EventDispatcher\EventSubscriberInterface;
+use Symfony\Component\Messenger\Event\WorkerMessageFailedEvent;
+use function Amp\Promise\wait;
+
+final class AckUnrecoverableMessageListener implements EventSubscriberInterface
+{
+    /**
+     * {@inheritdoc}
+     */
+    public static function getSubscribedEvents(): array
+    {
+        return [
+            WorkerMessageFailedEvent::class => ['onMessageFailed', 500],
+        ];
+    }
+
+    public function onMessageFailed(WorkerMessageFailedEvent $event): void
+    {
+        if ($event->willRetry()) {
+            return;
+        }
+
+        $envelope = $event->getEnvelope();
+        $message = NsqReceivedStamp::getMessageFromEnvelope($envelope);
+
+        if ($message->isProcessed()) {
+            return;
+        }
+
+        wait($message->finish());
+    }
+}

src/Messenger/NsqReceiver.php

@@ -0,0 +1,126 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Nsq\NsqBundle\Messenger;
+
+use JsonException;
+use Nsq\Config\ClientConfig;
+use Nsq\Consumer;
+use Nsq\Message;
+use Psr\Log\LoggerInterface;
+use Symfony\Component\Messenger\Envelope;
+use Symfony\Component\Messenger\Exception\MessageDecodingFailedException;
+use Symfony\Component\Messenger\Stamp\RedeliveryStamp;
+use Symfony\Component\Messenger\Stamp\TransportMessageIdStamp;
+use Symfony\Component\Messenger\Transport\Receiver\ReceiverInterface;
+use Symfony\Component\Messenger\Transport\Serialization\SerializerInterface;
+use function Amp\delay;
+use function Amp\Promise\wait;
+use function array_pop;
+use function json_decode;
+use const JSON_THROW_ON_ERROR;
+
+final class NsqReceiver implements ReceiverInterface
+{
+    private ?Consumer $consumer = null;
+
+    /**
+     * @var Message[]
+     */
+    private array $messages = [];
+
+    public function __construct(
+        private string $address,
+        private string $topic,
+        private string $channel,
+        private ClientConfig $clientConfig,
+        private SerializerInterface $serializer,
+        private LoggerInterface $logger,
+    ) {
+    }
+
+    /**
+     * {@inheritdoc}
+     */
+    public function get(): iterable
+    {
+        if ([] === $this->messages) {
+            $this->consume();
+
+            wait(delay(500));
+        }
+
+        $message = array_pop($this->messages);
+
+        if (null === $message) {
+            return [];
+        }
+
+        try {
+            $encodedEnvelope = json_decode($message->body, true, 512, JSON_THROW_ON_ERROR);
+        } catch (JsonException $e) {
+            wait($message->finish());
+
+            throw new MessageDecodingFailedException('', 0, $e);
+        }
+
+        try {
+            $envelope = $this->serializer->decode($encodedEnvelope);
+        } catch (MessageDecodingFailedException  $e) {
+            wait($message->finish());
+
+            throw $e;
+        }
+
+        return [
+            $envelope->with(
+                new NsqReceivedStamp($message),
+                new TransportMessageIdStamp($message->id),
+                new RedeliveryStamp($message->attempts - 1),
+            ),
+        ];
+    }
+
+    /**
+     * {@inheritdoc}
+     */
+    public function ack(Envelope $envelope): void
+    {
+        $message = NsqReceivedStamp::getMessageFromEnvelope($envelope);
+
+        wait($message->finish());
+    }
+
+    /**
+     * {@inheritdoc}
+     */
+    public function reject(Envelope $envelope): void
+    {
+        $message = NsqReceivedStamp::getMessageFromEnvelope($envelope);
+
+        if ($message->isProcessed()) {
+            return;
+        }
+
+        wait($message->finish());
+    }
+
+    private function consume(): void
+    {
+        if (null === $this->consumer) {
+            $this->consumer = new Consumer(
+                $this->address,
+                $this->topic,
+                $this->channel,
+                function (Message $message) {
+                    $this->messages[] = $message;
+                },
+                $this->clientConfig,
+                $this->logger,
+            );
+        }
+
+        wait($this->consumer->connect());
+    }
+}

src/Messenger/NsqSender.php

@@ -0,0 +1,78 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Nsq\NsqBundle\Messenger;
+
+use Nsq\Config\ClientConfig;
+use Nsq\Producer;
+use Psr\Log\LoggerInterface;
+use Symfony\Component\Messenger\Envelope;
+use Symfony\Component\Messenger\Stamp\DelayStamp;
+use Symfony\Component\Messenger\Transport\Sender\SenderInterface;
+use Symfony\Component\Messenger\Transport\Serialization\SerializerInterface;
+use function Amp\Promise\wait;
+use function json_encode;
+use const JSON_THROW_ON_ERROR;
+
+final class NsqSender implements SenderInterface
+{
+    private ?Producer $producer = null;
+
+    public function __construct(
+        private string $address,
+        private string $topic,
+        private ClientConfig $clientConfig,
+        private SerializerInterface $serializer,
+        private LoggerInterface $logger,
+    ) {
+    }
+
+    /**
+     * {@inheritdoc}
+     */
+    public function send(Envelope $envelope): Envelope
+    {
+        $producer = $this->getProducer();
+
+        /** @var DelayStamp|null $delayStamp */
+        $delayStamp = $envelope->last(DelayStamp::class);
+        $delay = null !== $delayStamp ? $delayStamp->getDelay() : 0;
+
+        $promise = null;
+
+        if (null !== $envelope->last(NsqReceivedStamp::class)) {
+            $message = NsqReceivedStamp::getMessageFromEnvelope($envelope);
+
+            if (!$message->isProcessed()) {
+                $promise = $message->requeue($delay);
+            }
+        }
+
+        if (null === $promise) {
+            $encodedMessage = $this->serializer->encode($envelope->withoutAll(NsqReceivedStamp::class));
+            $encodedMessage = json_encode($encodedMessage, JSON_THROW_ON_ERROR | JSON_UNESCAPED_UNICODE);
+
+            $promise = $producer->publish($this->topic, $encodedMessage, $delay);
+        }
+
+        wait($promise);
+
+        return $envelope;
+    }
+
+    private function getProducer(): Producer
+    {
+        if (null === $this->producer) {
+            $this->producer = new Producer(
+                $this->address,
+                $this->clientConfig,
+                $this->logger,
+            );
+        }
+
+        wait($this->producer->connect());
+
+        return $this->producer;
+    }
+}

src/Messenger/NsqTransport.php

@@ -4,35 +4,17 @@ declare(strict_types=1);
 
 namespace Nsq\NsqBundle\Messenger;
 
-use JsonException;
 use Nsq\Config\ClientConfig;
-use Nsq\Consumer;
-use Nsq\Message;
-use Nsq\Producer;
 use Psr\Log\LoggerInterface;
 use Symfony\Component\Messenger\Envelope;
-use Symfony\Component\Messenger\Exception\MessageDecodingFailedException;
-use Symfony\Component\Messenger\Stamp\DelayStamp;
-use Symfony\Component\Messenger\Stamp\TransportMessageIdStamp;
 use Symfony\Component\Messenger\Transport\Serialization\SerializerInterface;
 use Symfony\Component\Messenger\Transport\TransportInterface;
-use function Amp\delay;
-use function Amp\Promise\wait;
-use function array_pop;
-use function json_decode;
-use function json_encode;
-use const JSON_THROW_ON_ERROR;
 
 final class NsqTransport implements TransportInterface
 {
-    private ?Producer $producer = null;
+    private ?NsqReceiver $receiver = null;
 
-    private ?Consumer $consumer = null;
+    private ?NsqSender $sender = null;
-
-    /**
-     * @var Message[]
-     */
-    private array $messages = [];
 
     public function __construct(
         private string $address,
@@ -44,72 +26,12 @@ final class NsqTransport implements TransportInterface
     ) {
     }
 
-    /**
-     * {@inheritdoc}
-     */
-    public function send(Envelope $envelope): Envelope
-    {
-        $producer = $this->getProducer();
-
-        $encodedMessage = $this->serializer->encode($envelope->withoutAll(NsqReceivedStamp::class));
-        $encodedMessage = json_encode($encodedMessage, JSON_THROW_ON_ERROR | JSON_UNESCAPED_UNICODE);
-
-        /** @var DelayStamp|null $delayStamp */
-        $delayStamp = $envelope->last(DelayStamp::class);
-        $delay = null !== $delayStamp ? $delayStamp->getDelay() : null;
-
-        if (null === $delay) {
-            $promise = $producer->publish($this->topic, $encodedMessage);
-        } else {
-            $promise = $producer->defer($this->topic, $encodedMessage, $delay);
-        }
-
-        wait($promise);
-
-        return $envelope;
-    }
-
     /**
      * {@inheritdoc}
      */
     public function get(): iterable
     {
-        $message = array_pop($this->messages);
+        return ($this->receiver ?? $this->getReceiver())->get();
-
-        if (null === $message) {
-            $this->getConsumer();
-
-            wait(delay(500));
-
-            $message = array_pop($this->messages);
-        }
-
-        if (null === $message) {
-            return [];
-        }
-
-        try {
-            $encodedEnvelope = json_decode($message->body, true, 512, JSON_THROW_ON_ERROR);
-        } catch (JsonException $e) {
-            wait($message->finish());
-
-            throw new MessageDecodingFailedException('', 0, $e);
-        }
-
-        try {
-            $envelope = $this->serializer->decode($encodedEnvelope);
-        } catch (MessageDecodingFailedException  $e) {
-            wait($message->finish());
-
-            throw $e;
-        }
-
-        return [
-            $envelope->with(
-                new NsqReceivedStamp($message),
-                new TransportMessageIdStamp($message->id),
-            ),
-        ];
     }
 
     /**
@@ -117,9 +39,7 @@ final class NsqTransport implements TransportInterface
      */
     public function ack(Envelope $envelope): void
     {
-        $message = NsqReceivedStamp::getMessageFromEnvelope($envelope);
+        ($this->receiver ?? $this->getReceiver())->ack($envelope);
-
-        wait($message->finish());
     }
 
     /**
@@ -127,43 +47,37 @@ final class NsqTransport implements TransportInterface
      */
     public function reject(Envelope $envelope): void
     {
-        $message = NsqReceivedStamp::getMessageFromEnvelope($envelope);
+        ($this->receiver ?? $this->getReceiver())->reject($envelope);
-
-        wait($message->finish());
     }
 
-    private function getProducer(): Producer
+    /**
+     * {@inheritdoc}
+     */
+    public function send(Envelope $envelope): Envelope
     {
-        if (null === $this->producer) {
+        return ($this->sender ?? $this->getSender())->send($envelope);
-            $this->producer = new Producer(
-                $this->address,
-                $this->clientConfig,
-                $this->logger,
-            );
-        }
-
-        wait($this->producer->connect());
-
-        return $this->producer;
     }
 
-    private function getConsumer(): Consumer
+    private function getReceiver(): NsqReceiver
     {
-        if (null === $this->consumer) {
+        return $this->receiver = new NsqReceiver(
-            $this->consumer = new Consumer(
+            $this->address,
-                $this->address,
+            $this->topic,
-                $this->topic,
+            $this->channel,
-                $this->channel,
+            $this->clientConfig,
-                function (Message $message) {
+            $this->serializer,
-                    $this->messages[] = $message;
+            $this->logger,
-                },
+        );
-                $this->clientConfig,
+    }
-                $this->logger,
-            );
-        }
 
-        wait($this->consumer->connect());
+    private function getSender(): NsqSender
-
+    {
-        return $this->consumer;
+        return $this->sender = new NsqSender(
+            $this->address,
+            $this->topic,
+            $this->clientConfig,
+            $this->serializer,
+            $this->logger,
+        );
     }
 }

src/Messenger/NsqTransportFactory.php

@@ -29,20 +29,22 @@ final class NsqTransportFactory implements TransportFactoryInterface
      */
     public function createTransport(string $dsn, array $options, SerializerInterface $serializer): TransportInterface
     {
-        if (false === $parsedUrl = parse_url($dsn)) {
+        if (false === $components = parse_url($dsn)) {
             throw new InvalidArgumentException(sprintf('The given Nsq DSN "%s" is invalid.', $dsn));
         }
 
-        $nsqOptions = [];
+        $query = [];
-        if (isset($parsedUrl['query'])) {
+        if (isset($components['query'])) {
-            parse_str($parsedUrl['query'], $nsqOptions);
+            parse_str($components['query'], $query);
         }
 
-        $address = sprintf('tcp://%s:%s', $parsedUrl['host'] ?? 'nsqd', $parsedUrl['port'] ?? 4150);
+        $nsqOptions = $query + $options;
+
+        $address = sprintf('tcp://%s:%s', $components['host'], $components['port'] ?? $query['port'] ?? 4150);
         $topic = $nsqOptions['topic'] ?? 'symfony-messenger';
         $channel = $nsqOptions['channel'] ?? 'default';
 
-        $clientConfig = new ClientConfig();
+        $clientConfig = ClientConfig::fromArray($nsqOptions);
 
         return new NsqTransport(
             $address,

src/Resources/config/services.yml

@@ -3,3 +3,7 @@ services:
         tags:
             - 'messenger.transport_factory'
             - 'container.no_preload'
+
+    Nsq\NsqBundle\Messenger\AckUnrecoverableMessageListener:
+        tags:
+            - { name: kernel.event_subscriber }