Trunk enhancements branch

.gitattributes

@@ -34,7 +34,10 @@
 /psalm.xml              export-ignore
 /phpunit.xml.dist       export-ignore
 /Makefile               export-ignore
-/tests                  export-ignore
+/tests/App              export-ignore
+/tests/Integration      export-ignore
+/tests/Unit             export-ignore
+/tests/runtime          export-ignore
 /docs                   export-ignore
 
 # Avoid merge conflicts in CHANGELOG

.gitignore

@@ -19,4 +19,5 @@ composer.lock
 
 # PHPUnit
 .phpunit.result.cache
+.phpunit.cache
 coverage.html

README.md

@@ -68,7 +68,7 @@ $data = [
     'url' => $url,
     'destinationFile' => $filename,
 ];
-$message = new \Yiisoft\Queue\Message\Message('file-download', $data);
+$message = new \Yiisoft\Queue\Message\Message(FileDownloader::class, $data);
 ```
 
 Then you should push it to the queue:
@@ -80,7 +80,10 @@ $queue->push($message);
 Its handler may look like the following:
 
 ```php
-class FileDownloader
+use Yiisoft\Queue\Message\MessageInterface;
+use Yiisoft\Queue\Message\MessageHandlerInterface;
+
+class FileDownloader implements MessageHandlerInterface
 {
     private string $absolutePath;
 
@@ -89,21 +92,21 @@ class FileDownloader
         $this->absolutePath = $absolutePath;
     }
 
-    public function handle(\Yiisoft\Queue\Message\MessageInterface $downloadMessage): void
+    public function handle(\Yiisoft\Queue\Message\MessageInterface $downloadMessage): MessageInterface
     {
-        $fileName = $downloadMessage->getData()['destinationFile'];
+        $fileName = $message->getData()['destinationFile'];
         $path = "$this->absolutePath/$fileName"; 
-        file_put_contents($path, file_get_contents($downloadMessage->getData()['url']));
+        file_put_contents($path, file_get_contents($message->getData()['url']));
+
+        return $message;
     }
 }
 ```
 
 The last thing we should do is to create a configuration for the `Yiisoft\Queue\Worker\Worker`:
 
 ```php
-$handlers = ['file-download' => [new FileDownloader('/path/to/save/files'), 'handle']];
 $worker = new \Yiisoft\Queue\Worker\Worker(
-    $handlers, // Here it is
     $logger,
     $injector,
     $container

composer.json

@@ -28,11 +28,15 @@
     "require": {
         "php": "^8.1",
         "psr/container": "^1.0|^2.0",
+        "psr/event-dispatcher": "^1.0",
         "psr/log": "^2.0|^3.0",
         "symfony/console": "^5.4|^6.0",
         "yiisoft/definitions": "^1.0|^2.0|^3.0",
+        "yiisoft/event-dispatcher": "^1.1",
         "yiisoft/friendly-exception": "^1.0",
-        "yiisoft/injector": "^1.0"
+        "yiisoft/injector": "^1.0",
+        "yiisoft/var-dumper": "^1.7",
+        "yiisoft/yii-event": "^2.1"
     },
     "require-dev": {
         "maglnet/composer-require-checker": "^4.7",
@@ -49,7 +53,8 @@
     },
     "autoload": {
         "psr-4": {
-            "Yiisoft\\Queue\\": "src"
+            "Yiisoft\\Queue\\": "src",
+            "Yiisoft\\Queue\\Tests\\Shared\\": "tests/Shared"
         }
     },
     "autoload-dev": {
@@ -66,6 +71,8 @@
         },
         "config-plugin": {
             "di": "di.php",
+            "events-web": "events-web.php",
+            "events-console": "events-console.php",
             "params": "params.php"
         }
     },

config/di.php

@@ -3,60 +3,77 @@
 declare(strict_types=1);
 
 use Psr\Container\ContainerInterface;
