Trunk enhancements branch

composer.json

@@ -28,7 +28,7 @@
         "php-amqplib/php-amqplib": "^3.1.0",
         "yiisoft/factory": "^1.0",
         "yiisoft/friendly-exception": "^1.0",
-        "yiisoft/queue": "dev-master"
+        "yiisoft/queue": "dev-new"
     },
     "require-dev": {
         "maglnet/composer-require-checker": "^4.4",

src/Adapter.php

@@ -10,7 +10,9 @@
 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
 {
@@ -23,64 +25,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 . '.');
+        throw new NotImplementedException(sprintf('Status check is not supported by the adapter %s.', 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/ExistingMessagesConsumer.php

@@ -8,6 +8,7 @@
 use PhpAmqpLib\Message\AMQPMessage;
 use Throwable;
 use Yiisoft\Queue\Message\MessageInterface;
+use Yiisoft\Queue\Message\MessageSerializerInterface;
 
 /**
  * @internal

src/Middleware/DelayMiddleware.php

@@ -7,18 +7,32 @@
 use InvalidArgumentException;
 use PhpAmqpLib\Exchange\AMQPExchangeType;
 use PhpAmqpLib\Message\AMQPMessage;
+use Yiisoft\Queue\Adapter\AdapterInterface;
 use Yiisoft\Queue\AMQP\Adapter;
 use Yiisoft\Queue\AMQP\QueueProviderInterface;
 use Yiisoft\Queue\AMQP\Settings\ExchangeSettingsInterface;
 use Yiisoft\Queue\AMQP\Settings\QueueSettingsInterface;
-use Yiisoft\Queue\Middleware\Push\Implementation\DelayMiddlewareInterface;
-use Yiisoft\Queue\Middleware\Push\MessageHandlerPushInterface;
-use Yiisoft\Queue\Middleware\Push\PushRequest;
+use Yiisoft\Queue\Middleware\DelayMiddlewareInterface;
+use Yiisoft\Queue\Middleware\MessageHandlerInterface;
+use Yiisoft\Queue\Middleware\MiddlewareInterface;
+use Yiisoft\Queue\Middleware\Request;
 
-final class DelayMiddleware implements DelayMiddlewareInterface
+final class DelayMiddleware implements MiddlewareInterface, DelayMiddlewareInterface
 {
-    public function __construct(private float $delayInSeconds, private bool $forcePersistentMessages = true)
-    {
+    public function __construct(
+        private AdapterInterface $adapter,
+        private float $delayInSeconds,
+        private bool $forcePersistentMessages = true
+    ) {
+        if (!$adapter instanceof Adapter) {
+            throw new InvalidArgumentException(
+                sprintf(
+                    'This middleware works only with the %s. %s given.',
+                    Adapter::class,
+                    get_debug_type($adapter)
+                )
+            );
+        }
     }
 
     /**
@@ -39,28 +53,28 @@ public function getDelay(): float
         return $this->delayInSeconds;
     }
 
-    public function processPush(PushRequest $request, MessageHandlerPushInterface $handler): PushRequest
+    public function process(Request $request, MessageHandlerInterface $handler): Request
     {
-        $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."
-            );
-        }
+        $queueProvider = $this->adapter->getQueueProvider();
+        $originalExchangeSettings = $queueProvider->getExchangeSettings();
+        $delayedExchangeSettings = $this->getExchangeSettings($originalExchangeSettings);
+        $queueSettings = $this->getQueueSettings(
+            $queueProvider->getQueueSettings(),
+            $originalExchangeSettings
+        );
 
-        $queueProvider = $adapter->getQueueProvider();
-        $exchangeSettings = $this->getExchangeSettings($queueProvider->getExchangeSettings());
-        $queueSettings = $this->getQueueSettings($queueProvider->getQueueSettings(), $queueProvider->getExchangeSettings());
-        $adapter = $adapter->withQueueProvider(
+        $adapter = $this->adapter->withQueueProvider(
             $queueProvider
                 ->withMessageProperties($this->getMessageProperties($queueProvider))
-                ->withExchangeSettings($exchangeSettings)
+                ->withExchangeSettings($delayedExchangeSettings)
                 ->withQueueSettings($queueSettings)
         );
 
-        return $handler->handlePush($request->withAdapter($adapter));
+        return $handler->handle(
+            $request->withQueue(
+                $request->getQueue()->withAdapter($adapter)
+            )
+        );
     }
 
     /**
@@ -104,7 +118,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