Omit custom handlers

src/Adapter.php

@@ -10,6 +10,7 @@
 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;
 
 final class Adapter implements AdapterInterface
@@ -23,64 +24,63 @@ public function __construct(
 
     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,

src/MessageSerializer.php

@@ -7,6 +7,7 @@
 use InvalidArgumentException;
 use JsonException;
 use Yiisoft\Queue\AMQP\Exception\NoKeyInPayloadException;
+use Yiisoft\Queue\Message\IdEnvelope;
 use Yiisoft\Queue\Message\Message;
 use Yiisoft\Queue\Message\MessageInterface;
 
@@ -18,7 +19,7 @@
     public function serialize(MessageInterface $message): string
     {
         $payload = [
-            'id' => $message->getId(),
+            'id' => $message->getMetadata()[IdEnvelope::MESSAGE_ID_KEY] ?? null,
             'name' => $message->getHandlerName(),
             'data' => $message->getData(),
             'meta' => $message->getMetadata(),
@@ -54,7 +55,7 @@
             throw new NoKeyInPayloadException('meta', $payload);
         }
 
         return new Message(
             $name,
             $payload['data'] ?? null,
             $meta,

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)
         );
 
@@ -78,20 +85,17 @@ private function getMessageProperties(QueueProviderInterface $queueProvider): ar
 
     private function getQueueSettings(
         QueueSettingsInterface $queueSettings,
-        ?ExchangeSettingsInterface $exchangeSettings
+        ?ExchangeSettingsInterface $originalExchangeSettings
     ): QueueSettingsInterface {
-        $deliveryTime = time() + $this->delayInSeconds;
-
+        $arguments = [
+            'x-dead-letter-exchange' => ['S', $originalExchangeSettings?->getName() ?? ''],
+            'x-dead-routing-key' => ['S', $queueSettings->getName()],
+            'x-expires' => ['I', $this->delayInSeconds * 1000 + 30_000],
+            'x-message-ttl' => ['I', $this->delayInSeconds * 1000],
+        ];
         return $queueSettings
-            ->withName("{$queueSettings->getName()}.dlx.$deliveryTime")
-            ->withAutoDeletable(true)
-            ->withArguments(
-                [
-                    'x-dead-letter-exchange' => ['S', $exchangeSettings?->getName() ?? ''],
-                    'x-expires' => ['I', $this->delayInSeconds * 1000 + 30000],
-                    'x-message-ttl' => ['I', $this->delayInSeconds * 1000],
-                ]
-            );
+            ->withName("{$queueSettings->getName()}.dlx")
+            ->withArguments($arguments);
     }
 
     /**
@@ -104,7 +108,6 @@ private function getExchangeSettings(?ExchangeSettingsInterface $exchangeSetting
         /** @noinspection NullPointerExceptionInspection */
         return $exchangeSettings
             ?->withName("{$exchangeSettings->getName()}.dlx")
-            ->withAutoDelete(true)
             ->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

@@ -51,7 +51,7 @@ public function testMainFlow(): void
 
         $time = time();
         $queue->push(
-            new Message('simple', 'test-delay-middleware-main'),
+            new Message(SimpleMessageHandler::class, 'test-delay-middleware-main'),
             new DelayMiddleware(3),
         );
 
@@ -96,9 +96,7 @@ private function makeQueue(AdapterInterface $adapter): Queue
             $this->createMock(LoggerInterface::class),
             new PushMiddlewareDispatcher(
                 new MiddlewareFactoryPush(
-                    new SimpleContainer([
-                        'simple' => new SimpleMessageHandler(new FileHelper()),
-                    ]),
+                    new SimpleContainer([]),
                     new CallableFactory($this->createMock(ContainerInterface::class)),
                 ),
             ),

tests/Integration/TestCase.php

@@ -40,13 +40,21 @@ protected function tearDown(): void
 
     protected function queueListen(?string $queue = null): void
     {
-        // TODO Fail test on subprocess error exit code
         $command = [PHP_BINARY, dirname(__DIR__) . '/yii', 'queue:listen'];
         if ($queue !== null) {
             $command[] = "--channel=$queue";
         }
         $process = new Process($command);
         $this->processes[] = $process;
         $process->start();
+
+        if ($process->isTerminated()) {
+            throw new \RuntimeException(
+                sprintf(
+                    "Failed to start queue:listen process: \n%s",
+                    !empty($process->getErrorOutput()) ? $process->getErrorOutput() : $process->getOutput()
+                )
+            );
+        }
     }
 }

tests/Support/ExceptionListener.php

