Envelope

src/Adapter.php

@@ -10,84 +10,85 @@
 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,
     ) {
     }
 
     public function withChannel(string $channel): self
     {
-        $instance = clone $this;
-        $instance->queueProvider = $this->queueProvider->withChannelName($channel);
+        $new = clone $this;
+        $new->queueProvider = $this->queueProvider->withChannelName($channel);
 
-        return $instance;
+        return $new;
     }
 
     /**
-     * @param callable(MessageInterface): bool  $handlerCallback
+     * @param callable(MessageInterface): bool $handlerCallback
      */
     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
-     */
-    public function status(string $id): JobStatus
+    public function status(string|int $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(
             $payload,
             array_merge(['message_id' => uniqid(more_entropy: 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
     {
         $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));
                     $channel->basic_ack($amqpMessage->getDeliveryTag());
                 } catch (Throwable $exception) {
                     $consumerTag = $amqpMessage->getConsumerTag();

src/ExistingMessagesConsumer.php

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

src/Middleware/DelayMiddleware.php

@@ -43,20 +43,27 @@ public function processPush(PushRequest $request, MessageHandlerPushInterface $h
     {
         $adapter = $request->getAdapter();
         if (!$adapter instanceof Adapter) {
-            $type = get_debug_type($adapter);
-            $class = Adapter::class;
             throw new InvalidArgumentException(
-                "This middleware works only with the $class. $type given."
+                sprintf(
+                    'This middleware works only with the %s. %s given.',
+                    Adapter::class,
+                    get_debug_type($adapter)
+                )
             );
         }
 
         $queueProvider = $adapter->getQueueProvider();
-        $exchangeSettings = $this->getExchangeSettings($queueProvider->getExchangeSettings());
-        $queueSettings = $this->getQueueSettings($queueProvider->getQueueSettings(), $queueProvider->getExchangeSettings());
+        $originalExchangeSettings = $queueProvider->getExchangeSettings();
+        $delayedExchangeSettings = $this->getExchangeSettings($originalExchangeSettings);
+        $queueSettings = $this->getQueueSettings(
+            $queueProvider->getQueueSettings(),
+            $originalExchangeSettings
+        );
+
         $adapter = $adapter->withQueueProvider(
             $queueProvider
                 ->withMessageProperties($this->getMessageProperties($queueProvider))
-                ->withExchangeSettings($exchangeSettings)
+                ->withExchangeSettings($delayedExchangeSettings)
                 ->withQueueSettings($queueSettings)
         );
 
@@ -104,7 +111,7 @@ private function getExchangeSettings(?ExchangeSettingsInterface $exchangeSetting
         /** @noinspection NullPointerExceptionInspection */
         return $exchangeSettings
             ?->withName("{$exchangeSettings->getName()}.dlx")
-            ->withAutoDelete(true)
+            ->withAutoDelete(false)
             ->withType(AMQPExchangeType::TOPIC);
     }
 }

src/QueueProvider.php

@@ -31,6 +31,7 @@ public function __construct(
     public function __destruct()
     {
         $this->channel?->close();
+        //unset($this->channel);
     }
 
     public function getChannel(): AMQPChannel

src/Settings/Exchange.php

@@ -14,7 +14,7 @@ public function __construct(
         private string $type = AMQPExchangeType::DIRECT,
         private bool $passive = false,
         private bool $durable = false,
-        private bool $autoDelete = true,
+        private bool $autoDelete = false,
         private bool $internal = false,
         private bool $nowait = false,
         private AMQPTable|array $arguments = [],

src/Settings/Queue.php

@@ -14,7 +14,7 @@ public function __construct(
         private bool $passive = false,
         private bool $durable = false,
         private bool $exclusive = false,
-        private bool $autoDelete = true,
+        private bool $autoDelete = false,
         private bool $nowait = false,
         private AMQPTable|array $arguments = [],
         private ?int $ticket = null

tests/Integration/DelayMiddlewareTest.php

@@ -9,7 +9,7 @@
 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 +18,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 +45,7 @@ public function testMainFlow(): void
                 $this->createConnection(),
                 new QueueSettings(),
             ),
-            new MessageSerializer(),
+            new JsonMessageSerializer(),
             new SignalLoop(),
         );
         $queue = $this->makeQueue($adapter);
@@ -75,15 +76,15 @@ public function testMainFlowWithFakeAdapter(): void
                 $this->createConnection(),
                 new QueueSettings(),
             ),
-            new MessageSerializer(),
+            new JsonMessageSerializer(),
             new SignalLoop(),
         );
         $queue = $this->makeQueue($adapter);
 
         $this->expectException(InvalidArgumentException::class);
         $this->expectExceptionMessage("This middleware works only with the $adapterClass. $fakeAdapterClass given.");
         $queue->push(
-            new Message('simple', 'test-delay-middleware-main'),
+            new Message(SimpleMessageHandler::class, 'test-delay-middleware-main'),
             new DelayMiddleware(3),
         );
     }

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.
     }