Replace queue factory with queue providers

composer.json

@@ -41,6 +41,7 @@
         "psr/log": "^2.0|^3.0",
         "symfony/console": "^5.4|^6.0",
         "yiisoft/definitions": "^1.0|^2.0|^3.0",
+        "yiisoft/factory": "dev-strict",
         "yiisoft/friendly-exception": "^1.0",
         "yiisoft/injector": "^1.0"
     },

config/di.php

@@ -19,16 +19,22 @@
 use Yiisoft\Queue\Middleware\Push\MiddlewareFactoryPush;
 use Yiisoft\Queue\Middleware\Push\MiddlewareFactoryPushInterface;
 use Yiisoft\Queue\Middleware\Push\PushMiddlewareDispatcher;
+use Yiisoft\Queue\Provider\QueueFactoryQueueProvider;
+use Yiisoft\Queue\Provider\QueueProviderInterface;
 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;
 
 /* @var array $params */
 
 return [
+    QueueProviderInterface::class => [
+        'class' => QueueFactoryQueueProvider::class,
+        '__construct()' => [
+            'definitions' => $params['yiisoft/queue']['channel-definitions'],
+        ],
+    ],
     QueueWorker::class => [
         'class' => QueueWorker::class,
         '__construct()' => [$params['yiisoft/queue']['handlers']],
@@ -39,10 +45,6 @@
             ? $container->get(SignalLoop::class)
             : $container->get(SimpleLoop::class);
     },
-    QueueFactoryInterface::class => QueueFactory::class,
-    QueueFactory::class => [
-        '__construct()' => ['channelConfiguration' => $params['yiisoft/queue']['channel-definitions']],
-    ],
     QueueInterface::class => Queue::class,
     MiddlewareFactoryPushInterface::class => MiddlewareFactoryPush::class,
     MiddlewareFactoryConsumeInterface::class => MiddlewareFactoryConsume::class,

config/params.php

@@ -2,14 +2,13 @@
 
 declare(strict_types=1);
 
-use Yiisoft\Queue\Adapter\AdapterInterface;
 use Yiisoft\Queue\Command\ListenAllCommand;
 use Yiisoft\Queue\Command\ListenCommand;
 use Yiisoft\Queue\Command\RunCommand;
 use Yiisoft\Queue\Debug\QueueCollector;
-use Yiisoft\Queue\Debug\QueueFactoryInterfaceProxy;
+use Yiisoft\Queue\Debug\QueueProviderInterfaceProxy;
 use Yiisoft\Queue\Debug\QueueWorkerInterfaceProxy;
-use Yiisoft\Queue\QueueFactoryInterface;
+use Yiisoft\Queue\Provider\QueueProviderInterface;
 use Yiisoft\Queue\Worker\WorkerInterface;
 
 return [
@@ -22,9 +21,7 @@
     ],
     'yiisoft/queue' => [
         'handlers' => [],
-        'channel-definitions' => [
-            QueueFactoryInterface::DEFAULT_CHANNEL_NAME => AdapterInterface::class,
-        ],
+        'channel-definitions' => [],
         'middlewares-push' => [],
         'middlewares-consume' => [],
         'middlewares-fail' => [],
@@ -34,7 +31,7 @@
             QueueCollector::class,
         ],
         'trackedServices' => [
-            QueueFactoryInterface::class => [QueueFactoryInterfaceProxy::class, QueueCollector::class],
+            QueueProviderInterface::class => [QueueProviderInterfaceProxy::class, QueueCollector::class],
             WorkerInterface::class => [QueueWorkerInterfaceProxy::class, QueueCollector::class],
         ],
     ],

src/Adapter/StubAdapter.php

@@ -0,0 +1,34 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Yiisoft\Queue\Adapter;
+
+use Yiisoft\Queue\Enum\JobStatus;
+use Yiisoft\Queue\Message\MessageInterface;
+
+final class StubAdapter implements AdapterInterface
+{
+    public function runExisting(callable $handlerCallback): void
+    {
+    }
+
+    public function status(int|string $id): JobStatus
+    {
+        return JobStatus::done();
+    }
+
+    public function push(MessageInterface $message): MessageInterface
+    {
+        return $message;
+    }
+
+    public function subscribe(callable $handlerCallback): void
+    {
+    }
+
+    public function withChannel(string $channel): AdapterInterface
+    {
+        return clone $this;
+    }
+}