@@ -0,0 +1,19 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Yiisoft\Queue\AMQP\Tests\Support;
+
+use PHPUnit\Util\Exception as PHPUnitException;
+use Yiisoft\Queue\Message\MessageInterface;
+
+final class ExceptionListener
+{
+    public function __invoke(MessageInterface $message): void
+    {
+        $data = $message->getData();
+        if (null !== $data) {
+            throw new PHPUnitException((string) $data['payload']['time']);
+        }
+    }
+}

tests/Support/ExtendedSimpleMessageHandler.php

@@ -15,7 +15,7 @@ public function __construct(private FileHelper $fileHelper)
     {
     }
 
-    public function handle(MessageInterface $message): void
+    public function __invoke(MessageInterface $message): void
     {
         $data = $message->getData();
         if (null !== $data) {

tests/Support/FakeAdapter.php

@@ -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/Support/FileHelper.php

@@ -14,8 +14,15 @@ final class FileHelper
      */
     public function put(string $fileName, int|string $data): void
     {
-        if (file_put_contents("{$this->getRuntimeDir()}/$fileName", $data) === false) {
-            throw new RuntimeException("Runtime dir {$this->getRuntimeDir()} or file $fileName are not writable.");
+        $path = $this->getRuntimeDir() . DIRECTORY_SEPARATOR . $fileName;
+        if (file_put_contents($path, $data) === false) {
+            throw new RuntimeException(
+                sprintf(
+                    'Runtime dir %"s" or file "%s" are not writable.',
+                    $this->getRuntimeDir(),
+                    $fileName
+                )
+            );
         }
     }
 
@@ -28,7 +35,12 @@ public function get(string $filename): ?string
 
         $result = file_get_contents($path);
         if ($result === false) {
-            throw new RuntimeException("File '$path' exists but is not readable.");
+            throw new RuntimeException(
+                sprintf(
+                    'File "%s" exists but is not readable.',
+                    $path
+                )
+            );
         }
 
         return $result;

tests/Unit/MessageSerializerTest.php

@@ -14,6 +14,7 @@
 use Yiisoft\Queue\AMQP\QueueProvider;
 use Yiisoft\Queue\AMQP\Settings\Exchange as ExchangeSettings;
 use Yiisoft\Queue\AMQP\Settings\Queue as QueueSettings;
+use Yiisoft\Queue\AMQP\Tests\Support\ExtendedSimpleMessageHandler;
 
 /**
  * Testing message serialization options
@@ -27,11 +28,10 @@ final class MessageSerializerTest extends UnitTestCase
      */
     private function publishWithAMQPLib(string $queue, string $exchange, AMQPMessage $message): void
     {
-        $channel = $this
-            ->createConnection()
-            ->channel();
+        $channel = $this->createConnection()->channel();
+
         $channel->queue_declare($queue);
-        $channel->exchange_declare($exchange, AMQPExchangeType::DIRECT);
+        $channel->exchange_declare($exchange, AMQPExchangeType::DIRECT, auto_delete: true);
         $channel->queue_bind($queue, $exchange);
         $channel->basic_publish($message, $exchange);
     }
@@ -47,8 +47,8 @@ private function getCustomAdapter(string $queueExchangeName): Adapter
         );
         return new Adapter(
             $queueProvider
-                ->withQueueSettings(new QueueSettings($queueExchangeName))
-                ->withExchangeSettings(new ExchangeSettings($queueExchangeName)),
+                ->withQueueSettings(new QueueSettings($queueExchangeName, autoDelete: true))
+                ->withExchangeSettings(new ExchangeSettings($queueExchangeName, autoDelete: true)),
             new MessageSerializer(),
             $this->getLoop(),
         );
@@ -80,7 +80,7 @@ public function testNoKeyInPayloadExceptionId(): void
             $queueExchangeName,
             $queueExchangeName,
             new AMQPMessage(
-                json_encode(['name' => 'ext-simple', 'id' => 1], JSON_THROW_ON_ERROR),
+                json_encode(['name' => ExtendedSimpleMessageHandler::class, 'id' => 1], JSON_THROW_ON_ERROR),
                 ['content_type' => 'text/json', 'delivery_mode' => AMQPMessage::DELIVERY_MODE_PERSISTENT]
             )
         );
@@ -100,7 +100,7 @@ public function testNoKeyInPayloadExceptionMeta(): void
             $queueExchangeName,
             $queueExchangeName,
             new AMQPMessage(
-                json_encode(['name' => 'ext-simple', 'meta' => ''], JSON_THROW_ON_ERROR),
+                json_encode(['name' => ExtendedSimpleMessageHandler::class, 'meta' => ''], JSON_THROW_ON_ERROR),
                 ['content_type' => 'text/json', 'delivery_mode' => AMQPMessage::DELIVERY_MODE_PERSISTENT]
             )
         );