+use Yiisoft\Config\ConfigInterface;
+use Yiisoft\Definitions\Reference;
+use Yiisoft\EventDispatcher\Dispatcher\Dispatcher;
+use Yiisoft\EventDispatcher\Provider\Provider;
+use Yiisoft\Injector\Injector;
+use Yiisoft\Queue\Adapter\AdapterInterface;
+use Yiisoft\Queue\Adapter\SynchronousAdapter;
 use Yiisoft\Queue\Cli\LoopInterface;
 use Yiisoft\Queue\Cli\SignalLoop;
 use Yiisoft\Queue\Cli\SimpleLoop;
-use Yiisoft\Queue\Command\ListenAllCommand;
-use Yiisoft\Queue\Command\RunCommand;
 use Yiisoft\Queue\Message\JsonMessageSerializer;
 use Yiisoft\Queue\Message\MessageSerializerInterface;
-use Yiisoft\Queue\Middleware\Consume\ConsumeMiddlewareDispatcher;
-use Yiisoft\Queue\Middleware\Consume\MiddlewareFactoryConsume;
-use Yiisoft\Queue\Middleware\Consume\MiddlewareFactoryConsumeInterface;
-use Yiisoft\Queue\Middleware\FailureHandling\FailureMiddlewareDispatcher;
-use Yiisoft\Queue\Middleware\FailureHandling\MiddlewareFactoryFailure;
-use Yiisoft\Queue\Middleware\FailureHandling\MiddlewareFactoryFailureInterface;
-use Yiisoft\Queue\Middleware\Push\MiddlewareFactoryPush;
-use Yiisoft\Queue\Middleware\Push\MiddlewareFactoryPushInterface;
-use Yiisoft\Queue\Middleware\Push\PushMiddlewareDispatcher;
+use Yiisoft\Queue\Middleware\MiddlewareDispatcher;
+use Yiisoft\Queue\Middleware\MiddlewareFactory;
+use Yiisoft\Queue\Middleware\MiddlewareFactoryInterface;
+use Yiisoft\Queue\Command\ListenAllCommand;
+use Yiisoft\Queue\Command\RunCommand;
 use Yiisoft\Queue\Queue;
 use Yiisoft\Queue\QueueFactory;
 use Yiisoft\Queue\QueueFactoryInterface;
 use Yiisoft\Queue\QueueInterface;
 use Yiisoft\Queue\Worker\Worker as QueueWorker;
 use Yiisoft\Queue\Worker\WorkerInterface;
