Allow Symfony 6

composer.json

@@ -13,9 +13,9 @@
     "require": {
         "php": "^8.0.1",
         "ext-json": "*",
-        "nsq/nsq": "^0.5.1",
-        "symfony/framework-bundle": "^5.0",
-        "symfony/messenger": "^5.0"
+        "nsq/nsq": "0.7",
+        "symfony/framework-bundle": "^5.0 || ^6.0",
+        "symfony/messenger": "^5.0 || ^6.0"
     },
     "require-dev": {
         "ergebnis/composer-normalize": "^2.13"
@@ -25,5 +25,10 @@
             "Nsq\\NsqBundle\\": "src/"
         }
     },
-    "minimum-stability": "dev"
+    "minimum-stability": "dev",
+    "config": {
+        "allow-plugins": {
+            "ergebnis/composer-normalize": true
+        }
+    }
 }

src/DependencyInjection/NsqExtension.php

@@ -14,7 +14,7 @@ class NsqExtension extends Extension
      */
     public function load(array $configs, ContainerBuilder $container): void
     {
-        $loader = new Loader\YamlFileLoader($container, new FileLocator(__DIR__.'/../Resources/config'));
-        $loader->load('services.yml');
+        $loader = new Loader\PhpFileLoader($container, new FileLocator(__DIR__.'/../config'));
+        $loader->load('services.php');
     }
 }

src/EventListener/AckUnrecoverableMessageListener.php

@@ -0,0 +1,39 @@
+<?php
+declare(strict_types=1);
+
+namespace Nsq\NsqBundle\EventListener;
+
+use Nsq\NsqBundle\Messenger\NsqReceivedStamp;
+use Symfony\Component\EventDispatcher\EventSubscriberInterface;
+use Symfony\Component\Messenger\Event\WorkerMessageFailedEvent;
+use Symfony\Component\Messenger\Exception\UnrecoverableExceptionInterface;
+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->getThrowable() instanceof UnrecoverableExceptionInterface) {
+            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,72 @@
+<?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;
+
+        if (null === $envelope->last(NsqReceivedStamp::class)) {
+            $encodedMessage = $this->serializer->encode($envelope);
+            $encodedMessage = json_encode($encodedMessage, JSON_THROW_ON_ERROR | JSON_UNESCAPED_UNICODE);
+
+            wait($producer->publish($this->topic, $encodedMessage, $delay));
+        } else {
+            $message = NsqReceivedStamp::getMessageFromEnvelope($envelope);
+
+            if (!$message->isProcessed()) {
+                wait($message->requeue($delay));
+            }
+        }
+
+        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,36 +4,17 @@ declare(strict_types=1);
 
 namespace Nsq\NsqBundle\Messenger;
 
-use Amp\Loop;
-use Amp\Promise;
-use Generator;
-use JsonException;
 use Nsq\Config\ClientConfig;
-use Nsq\Message;
-use Nsq\Producer;
-use Nsq\Reader;
 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\Promise\wait;
-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 ?Reader $reader = null;
-
-    /**
-     * @var Promise<Message>|null
-     */
-    private ?Promise $deferred = null;
+    private ?NsqSender $sender = null;
 
     public function __construct(
         private string $address,
@@ -45,78 +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
     {
-        $reader = $this->getReader();
-
-        $promise = $this->deferred ??= $reader->consume();
-
-        /** @var Message|null $message */
-        $message = null;
-        Loop::run(function () use (&$message, $promise): Generator {
-            Loop::delay(500, static function () {
-                Loop::stop();
-            });
-
-            $message = yield $promise;
-        });
-
-        if (null === $message) {
-            return [];
-        }
-
-        $this->deferred = null;
-
-        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),
-            ),
-        ];
+        return ($this->receiver ?? $this->getReceiver())->get();
     }
 
     /**
@@ -124,9 +39,7 @@ final class NsqTransport implements TransportInterface
      */
     public function ack(Envelope $envelope): void
     {
-        $message = NsqReceivedStamp::getMessageFromEnvelope($envelope);
-
-        wait($message->finish());
+        ($this->receiver ?? $this->getReceiver())->ack($envelope);
     }
 
     /**
@@ -134,40 +47,37 @@ final class NsqTransport implements TransportInterface
      */
     public function reject(Envelope $envelope): void
     {
-        $message = NsqReceivedStamp::getMessageFromEnvelope($envelope);
-
-        wait($message->finish());
+        ($this->receiver ?? $this->getReceiver())->reject($envelope);
     }
 
-    private function getProducer(): Producer
+    /**
+     * {@inheritdoc}
+     */
+    public function send(Envelope $envelope): Envelope
     {
-        if (null === $this->producer) {
-            $this->producer = new Producer(
-                $this->address,
-                $this->clientConfig,
-                $this->logger,
-            );
-        }
-
-        wait($this->producer->connect());
-
-        return $this->producer;
+        return ($this->sender ?? $this->getSender())->send($envelope);
     }
 
-    private function getReader(): Reader
+    private function getReceiver(): NsqReceiver
     {
-        if (null === $this->reader) {
-            $this->reader = new Reader(
-                $this->address,
-                $this->topic,
-                $this->channel,
-                $this->clientConfig,
-                $this->logger,
-            );
-        }
+        return $this->receiver = new NsqReceiver(
+            $this->address,
+            $this->topic,
+            $this->channel,
+            $this->clientConfig,
+            $this->serializer,
+            $this->logger,
+        );
+    }
 
-        wait($this->reader->connect());
-
-        return $this->reader;
+    private function getSender(): NsqSender
+    {
+        return $this->sender = new NsqSender(
+            $this->address,
+            $this->topic,
+            $this->clientConfig,
+            $this->serializer,
+            $this->logger,
+        );
     }
 }

src/Messenger/NsqTransportFactory.php

@@ -29,20 +29,23 @@ 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 = [];
-        if (isset($parsedUrl['query'])) {
-            parse_str($parsedUrl['query'], $nsqOptions);
+        $query = [];
+        if (isset($components['query'])) {
+            parse_str($components['query'], $query);
         }
 
-        $address = sprintf('tcp://%s:%s', $parsedUrl['host'] ?? 'nsqd', $parsedUrl['port'] ?? 4150);
+        $nsqOptions = $query + $options;
+        $nsqOptions['rdyCount'] = 1;
+
+        $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

@@ -1,5 +0,0 @@
-services:
-    Nsq\NsqBundle\Messenger\NsqTransportFactory:
-        tags:
-            - 'messenger.transport_factory'
-            - 'container.no_preload'

src/config/services.php

@@ -0,0 +1,18 @@
+<?php
+
+declare(strict_types=1);
+
+use Symfony\Component\DependencyInjection\Loader\Configurator\ContainerConfigurator;
+
+return static function (ContainerConfigurator $configurator): void {
+    $services = $configurator->services();
+
+    $services->set(Nsq\NsqBundle\Messenger\NsqTransportFactory::class)
+        ->tag('messenger.transport_factory')
+        ->tag('container.no_preload')
+        ;
+
+    $services->set(\Nsq\NsqBundle\EventListener\AckUnrecoverableMessageListener::class)
+        ->tag('kernel.event_subscriber')
+        ;
+};