Amphp (#2)

composer.json

@@ -11,8 +11,9 @@
         }
     ],
     "require": {
+        "php": "^8.0.1",
         "ext-json": "*",
-        "nsq/nsq": "0.4.1",
+        "nsq/nsq": "^0.5.1",
         "symfony/framework-bundle": "^5.0",
         "symfony/messenger": "^5.0"
     },

src/Messenger/NsqTransport.php

@@ -4,49 +4,54 @@
 
 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\Subscriber;
+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\PhpSerializer;
 use Symfony\Component\Messenger\Transport\Serialization\SerializerInterface;
 use Symfony\Component\Messenger\Transport\TransportInterface;
-use Throwable;
+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 SerializerInterface $serializer;
+    private ?Producer $producer = null;
 
-    private ?LoggerInterface $logger;
+    private ?Reader $reader = null;
 
-    private ?Generator $generator = null;
+    /**
+     * @var Promise<Message>|null
+     */
+    private ?Promise $deferred = null;
 
     public function __construct(
-        private Producer $producer,
-        private Subscriber $subscriber,
+        private string $address,
         private string $topic,
         private string $channel,
-        SerializerInterface $serializer = null,
-        LoggerInterface $logger = null,
+        private ClientConfig $clientConfig,
+        private SerializerInterface $serializer,
+        private LoggerInterface $logger,
     ) {
-        $this->serializer = $serializer ?? new PhpSerializer();
-        $this->logger = $logger;
     }
 
     /**
      * {@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);
 
@@ -55,11 +60,13 @@ public function send(Envelope $envelope): Envelope
         $delay = null !== $delayStamp ? $delayStamp->getDelay() : null;
 
         if (null === $delay) {
-            $this->producer->pub($this->topic, $encodedMessage);
+            $promise = $producer->publish($this->topic, $encodedMessage);
         } else {
-            $this->producer->dpub($this->topic, $encodedMessage, $delay / 1000);
+            $promise = $producer->defer($this->topic, $encodedMessage, $delay);
         }
 
+        wait($promise);
+
         return $envelope;
     }
 
@@ -68,52 +75,46 @@ public function send(Envelope $envelope): Envelope
      */
     public function get(): iterable
     {
-        try {
-            $this->producer->receive(0); // keepalive, handle heartbeat messages
-        } catch (Throwable $e) {
-            $this->logger->error('Producer keepalive failed.', ['exception' => $e]);
-        }
+        $reader = $this->getReader();
 
-        $generator = $this->generator;
-        if (null === $generator) {
-            $this->generator = $generator = $this->subscriber->subscribe($this->topic, $this->channel);
-        } else {
-            try {
-                $generator->next();
-            } catch (Throwable $e) {
-                $this->logger->error('Consumer next failed.', ['exception' => $e]);
+        $promise = $this->deferred ??= $reader->consume();
 
-                return [];
-            }
-        }
+        /** @var Message|null $message */
+        $message = null;
+        Loop::run(function () use (&$message, $promise): Generator {
+            Loop::delay(500, static function () {
+                Loop::stop();
+            });
 
-        /** @var Message|null $nsqMessage */
-        $nsqMessage = $generator->current();
+            $message = yield $promise;
+        });
 
-        if (null === $nsqMessage) {
+        if (null === $message) {
             return [];
         }
 
+        $this->deferred = null;
+
         try {
-            $encodedEnvelope = json_decode($nsqMessage->body, true, 512, JSON_THROW_ON_ERROR);
+            $encodedEnvelope = json_decode($message->body, true, 512, JSON_THROW_ON_ERROR);
         } catch (JsonException $e) {
-            $nsqMessage->finish();
+            wait($message->finish());
 
             throw new MessageDecodingFailedException('', 0, $e);
         }
 
         try {
             $envelope = $this->serializer->decode($encodedEnvelope);
         } catch (MessageDecodingFailedException  $e) {
-            $nsqMessage->finish();
+            wait($message->finish());
 
             throw $e;
         }
 
         return [
             $envelope->with(
-                new NsqReceivedStamp($nsqMessage),
-                new TransportMessageIdStamp($nsqMessage->id),
+                new NsqReceivedStamp($message),
+                new TransportMessageIdStamp($message->id),
             ),
         ];
     }
@@ -125,7 +126,7 @@ public function ack(Envelope $envelope): void
     {
         $message = NsqReceivedStamp::getMessageFromEnvelope($envelope);
 
-        $message->finish();
+        wait($message->finish());
     }
 
     /**
@@ -135,6 +136,38 @@ public function reject(Envelope $envelope): void
     {
         $message = NsqReceivedStamp::getMessageFromEnvelope($envelope);
 
-        $message->finish();
+        wait($message->finish());
+    }
+
+    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;
+    }
+
+    private function getReader(): Reader
+    {
+        if (null === $this->reader) {
+            $this->reader = new Reader(
+                $this->address,
+                $this->topic,
+                $this->channel,
+                $this->clientConfig,
+                $this->logger,
+            );
+        }
+
+        wait($this->reader->connect());
+
+        return $this->reader;
     }
 }

src/Messenger/NsqTransportFactory.php

@@ -4,10 +4,7 @@
 
 namespace Nsq\NsqBundle\Messenger;
 
-use Nsq\Consumer;
-use Nsq\Producer;
-use Nsq\Subscriber;
-use Psr\Log\LoggerAwareTrait;
+use Nsq\Config\ClientConfig;
 use Psr\Log\LoggerInterface;
 use Psr\Log\NullLogger;
 use Symfony\Component\Messenger\Exception\InvalidArgumentException;
@@ -20,7 +17,7 @@
 
 final class NsqTransportFactory implements TransportFactoryInterface
 {
-    use LoggerAwareTrait;
+    private LoggerInterface $logger;
 
     public function __construct(LoggerInterface $logger = null)
     {
@@ -42,20 +39,16 @@ public function createTransport(string $dsn, array $options, SerializerInterface
         }
 
         $address = sprintf('tcp://%s:%s', $parsedUrl['host'] ?? 'nsqd', $parsedUrl['port'] ?? 4150);
+        $topic = $nsqOptions['topic'] ?? 'symfony-messenger';
+        $channel = $nsqOptions['channel'] ?? 'default';
+
+        $clientConfig = new ClientConfig();
 
         return new NsqTransport(
-            new Producer(
-                address: $address,
-                logger: $this->logger,
-            ),
-            new Subscriber(
-                new Consumer(
-                    address: $address,
-                    logger: $this->logger,
-                )
-            ),
-            $nsqOptions['topic'] ?? 'symfony-messenger',
-            $nsqOptions['channel'] ?? 'default',
+            $address,
+            $topic,
+            $channel,
+            $clientConfig,
             $serializer,
             $this->logger,
         );