Update Adapter.php

config/di.php

@@ -2,15 +2,12 @@
 
 declare(strict_types=1);
 
-use Yiisoft\Queue\AMQP\MessageSerializer;
-use Yiisoft\Queue\AMQP\MessageSerializerInterface;
 use Yiisoft\Queue\AMQP\QueueProvider;
 use Yiisoft\Queue\AMQP\QueueProviderInterface;
 use Yiisoft\Queue\AMQP\Settings\Queue;
 use Yiisoft\Queue\AMQP\Settings\QueueSettingsInterface;
 
 return [
-    MessageSerializerInterface::class => MessageSerializer::class,
     QueueProviderInterface::class => QueueProvider::class,
     QueueSettingsInterface::class => Queue::class,
 ];

src/Adapter.php

@@ -10,84 +10,114 @@
 use Yiisoft\Queue\AMQP\Exception\NotImplementedException;
 use Yiisoft\Queue\Cli\LoopInterface;
 use Yiisoft\Queue\Enum\JobStatus;
+use Yiisoft\Queue\Message\IdEnvelope;
 use Yiisoft\Queue\Message\MessageInterface;
+use Yiisoft\Queue\Message\MessageSerializerInterface;
 
-final class Adapter implements AdapterInterface
-{
+final class Adapter implements AdapterInterface {
+    /**
+     * @param QueueProviderInterface $queueProvider
+     * @param MessageSerializerInterface $serializer
+     * @param LoopInterface $loop
+     * @param string $channel
+     */
     public function __construct(
-        private QueueProviderInterface $queueProvider,
-        private MessageSerializerInterface $serializer,
-        private LoopInterface $loop,
+        private QueueProviderInterface              $queueProvider,
+        private readonly MessageSerializerInterface $serializer,
+        private readonly LoopInterface              $loop,
+        private string $channel
     ) {
+
     }
 
-    public function withChannel(string $channel): self
-    {
-        $instance = clone $this;
-        $instance->queueProvider = $this->queueProvider->withChannelName($channel);
+    /**
+     * @return string
+     */
+    public function getChannelName(): string {
+        return $this->channel;
+    }
 
-        return $instance;
+    /**
+     * @param string $channel
+     * @return $this
+     */
+    public function withChannel(string $channel): self {
+        if ($channel === $this->channel) {
+            return $this;
+        }
+
+        $new = clone $this;
+        $new->channel = $channel;
+
+        return $new;
     }
 
     /**
-     * @param callable(MessageInterface): bool  $handlerCallback
+     * @param callable(MessageInterface): bool $handlerCallback
      */
-    public function runExisting(callable $handlerCallback): void
-    {
+    public function runExisting(callable $handlerCallback): void {
         $channel = $this->queueProvider->getChannel();
-        (new ExistingMessagesConsumer($channel, $this->queueProvider
-            ->getQueueSettings()
-            ->getName(), $this->serializer))
-            ->consume($handlerCallback);
+        $queueName = $this->queueProvider->getQueueSettings()->getName();
+        $consumer = new ExistingMessagesConsumer(
+            $channel,
+            $queueName,
+            $this->serializer
+        );
+
+        $consumer->consume($handlerCallback);
     }
 
     /**
-     * @return never
+     * @param string|int $id
+     * @return JobStatus
      */
-    public function status(string $id): JobStatus
-    {
-        throw new NotImplementedException('Status check is not supported by the adapter ' . self::class . '.');
+    public function status(string|int $id): JobStatus {
+        throw new NotImplementedException(sprintf('Status check is not supported by the adapter %s.', self::class));
     }
 
-    public function push(MessageInterface $message): void
-    {
+    /**
+     * @param MessageInterface $message
+     * @return MessageInterface
+     */
+    public function push(MessageInterface $message): MessageInterface {
         $payload = $this->serializer->serialize($message);
         $amqpMessage = new AMQPMessage(
             $payload,
-            array_merge(['message_id' => uniqid(more_entropy: true)], $this->queueProvider->getMessageProperties())
+            array_merge(['message_id' => uniqid('', true)], $this->queueProvider->getMessageProperties())
         );
         $exchangeSettings = $this->queueProvider->getExchangeSettings();
-        $this->queueProvider
-            ->getChannel()
-            ->basic_publish(
-                $amqpMessage,
-                $exchangeSettings?->getName() ?? '',
-                $exchangeSettings ? '' : $this->queueProvider
-                    ->getQueueSettings()
-                    ->getName()
-            );
+        $channel = $this->queueProvider->getChannel();
+        $channel->basic_publish(
+            $amqpMessage,
+            $exchangeSettings?->getName() ?? '',
+            $exchangeSettings ? '' : $this->queueProvider
+                ->getQueueSettings()
+                ->getName()
+        );
         /** @var string $messageId */
         $messageId = $amqpMessage->get('message_id');
-        $message->setId($messageId);
+
+        return new IdEnvelope($message, $messageId);
     }
 
-    public function subscribe(callable $handlerCallback): void
-    {
+    /**
+     * @param callable $handlerCallback
+     * @return void
+     */
+    public function subscribe(callable $handlerCallback): void {
         $channel = $this->queueProvider->getChannel();
+        $queueName = $this->queueProvider->getQueueSettings()->getName();
+
         $channel->basic_consume(
-            $this->queueProvider
-                ->getQueueSettings()
-                ->getName(),
-            $this->queueProvider
-                ->getQueueSettings()
-                ->getName(),
+            $queueName,
+            $queueName,
             false,
             false,
             false,
             true,
             function (AMQPMessage $amqpMessage) use ($handlerCallback, $channel): void {
                 try {
-                    $handlerCallback($this->serializer->unserialize($amqpMessage->body));
+                    $handlerCallback($this->serializer->unserialize($amqpMessage->getBody()));
                     $channel->basic_ack($amqpMessage->getDeliveryTag());
                 } catch (Throwable $exception) {
                     $consumerTag = $amqpMessage->getConsumerTag();
@@ -105,13 +135,18 @@ function (AMQPMessage $amqpMessage) use ($handlerCallback, $channel): void {
         }
     }
 
-    public function getQueueProvider(): QueueProviderInterface
-    {
+    /**
+     * @return QueueProviderInterface
+     */
+    public function getQueueProvider(): QueueProviderInterface {
         return $this->queueProvider;
     }
 
-    public function withQueueProvider(QueueProviderInterface $queueProvider): self
-    {
+    /**
+     * @param QueueProviderInterface $queueProvider
+     * @return $this
+     */
+    public function withQueueProvider(QueueProviderInterface $queueProvider): self {
         $new = clone $this;
         $new->queueProvider = $queueProvider;
 

src/Exception/NoKeyInPayloadException.php

@@ -7,7 +7,7 @@
 use InvalidArgumentException;
 use Throwable;
 use Yiisoft\FriendlyException\FriendlyExceptionInterface;
-use Yiisoft\Queue\AMQP\MessageSerializerInterface;
+use Yiisoft\Queue\Message\MessageSerializerInterface;
 
 class NoKeyInPayloadException extends InvalidArgumentException implements FriendlyExceptionInterface
 {

src/ExistingMessagesConsumer.php

@@ -8,6 +8,7 @@
 use PhpAmqpLib\Message\AMQPMessage;
 use Throwable;
 use Yiisoft\Queue\Message\MessageInterface;
+use Yiisoft\Queue\Message\MessageSerializerInterface;
 
 /**
  * @internal
@@ -37,7 +38,7 @@ public function consume(callable $callback): void
             false,
             function (AMQPMessage $amqpMessage) use ($callback): void {
                 try {
-                    $message = $this->serializer->unserialize($amqpMessage->body);
+                    $message = $this->serializer->unserialize($amqpMessage->getBody());
                     if ($this->messageConsumed = $callback($message)) {
                         $this->channel->basic_ack($amqpMessage->getDeliveryTag());
                     }

tests/Integration/DelayMiddlewareTest.php

@@ -9,7 +9,6 @@
 use Psr\Log\LoggerInterface;
 use Yiisoft\Queue\Adapter\AdapterInterface;
 use Yiisoft\Queue\AMQP\Adapter;
-use Yiisoft\Queue\AMQP\MessageSerializer;
 use Yiisoft\Queue\AMQP\Middleware\DelayMiddleware;
 use Yiisoft\Queue\AMQP\QueueProvider;
 use Yiisoft\Queue\AMQP\Settings\Queue as QueueSettings;
@@ -18,6 +17,7 @@
 use Yiisoft\Queue\AMQP\Tests\Support\SimpleMessageHandler;
 use Yiisoft\Queue\Cli\LoopInterface;
 use Yiisoft\Queue\Cli\SignalLoop;
+use Yiisoft\Queue\Message\JsonMessageSerializer;
 use Yiisoft\Queue\Message\Message;
 use Yiisoft\Queue\Middleware\CallableFactory;
 use Yiisoft\Queue\Middleware\Push\MiddlewareFactoryPush;
@@ -44,7 +44,7 @@ public function testMainFlow(): void
                 $this->createConnection(),
                 new QueueSettings(),
             ),
-            new MessageSerializer(),
+            new JsonMessageSerializer(),
             new SignalLoop(),
         );
         $queue = $this->makeQueue($adapter);
@@ -75,7 +75,7 @@ public function testMainFlowWithFakeAdapter(): void
                 $this->createConnection(),
                 new QueueSettings(),
             ),
-            new MessageSerializer(),
+            new JsonMessageSerializer(),
             new SignalLoop(),
         );
         $queue = $this->makeQueue($adapter);

tests/Support/FakeAdapter.php

@@ -5,11 +5,11 @@
 namespace Yiisoft\Queue\AMQP\Tests\Support;
 
 use Yiisoft\Queue\Adapter\AdapterInterface;
-use Yiisoft\Queue\AMQP\MessageSerializerInterface;
 use Yiisoft\Queue\AMQP\QueueProviderInterface;
 use Yiisoft\Queue\Cli\LoopInterface;
 use Yiisoft\Queue\Enum\JobStatus;
 use Yiisoft\Queue\Message\MessageInterface;
+use Yiisoft\Queue\Message\MessageSerializerInterface;
 
 final class FakeAdapter implements AdapterInterface
 {
@@ -25,12 +25,12 @@ public function runExisting(callable $handlerCallback): void
         // TODO: Implement runExisting() method.
     }
 
-    public function status(string $id): JobStatus
+    public function status(string|int $id): JobStatus
     {
         // TODO: Implement status() method.
     }
 
-    public function push(MessageInterface $message): void
+    public function push(MessageInterface $message): MessageInterface
     {
         // TODO: Implement push() method.
     }

tests/Unit/ExchangeSettingsTest.php

@@ -7,10 +7,10 @@
 use PhpAmqpLib\Exchange\AMQPExchangeType;
 use PhpAmqpLib\Wire\AMQPTable;
 use Yiisoft\Queue\AMQP\Adapter;
-use Yiisoft\Queue\AMQP\MessageSerializer;
 use Yiisoft\Queue\AMQP\QueueProvider;
 use Yiisoft\Queue\AMQP\Settings\Exchange as ExchangeSettings;
 use Yiisoft\Queue\AMQP\Settings\Queue as QueueSettings;
+use Yiisoft\Queue\Message\JsonMessageSerializer;
 
 final class ExchangeSettingsTest extends UnitTestCase
 {
@@ -38,7 +38,7 @@ public function testCommonSettings(): void
                         ])
                     )
                 ),
-            new MessageSerializer(),
+            new JsonMessageSerializer(),
             $this->getLoop(),
         );
         $exchangeSettings = $adapter->getQueueProvider()->getExchangeSettings();