src/Adapter/SynchronousAdapter.php

@@ -7,7 +7,7 @@
 use InvalidArgumentException;
 use Yiisoft\Queue\Enum\JobStatus;
 use Yiisoft\Queue\Message\MessageInterface;
-use Yiisoft\Queue\QueueFactory;
+use Yiisoft\Queue\Queue;
 use Yiisoft\Queue\QueueInterface;
 use Yiisoft\Queue\Worker\WorkerInterface;
 use Yiisoft\Queue\Message\IdEnvelope;
@@ -20,7 +20,7 @@ final class SynchronousAdapter implements AdapterInterface
     public function __construct(
         private WorkerInterface $worker,
         private QueueInterface $queue,
-        private string $channel = QueueFactory::DEFAULT_CHANNEL_NAME,
+        private string $channel = Queue::DEFAULT_CHANNEL_NAME,
     ) {
     }
 

src/Cli/StubLoop.php

@@ -0,0 +1,18 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Yiisoft\Queue\Cli;
+
+final class StubLoop implements LoopInterface
+{
+    public function __construct(
+        private readonly bool $canContinue = true,
+    ) {
+    }
+
+    public function canContinue(): bool
+    {
+        return $this->canContinue;
+    }
+}

src/Command/ListenAllCommand.php

@@ -10,7 +10,7 @@
 use Symfony\Component\Console\Input\InputOption;
 use Symfony\Component\Console\Output\OutputInterface;
 use Yiisoft\Queue\Cli\LoopInterface;
