Add queue collector

composer.json

@@ -36,7 +36,8 @@
         "yiisoft/view": "^4.0|^5.0|^6.0",
         "yiisoft/yii-console": "^1.0",
         "yiisoft/yii-filesystem": "dev-master",
-        "yiisoft/yii-http": "^1.0"
+        "yiisoft/yii-http": "^1.0",
+        "yiisoft/yii-queue": "3.0.x-dev"
     },
     "require-dev": {
         "nyholm/psr7": "^1.3",

config/params.php

@@ -14,6 +14,9 @@
 use Yiisoft\Yii\Debug\Collector\EventCollector;
 use Yiisoft\Yii\Debug\Collector\LogCollector;
 use Yiisoft\Yii\Debug\Collector\MiddlewareCollector;
+use Yiisoft\Yii\Debug\Collector\QueueCollector;
+use Yiisoft\Yii\Debug\Collector\QueueFactoryInterfaceProxy;
+use Yiisoft\Yii\Debug\Collector\QueueWorkerInterfaceProxy;
 use Yiisoft\Yii\Debug\Collector\RequestCollector;
 use Yiisoft\Yii\Debug\Collector\RouterCollector;
 use Yiisoft\Yii\Debug\Collector\ServiceCollector;
@@ -27,6 +30,8 @@
 use Yiisoft\Yii\Debug\Collector\LoggerInterfaceProxy;
 use Yiisoft\Yii\Debug\Collector\UrlMatcherInterfaceProxy;
 use Yiisoft\Yii\Debug\Collector\ValidatorInterfaceProxy;
+use Yiisoft\Yii\Queue\QueueFactoryInterface;
+use Yiisoft\Yii\Queue\Worker\WorkerInterface;
 
 /**
  * @var $params array
@@ -40,6 +45,7 @@
             EventCollector::class,
             ServiceCollector::class,
             ValidatorCollector::class,
+            QueueCollector::class,
         ],
         'collectors.web' => [
             WebAppInfoCollector::class,
@@ -56,6 +62,8 @@
         'trackedServices' => [
             LoggerInterface::class => [LoggerInterfaceProxy::class, LogCollector::class],
             EventDispatcherInterface::class => [EventDispatcherInterfaceProxy::class, EventCollector::class],
+            QueueFactoryInterface::class => [QueueFactoryInterfaceProxy::class, QueueCollector::class],
+            WorkerInterface::class => [QueueWorkerInterfaceProxy::class, QueueCollector::class],
             UrlMatcherInterface::class => [UrlMatcherInterfaceProxy::class, RouterCollector::class],
             ValidatorInterface::class => [ValidatorInterfaceProxy::class, ValidatorCollector::class],
             AssetLoaderInterface::class => [AssetLoaderInterfaceProxy::class, AssetCollector::class],

src/Collector/QueueCollector.php

@@ -0,0 +1,73 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Yiisoft\Yii\Debug\Collector;
+
+use Yiisoft\Yii\Queue\Message\MessageInterface;
+use Yiisoft\Yii\Queue\QueueInterface;
+
+final class QueueCollector implements CollectorInterface, IndexCollectorInterface
+{
+    use CollectorTrait;
+
+    private array $pushes = [];
+    private array $statuses = [];
+    private array $processingMessages = [];
+
+    public function getCollected(): array
+    {
+        return [
+            'pushes' => $this->pushes,
+            'statuses' => $this->statuses,
+            'processingMessages' => $this->processingMessages,
+        ];
+    }
+
+    public function collectStatus(string $id): void
+    {
+        if (!$this->isActive()) {
+            return;
+        }
+
+        $this->statuses[] = $id;
+    }
+
+    public function collectPush(string $channel, MessageInterface $message): void
+    {
+        if (!$this->isActive()) {
+            return;
+        }
+        $this->pushes[$channel][] = $message;
+    }
+
+    public function collectWorkerProcessing(MessageInterface $message, QueueInterface $queue)
+    {
+        if (!$this->isActive()) {
+            return;
+        }
+        $this->processingMessages[$queue->getChannelName()][] = $message;
+    }
+
+    private function reset(): void
+    {
+        $this->pushes = [];
+        $this->statuses = [];
+        $this->processingMessages = [];
+    }
+
+    public function getIndexData(): array
+    {
+        $countPushes = array_sum(array_map(fn ($messages) => count($messages), $this->pushes));
+        $countStatuses = count($this->statuses);
+        $countProcessingMessages = array_sum(array_map(fn ($messages) => count($messages), $this->processingMessages));
+
+        return [
+            'queue' => [
+                'countPushes' => $countPushes,
+                'countStatuses' => $countStatuses,
+                'countProcessingMessages' => $countProcessingMessages,
+            ],
+        ];
+    }
+}

src/Collector/QueueDecorator.php

@@ -0,0 +1,53 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Yiisoft\Yii\Debug\Collector;
+
+use Yiisoft\Yii\Queue\Adapter\AdapterInterface;
+use Yiisoft\Yii\Queue\Enum\JobStatus;
+use Yiisoft\Yii\Queue\Message\MessageInterface;
+use Yiisoft\Yii\Queue\QueueInterface;
+
+final class QueueDecorator implements QueueInterface
+{
+    public function __construct(
+        private QueueInterface $queue,
+        private QueueCollector $collector,
+    ) {
+    }
+
+    public function status(string $id): JobStatus
+    {
+        $result = $this->queue->status($id);
+        $this->collector->collectStatus($id);
+
+        return $result;
+    }
+
+    public function push(MessageInterface $message): void
+    {
+        $this->queue->push($message);
+        $this->collector->collectPush($this->queue->getChannelName(), $message);
+    }
+
+    public function run(int $max = 0): void
+    {
+        $this->queue->run($max);
+    }
+
+    public function listen(): void
+    {
+        $this->queue->listen();
+    }
+
+    public function withAdapter(AdapterInterface $adapter): QueueInterface
+    {
+        return new self($this->queue->withAdapter($adapter), $this->collector);
+    }
+
+    public function getChannelName(): string
+    {
+        return $this->queue->getChannelName();
+    }
+}

src/Collector/QueueFactoryInterfaceProxy.php

@@ -0,0 +1,24 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Yiisoft\Yii\Debug\Collector;
+
+use Yiisoft\Yii\Queue\QueueFactoryInterface;
+use Yiisoft\Yii\Queue\QueueInterface;
+
+final class QueueFactoryInterfaceProxy implements QueueFactoryInterface
+{
+    public function __construct(
+        private QueueFactoryInterface $queueFactory,
+        private QueueCollector $collector,
+    ) {
+    }
+
+    public function get(string $channel = self::DEFAULT_CHANNEL_NAME): QueueInterface
+    {
+        $queue = $this->queueFactory->get($channel);
+
+        return new QueueDecorator($queue, $this->collector);
+    }
+}

src/Collector/QueueWorkerInterfaceProxy.php

@@ -0,0 +1,24 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Yiisoft\Yii\Debug\Collector;
+
+use Yiisoft\Yii\Queue\Message\MessageInterface;
+use Yiisoft\Yii\Queue\QueueInterface;
+use Yiisoft\Yii\Queue\Worker\WorkerInterface;
+
+final class QueueWorkerInterfaceProxy implements WorkerInterface
+{
+    public function __construct(
+        private WorkerInterface $worker,
+        private QueueCollector $collector,
+    ) {
+    }
+
+    public function process(MessageInterface $message, QueueInterface $queue): void
+    {
+        $this->collector->collectWorkerProcessing($message, $queue);
+        $this->worker->process($message, $queue);
+    }
+}

tests/Collector/QueueCollectorTest.php

@@ -0,0 +1,89 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Yiisoft\Yii\Debug\Tests\Collector;
+
+use Yiisoft\Validator\Result;
+use Yiisoft\Validator\Rule\Number;
+use Yiisoft\Yii\Debug\Collector\CollectorInterface;
+use Yiisoft\Yii\Debug\Collector\IndexCollectorInterface;
+use Yiisoft\Yii\Debug\Collector\QueueCollector;
+use Yiisoft\Yii\Debug\Tests\Support\DummyQueue;
+use Yiisoft\Yii\Queue\Message\Message;
+
+final class QueueCollectorTest extends CollectorTestCase
+{
+    private Message $pushMessage;
+
+    protected function setUp(): void
+    {
+        parent::setUp();
+        $this->pushMessage = new Message('task', ['id' => 500]);
+    }
+
+    /**
+     * @param CollectorInterface|QueueCollector $collector
+     */
+    protected function collectTestData(CollectorInterface $collector): void
+    {
+        $ruleNumber = new Number(min: 200);
+        $result = new Result();
+        $result->addError($ruleNumber->getTooSmallMessage());
+
+        $collector->collectStatus('12345');
+        $collector->collectPush('chan1', $this->pushMessage);
+        $collector->collectPush('chan2', $this->pushMessage);
+        $collector->collectWorkerProcessing(
+            $this->pushMessage,
+            new DummyQueue('chan1'),
+        );
+        $collector->collectWorkerProcessing(
+            $this->pushMessage,
+            new DummyQueue('chan1'),
+        );
+        $collector->collectWorkerProcessing(
+            $this->pushMessage,
+            new DummyQueue('chan2'),
+        );
+    }
+
+    protected function getCollector(): CollectorInterface
+    {
+        return new QueueCollector();
+    }
+
+    protected function checkCollectedData(CollectorInterface $collector): void
+    {
+        parent::checkCollectedData($collector);
+        [
+            'pushes' => $pushes,
+            'statuses' => $statuses,
+            'processingMessages' => $processingMessages,
+        ] = $collector->getCollected();
+
+        $this->assertEquals(['chan1' => [$this->pushMessage], 'chan2' => [$this->pushMessage]], $pushes);
+        $this->assertEquals(['12345'], $statuses);
+        $this->assertEquals(
+            [
+                'chan1' => [$this->pushMessage, $this->pushMessage],
+                'chan2' => [$this->pushMessage],
+            ],
+            $processingMessages
+        );
+    }
+
+    protected function checkIndexData(CollectorInterface|IndexCollectorInterface $collector): void
+    {
+        parent::checkIndexData($collector);
+        [
+            'countPushes' => $countPushes,
+            'countStatuses' => $countStatuses,
+            'countProcessingMessages' => $countProcessingMessages,
+        ] = $collector->getIndexData()['queue'];
+
+        $this->assertEquals(2, $countPushes);
+        $this->assertEquals(1, $countStatuses);
+        $this->assertEquals(3, $countProcessingMessages);
+    }
+}