+use Yiisoft\Yii\Event\ListenerCollectionFactory;
 
 /* @var array $params */
 
 return [
     QueueWorker::class => [
         'class' => QueueWorker::class,
-        '__construct()' => [$params['yiisoft/queue']['handlers']],
+        '__construct()' => [
+            'eventDispatcher' => Reference::to('queue.dispatcher'),
+        ],
     ],
     WorkerInterface::class => QueueWorker::class,
     LoopInterface::class => static function (ContainerInterface $container): LoopInterface {
-        return extension_loaded('pcntl')
-            ? $container->get(SignalLoop::class)
-            : $container->get(SimpleLoop::class);
+        return $container->get(
+            extension_loaded('pcntl')
+                ? SignalLoop::class
+                : SimpleLoop::class
+        );
+    },
+    'queue.middlewareDispatcher.push' => static function (Injector $injector) use ($params) {
+        return $injector->make(
+            MiddlewareDispatcher::class,
+            ['middlewareDefinitions' => $params['yiisoft/queue']['middlewares-push']]
+        );
     },
+    Queue::class => [
+        '__construct()' => [
+            'adapter' => Reference::to(AdapterInterface::class),
+            'pushMiddlewareDispatcher' => Reference::to('queue.middlewareDispatcher.push'),
+        ],
+    ],
     QueueFactoryInterface::class => QueueFactory::class,
     QueueFactory::class => [
         '__construct()' => ['channelConfiguration' => $params['yiisoft/queue']['channel-definitions']],
     ],
+    AdapterInterface::class => SynchronousAdapter::class,
+
     QueueInterface::class => Queue::class,
-    MiddlewareFactoryPushInterface::class => MiddlewareFactoryPush::class,
-    MiddlewareFactoryConsumeInterface::class => MiddlewareFactoryConsume::class,
-    MiddlewareFactoryFailureInterface::class => MiddlewareFactoryFailure::class,
-    PushMiddlewareDispatcher::class => [
-        '__construct()' => ['middlewareDefinitions' => $params['yiisoft/queue']['middlewares-push']],
-    ],
-    ConsumeMiddlewareDispatcher::class => [
-        '__construct()' => ['middlewareDefinitions' => $params['yiisoft/queue']['middlewares-consume']],
-    ],
-    FailureMiddlewareDispatcher::class => [
-        '__construct()' => ['middlewareDefinitions' => $params['yiisoft/queue']['middlewares-fail']],
-    ],
     MessageSerializerInterface::class => JsonMessageSerializer::class,
+    MiddlewareFactoryInterface::class => MiddlewareFactory::class,
+
+    'queue.dispatcher' => static function (ConfigInterface $config, ListenerCollectionFactory $factory) {
+        $listeners = $factory->create($config->get('queue'));
+
+        $provider = new Provider($listeners);
+
+        return new Dispatcher($provider);
+    },
     RunCommand::class => [
         '__construct()' => [
             'channels' => array_keys($params['yiisoft/queue']['channel-definitions']),

config/events-console.php

@@ -0,0 +1,20 @@
+<?php
+
+declare(strict_types=1);
+
+use Psr\Container\ContainerInterface;
+use Yiisoft\Queue\Adapter\AdapterInterface;
+use Yiisoft\Queue\Adapter\SynchronousAdapter;
+use Yiisoft\Queue\QueueInterface;
+
+/* @var array $params */
+
+return [
+    \Yiisoft\Yii\Console\Event\ApplicationShutdown::class => [
+        function (AdapterInterface $adapter, ContainerInterface $container) {
+            if ($adapter instanceof SynchronousAdapter) {
+                $container->get(QueueInterface::class)->run(0);
+            }
+        },
+    ],
+];

config/events-web.php

@@ -0,0 +1,20 @@
+<?php
+
+declare(strict_types=1);
+
+use Psr\Container\ContainerInterface;
+use Yiisoft\Queue\Adapter\AdapterInterface;
+use Yiisoft\Queue\Adapter\SynchronousAdapter;
+use Yiisoft\Queue\QueueInterface;
+
+/* @var array $params */
+
+return [
+    \Yiisoft\Yii\Http\Event\AfterRequest::class => [
+        function (AdapterInterface $adapter, ContainerInterface $container) {
+            if ($adapter instanceof SynchronousAdapter) {
+                $container->get(QueueInterface::class)->run(0);
+            }
+        },
+    ],
+];

config/params.php

@@ -21,7 +21,6 @@
         ],
     ],
     'yiisoft/queue' => [
-        'handlers' => [],
         'channel-definitions' => [
             QueueFactoryInterface::DEFAULT_CHANNEL_NAME => AdapterInterface::class,
         ],

config/queue.php

@@ -0,0 +1,7 @@
+<?php
+
+declare(strict_types=1);
+
+/* @var array $params */
+
+return [];

src/Adapter/SynchronousAdapter.php

@@ -7,9 +7,8 @@
 use InvalidArgumentException;
 use Yiisoft\Queue\Enum\JobStatus;
 use Yiisoft\Queue\Message\MessageInterface;
-use Yiisoft\Queue\QueueFactory;
-use Yiisoft\Queue\QueueInterface;
-use Yiisoft\Queue\Worker\WorkerInterface;
+use Yiisoft\Queue\Message\MessageSerializerInterface;
+use Yiisoft\Queue\QueueFactoryInterface;
 use Yiisoft\Queue\Message\IdEnvelope;
 
 final class SynchronousAdapter implements AdapterInterface
@@ -18,26 +17,18 @@ final class SynchronousAdapter implements AdapterInterface
     private int $current = 0;
 
     public function __construct(
-        private WorkerInterface $worker,
-        private QueueInterface $queue,
-        private string $channel = QueueFactory::DEFAULT_CHANNEL_NAME,
+        private MessageSerializerInterface $messageSerializer,
+        private string $channel = QueueFactoryInterface::DEFAULT_CHANNEL_NAME,
     ) {
     }
 
-    public function __destruct()
-    {
-        $this->runExisting(function (MessageInterface $message): bool {
-            $this->worker->process($message, $this->queue);
-
-            return true;
-        });
-    }
-
     public function runExisting(callable $handlerCallback): void
     {
         $result = true;
-        while (isset($this->messages[$this->current]) && $result === true) {
-            $result = $handlerCallback($this->messages[$this->current]);
+        while ($result === true && isset($this->messages[$this->current])) {
+            $result = $handlerCallback(
+                $this->messageSerializer->unserialize($this->messages[$this->current])
+            );
             unset($this->messages[$this->current]);
             $this->current++;
         }
@@ -65,9 +56,10 @@ public function status(string|int $id): JobStatus
     public function push(MessageInterface $message): MessageInterface
     {
         $key = count($this->messages) + $this->current;
-        $this->messages[] = $message;
+        $newMessage = new IdEnvelope($message, $key);
+        $this->messages[] = $this->messageSerializer->serialize($newMessage);
 
-        return new IdEnvelope($message, $key);
+        return $newMessage;
     }
 
     public function subscribe(callable $handlerCallback): void

src/Debug/QueueCollector.php

@@ -4,11 +4,11 @@
 
 namespace Yiisoft\Queue\Debug;
 
+use Yiisoft\Queue\Middleware\MiddlewareInterface;
 use Yiisoft\Yii\Debug\Collector\CollectorTrait;
 use Yiisoft\Yii\Debug\Collector\SummaryCollectorInterface;
 use Yiisoft\Queue\Enum\JobStatus;
 use Yiisoft\Queue\Message\MessageInterface;
-use Yiisoft\Queue\Middleware\Push\MiddlewarePushInterface;
 use Yiisoft\Queue\QueueInterface;
 
 final class QueueCollector implements SummaryCollectorInterface
@@ -53,7 +53,7 @@ public function collectStatus(string $id, JobStatus $status): void
     public function collectPush(
         string $channel,
         MessageInterface $message,
-        string|array|callable|MiddlewarePushInterface ...$middlewareDefinitions,
+        string|array|callable|MiddlewareInterface ...$middlewareDefinitions,
     ): void {
         if (!$this->isActive()) {
             return;

src/Debug/QueueDecorator.php

@@ -7,7 +7,7 @@
 use Yiisoft\Queue\Adapter\AdapterInterface;
 use Yiisoft\Queue\Enum\JobStatus;
 use Yiisoft\Queue\Message\MessageInterface;
-use Yiisoft\Queue\Middleware\Push\MiddlewarePushInterface;
+use Yiisoft\Queue\Middleware\MiddlewareInterface;
 use Yiisoft\Queue\QueueInterface;
 
 final class QueueDecorator implements QueueInterface
@@ -28,7 +28,7 @@
 
     public function push(
         MessageInterface $message,
-        string|array|callable|MiddlewarePushInterface ...$middlewareDefinitions
+        string|array|callable|MiddlewareInterface ...$middlewareDefinitions
     ): MessageInterface {
         $message = $this->queue->push($message, ...$middlewareDefinitions);
         $this->collector->collectPush($this->queue->getChannelName(), $message, ...$middlewareDefinitions);
@@ -61,4 +61,9 @@
         $new->queue = $this->queue->withChannelName($channel);
         return $new;
     }
+
+    public function getAdapter(): ?AdapterInterface
+    {
+        return $this->queue->getAdapter();
+    }
 }