-use Yiisoft\Queue\QueueFactoryInterface;
+use Yiisoft\Queue\Provider\QueueProviderInterface;
 
 final class ListenAllCommand extends Command
 {
@@ -20,8 +20,11 @@ final class ListenAllCommand extends Command
     'Listens all configured queues by default in case you\'re using yiisoft/config. ' .
     'Needs to be stopped manually.';
 
-    public function __construct(private QueueFactoryInterface $queueFactory, private LoopInterface $loop, private array $channels)
-    {
+    public function __construct(
+        private readonly QueueProviderInterface $queueProvider,
+        private readonly LoopInterface $loop,
+        private readonly array $channels
+    ) {
         parent::__construct();
     }
 
@@ -45,7 +48,7 @@ public function configure(): void
                 'm',
                 InputOption::VALUE_REQUIRED,
                 'Maximum number of messages to process in each channel before switching to another channel. ' .
-                   'Default is 0 (no limits).',
+                'Default is 0 (no limits).',
                 0,
             );
 
@@ -57,17 +60,17 @@ protected function execute(InputInterface $input, OutputInterface $output): int
         $queues = [];
         /** @var string $channel */
         foreach ($input->getArgument('channel') as $channel) {
-            $queues[] = $this->queueFactory->get($channel);
+            $queues[] = $this->queueProvider->get($channel);
         }
 
         while ($this->loop->canContinue()) {
             $hasMessages = false;
             foreach ($queues as $queue) {
-                $hasMessages = $queue->run((int)$input->getOption('maximum')) > 0 || $hasMessages;
+                $hasMessages = $queue->run((int) $input->getOption('maximum')) > 0 || $hasMessages;
             }
 
             if (!$hasMessages) {
-                $pauseSeconds = (int)$input->getOption('pause');
+                $pauseSeconds = (int) $input->getOption('pause');
                 if ($pauseSeconds < 0) {
                     $pauseSeconds = 1;
                 }

src/Command/ListenCommand.php

@@ -8,16 +8,17 @@
 use Symfony\Component\Console\Input\InputArgument;
 use Symfony\Component\Console\Input\InputInterface;
 use Symfony\Component\Console\Output\OutputInterface;
-use Yiisoft\Queue\QueueFactory;
-use Yiisoft\Queue\QueueFactoryInterface;
+use Yiisoft\Queue\Provider\QueueProviderInterface;
+use Yiisoft\Queue\Queue;
 
 final class ListenCommand extends Command
 {
     protected static $defaultName = 'queue:listen';
     protected static $defaultDescription = 'Listens the queue and executes messages as they come. Needs to be stopped manually.';
 
-    public function __construct(private QueueFactoryInterface $queueFactory)
-    {
+    public function __construct(
+        private readonly QueueProviderInterface $queueProvider
+    ) {
         parent::__construct();
     }
 
@@ -27,13 +28,13 @@ public function configure(): void
             'channel',
             InputArgument::OPTIONAL,
             'Queue channel name to connect to',
-            QueueFactory::DEFAULT_CHANNEL_NAME
+            Queue::DEFAULT_CHANNEL_NAME,
         );
     }
 
     protected function execute(InputInterface $input, OutputInterface $output): int
     {
-        $this->queueFactory
+        $this->queueProvider
             ->get($input->getArgument('channel'))
             ->listen();
 

src/Command/RunCommand.php

@@ -9,16 +9,18 @@
 use Symfony\Component\Console\Input\InputInterface;
 use Symfony\Component\Console\Input\InputOption;
 use Symfony\Component\Console\Output\OutputInterface;
-use Yiisoft\Queue\QueueFactoryInterface;
+use Yiisoft\Queue\Provider\QueueProviderInterface;
 
 final class RunCommand extends Command
 {
     protected static $defaultName = 'queue:run';
     protected static $defaultDescription = 'Runs all the existing messages in the given queues. ' .
         'Exits once messages are over.';
 
-    public function __construct(private QueueFactoryInterface $queueFactory, private array $channels)
-    {
+    public function __construct(
+        private readonly QueueProviderInterface $queueProvider,
+        private readonly array $channels,
+    ) {
         parent::__construct();
     }
 
@@ -45,7 +47,7 @@ protected function execute(InputInterface $input, OutputInterface $output): int
         /** @var string $channel */
         foreach ($input->getArgument('channel') as $channel) {
             $output->write("Processing channel $channel... ");
-            $count = $this->queueFactory
+            $count = $this->queueProvider
                 ->get($channel)
                 ->run((int)$input->getOption('maximum'));
 

src/Debug/QueueProviderInterfaceProxy.php

@@ -0,0 +1,28 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Yiisoft\Queue\Debug;
+
+use Yiisoft\Queue\Provider\QueueProviderInterface;
+use Yiisoft\Queue\QueueInterface;
+
+final class QueueProviderInterfaceProxy implements QueueProviderInterface
+{
+    public function __construct(
+        private readonly QueueProviderInterface $queueProvider,
+        private readonly QueueCollector $collector,
+    ) {
+    }
+
+    public function get(string $channel): QueueInterface
+    {
+        $queue = $this->queueProvider->get($channel);
+        return new QueueDecorator($queue, $this->collector);
+    }
+
+    public function has(string $channel): bool
+    {
+        return $this->queueProvider->has($channel);
+    }
+}

src/Exception/AdapterConfiguration/AdapterNotConfiguredException.php

@@ -7,7 +7,6 @@
 use RuntimeException;
 use Yiisoft\FriendlyException\FriendlyExceptionInterface;
 use Yiisoft\Queue\Queue;
-use Yiisoft\Queue\QueueFactory;
 
 class AdapterNotConfiguredException extends RuntimeException implements FriendlyExceptionInterface
 {
@@ -21,7 +20,6 @@ public function getName(): string
     public function getSolution(): ?string
     {
         $queueClass = Queue::class;
-        $factoryClass = QueueFactory::class;
 
         return <<<SOLUTION
             Adapter property must be set in the Queue object before you can use it.
@@ -32,7 +30,6 @@ public function getSolution(): ?string
             References:
             - $queueClass::\$adapter
             - $queueClass::withAdapter()
-            - $factoryClass::get()
             SOLUTION;
     }
 }