tests/Collector/QueueDecoratorTest.php

@@ -0,0 +1,30 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Yiisoft\Yii\Debug\Tests\Collector;
+
+use PHPUnit\Framework\TestCase;
+use Yiisoft\Yii\Debug\Collector\QueueCollector;
+use Yiisoft\Yii\Debug\Collector\QueueDecorator;
+use Yiisoft\Yii\Queue\Adapter\AdapterInterface;
+use Yiisoft\Yii\Queue\QueueInterface;
+
+class QueueDecoratorTest extends TestCase
+{
+    public function testWithAdapter()
+    {
+        $queue = $this->createMock(QueueInterface::class);
+        $collector = new QueueCollector();
+        $decorator = new QueueDecorator(
+            $queue,
+            $collector,
+        );
+
+        $queueAdapter = $this->createMock(AdapterInterface::class);
+
+        $newDecorator = $decorator->withAdapter($queueAdapter);
+
+        $this->assertInstanceOf(QueueDecorator::class, $newDecorator);
+    }
+}

tests/Support/DummyQueue.php

@@ -0,0 +1,45 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Yiisoft\Yii\Debug\Tests\Support;
+
+use Exception;
+use Yiisoft\Yii\Queue\Adapter\AdapterInterface;
+use Yiisoft\Yii\Queue\Enum\JobStatus;
+use Yiisoft\Yii\Queue\Message\MessageInterface;
+use Yiisoft\Yii\Queue\QueueInterface;
+
+final class DummyQueue implements QueueInterface
+{
+    public function __construct(private $channelName)
+    {
+    }
+
+    public function push(MessageInterface $message): void
+    {
+    }
+
+    public function run(int $max = 0): void
+    {
+    }
+
+    public function listen(): void
+    {
+    }
+
+    public function status(string $id): JobStatus
+    {
+        throw new Exception('`status()` method is not implemented yet.');
+    }
+
+    public function withAdapter(AdapterInterface $adapter): QueueInterface
+    {
+        throw new Exception('`withAdapter()` method is not implemented yet.');
+    }
+
+    public function getChannelName(): string
+    {
+        return $this->channelName;
+    }
+}