Update due to yiisoft/queue changes

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

@@ -4,27 +4,31 @@
 
 namespace Yiisoft\Queue\AMQP;
 
+use BackedEnum;
 use PhpAmqpLib\Message\AMQPMessage;
 use Throwable;
 use Yiisoft\Queue\Adapter\AdapterInterface;
 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
 {
     public function __construct(
         private QueueProviderInterface $queueProvider,
-        private MessageSerializerInterface $serializer,
-        private LoopInterface $loop,
+        private readonly MessageSerializerInterface $serializer,
+        private readonly LoopInterface $loop,
     ) {
     }
 
-    public function withChannel(string $channel): self
+    public function withChannel(BackedEnum|string $channel): self
     {
         $instance = clone $this;
-        $instance->queueProvider = $this->queueProvider->withChannelName($channel);
+        $channelName = is_string($channel) ? $channel : (string) $channel->value;
+        $instance->queueProvider = $this->queueProvider->withChannelName($channelName);
 
         return $instance;
     }
@@ -44,12 +48,12 @@
     /**
      * @return never
      */
-    public function status(string $id): JobStatus
+    public function status(int|string $id): JobStatus
     {
         throw new NotImplementedException('Status check is not supported by the adapter ' . self::class . '.');
     }
 
-    public function push(MessageInterface $message): void
+    public function push(MessageInterface $message): MessageInterface
     {
         $payload = $this->serializer->serialize($message);
         $amqpMessage = new AMQPMessage(
@@ -68,7 +72,8 @@
             );
         /** @var string $messageId */
         $messageId = $amqpMessage->get('message_id');
-        $message->setId($messageId);
+
+        return new IdEnvelope($message, $messageId);
     }
 
     public function subscribe(callable $handlerCallback): void
@@ -81,18 +86,18 @@
             $this->queueProvider
                 ->getQueueSettings()
                 ->getName(),
             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();
                     if ($consumerTag !== null) {
                         $channel->basic_cancel($consumerTag);
                     }
 
                     throw $exception;
@@ -117,4 +122,9 @@
 
         return $new;
     }
+
+    public function getChannel(): string
+    {
+        return $this->queueProvider->getQueueSettings()->getName();
+    }
 }

src/ExistingMessagesConsumer.php

@@ -8,6 +8,7 @@
 use PhpAmqpLib\Message\AMQPMessage;
 use Throwable;
 use Yiisoft\Queue\Message\MessageInterface;
+use Yiisoft\Queue\Message\MessageSerializerInterface as MessageSerializerInterfaceAlias;
 
 /**
  * @internal
@@ -17,9 +18,9 @@
     private bool $messageConsumed = false;
 
     public function __construct(
-        private AMQPChannel $channel,
-        private string $queueName,
-        private MessageSerializerInterface $serializer
+        private readonly AMQPChannel $channel,
+        private readonly string $queueName,
+        private readonly MessageSerializerInterfaceAlias $serializer
     ) {
     }
 
@@ -31,15 +32,15 @@
         $this->channel->basic_consume(
             $this->queueName,
             '',
             false,
             false,
             false,
             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());
                     }
                 } catch (Throwable $exception) {
                     $this->messageConsumed = false;
@@ -53,7 +54,7 @@
             }
         );
 
         do {
             $this->messageConsumed = false;
             $this->channel->wait(null, true);
         } while ($this->messageConsumed === true);

src/Middleware/DelayMiddleware.php

@@ -17,7 +17,7 @@
 
 final class DelayMiddleware implements DelayMiddlewareInterface
 {
-    public function __construct(private float $delayInSeconds, private bool $forcePersistentMessages = true)
+    public function __construct(private float $delayInSeconds, private readonly bool $forcePersistentMessages = true)
     {
     }
 

src/QueueProvider.php

@@ -18,7 +18,7 @@ final class QueueProvider implements QueueProviderInterface
     private ?AMQPChannel $channel = null;
 
     public function __construct(
-        private AbstractConnection $connection,
+        private readonly AbstractConnection $connection,
         private QueueSettingsInterface $queueSettings,
         private ?ExchangeSettingsInterface $exchangeSettings = null,
         private array $messageProperties = [],

src/Settings/Queue.php

@@ -5,12 +5,12 @@
 namespace Yiisoft\Queue\AMQP\Settings;
 
 use PhpAmqpLib\Wire\AMQPTable;
-use Yiisoft\Queue\QueueFactoryInterface;
+use Yiisoft\Queue\QueueInterface;
 
 final class Queue implements QueueSettingsInterface
 {
     public function __construct(
-        private string $queueName = QueueFactoryInterface::DEFAULT_CHANNEL_NAME,
+        private string $queueName = QueueInterface::DEFAULT_CHANNEL,
         private bool $passive = false,
         private bool $durable = false,
         private bool $exclusive = false,

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/Integration/TestCase.php

@@ -48,5 +48,10 @@ protected function queueListen(?string $queue = null): void
         $process = new Process($command);
         $this->processes[] = $process;
         $process->start();
+
+        usleep(500000);
+        if (!$process->isRunning()) {
+            throw new \Exception('Failed to start queue listener process');
+        }
     }
 }

tests/Support/ExtendedSimpleMessageHandler.php

@@ -11,7 +11,7 @@
  */
 final class ExtendedSimpleMessageHandler
 {
-    public function __construct(private FileHelper $fileHelper)
+    public function __construct(private readonly FileHelper $fileHelper)
     {
     }