Extract receiver and sender from transport

src/Messenger/NsqReceiver.php

@@ -0,0 +1,120 @@
+<?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\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),
+            ),
+        ];
+    }
+
+    /**
+     * {@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);
+
+        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,70 @@
+<?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();
+
+        $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;
+    }
+
+    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,70 +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
     {
-        if ([] === $this->messages) {
+        return ($this->receiver ?? $this->getReceiver())->get();
-            $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),
-            ),
-        ];
     }
 
     /**
@@ -115,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());
     }
 
     /**
@@ -125,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,
+        );
     }
 }