Split queue consumer contract

README.md

@@ -138,6 +138,8 @@ For setting up all classes manually, see the [Manual configuration](docs/guide/e
 To send a message to the queue, get the queue instance and call `push()`. Typically the queue is injected as a dependency:
 
 ```php
+use Yiisoft\Queue\QueueInterface;
+
 final readonly class Foo
 {
     public function __construct(private QueueInterface $queue) {}
@@ -159,11 +161,13 @@ By default, Yii Framework uses [yiisoft/yii-console](https://github.com/yiisoft/
 ```bash
 ./yii queue:run # Handle all existing messages in the queue
 ./yii queue:listen [queueName] # Start a daemon listening for new messages permanently from the specified queue
-./yii queue:listen-all [queueName [queueName2 [...]]] # Start a daemon listening for new messages permanently from all queues or specified list of queues (use with caution in production, recommended for dev only)
+./yii queue:listen-all [queueName [queueName2 [...]]] # Start a daemon listening for new messages permanently from all consumer-capable queues or specified list of queues (use with caution in production, recommended for dev only)
 ```
 
 See [Console commands](docs/guide/en/console-commands.md) for more details.
 
+Consumers are represented by `Yiisoft\Queue\QueueConsumerInterface`; it contains `run()` and `listen()`.
+
 > In case you're running the queue in synchronous mode (no adapter), `queue:listen` logs an info message and exits. The messages are processed immediately when pushed.
 
 ## Documentation

docs/guide/en/configuration-manual.md

@@ -98,16 +98,25 @@ $provider = new PredefinedQueueProvider([
 
 ## Running the queue
 
+Message consumption methods are available on `Yiisoft\Queue\QueueConsumerInterface`.
+The built-in `Queue` implements both `QueueInterface` for producing messages and `QueueConsumerInterface` for consuming them.
+
 ### Processing existing messages
 
 ```php
+use Yiisoft\Queue\QueueConsumerInterface;
+
+/** @var QueueConsumerInterface $queue */
 $queue->run();      // Process all messages
 $queue->run(10);    // Process up to 10 messages
 ```
 
 ### Listening for new messages
 
 ```php
+use Yiisoft\Queue\QueueConsumerInterface;
+
+/** @var QueueConsumerInterface $queue */
 $queue->listen();   // Run indefinitely
 ```
 

docs/guide/en/console-commands.md

@@ -6,7 +6,7 @@ If you are using [yiisoft/config](https://github.com/yiisoft/config) and [yiisof
 
 If you are using [symfony/console](https://github.com/symfony/console) directly, you should register the commands manually.
 
-> **Note:** The default queue name list (used when no queue names are passed to a command) is only available when using [yiisoft/config](https://github.com/yiisoft/config) and [yiisoft/yii-console](https://github.com/yiisoft/yii-console). Without them, you must pass the queue name list explicitly to the command constructor.
+> **Note:** When no queue names are passed to `queue:run` or `queue:listen-all`, the commands use the queue names provided by `QueueProviderInterface::getNames()` and skip queues that don't implement `QueueConsumerInterface`. Explicitly passed queue names must be consumer-capable.
 
 In [yiisoft/app](https://github.com/yiisoft/app) the `yii` console binary is provided out of the box.
 If you are using [yiisoft/yii-console](https://github.com/yiisoft/yii-console) or `symfony/console` without that template, invoke these commands the same way you invoke other console commands in your application.
@@ -17,7 +17,7 @@ The command `queue:run` obtains and handles messages until the queue is empty, t
 
 You can also narrow the scope of processed messages by specifying queue name(s) and maximum number of messages to process:
 
-- Specify one or more queue names to process. Messages from other queues will be ignored. Defaults to all registered queue names.
+- Specify one or more queue names to process. Messages from other queues will be ignored. Defaults to all registered consumer-capable queue names.
 - Use `--limit` to limit the number of messages processed. When set, command will exit either when all the messages are processed or when the maximum count is reached.
 
 The full command signature is:
@@ -39,7 +39,7 @@ yii queue:listen [queueName]
 
 The following command iterates through multiple queues and is meant to be used in development environment only, as it consumes a lot of CPU for iterating through queues. You can pass to it:
 
-- `queueName` argument(s). Specify one or more queue names to process. Messages from other queues will be ignored. Defaults to all registered queue names.
+- `queueName` argument(s). Specify one or more queue names to process. Messages from other queues will be ignored. Defaults to all registered consumer-capable queue names.
 - `--limit` option to limit the number of messages processed before switching to another queue. E.g. you set `--limit` to 500 and right now you have 1000 messages in `queue1`. This command will consume only 500 of them, then it will switch to `queue2` to see if there are any messages there. Defaults to `0` (no limit).
 - `--pause` option to specify the number of seconds to pause between checking queues when no messages are found. Defaults to `1`.
 

docs/guide/en/queue-names-advanced.md

@@ -7,6 +7,7 @@ Use this reference when you need to understand how queue names map to adapters,
 - A queue name (string or `BackedEnum`) is passed to `Yiisoft\Queue\Provider\QueueProviderInterface::get($queueName)`.
 - The provider returns a `Yiisoft\Queue\QueueInterface` instance configured for that name.
 - `QueueInterface::getName()` can be used for introspection; it returns the logical name the queue was created with.
+- Queues that can consume messages also implement `Yiisoft\Queue\QueueConsumerInterface`.
 
 ## Provider implementations
 

src/Command/ListenAllCommand.php

@@ -11,13 +11,18 @@
 use Symfony\Component\Console\Input\InputOption;
 use Symfony\Component\Console\Output\OutputInterface;
 use Yiisoft\Queue\Cli\LoopInterface;
+use Yiisoft\Queue\Provider\InvalidQueueConfigException;
 use Yiisoft\Queue\Provider\QueueProviderInterface;
+use Yiisoft\Queue\QueueConsumerInterface;
+
+use function get_debug_type;
+use function sprintf;
 
 #[AsCommand(
     'queue:listen-all',
     'Listens the all the given queues and executes messages as they come. '
         . 'Meant to be used in development environment only. '
-        . 'Listens all configured queues by default in case you\'re using yiisoft/config. '
+        . 'Listens all consumer-capable configured queues by default. '
         . 'Needs to be stopped manually.',
 )]
 final class ListenAllCommand extends Command
@@ -38,7 +43,7 @@
             'queue',
             InputArgument::OPTIONAL | InputArgument::IS_ARRAY,
             'Queue name list to connect to',
-            $this->queueProvider->getNames(),
+            [],
         )
             ->addOption(
                 'pause',
@@ -61,14 +66,28 @@
 
     protected function execute(InputInterface $input, OutputInterface $output): int
     {
+        /** @var string[] $queueNames */
+        $queueNames = $input->getArgument('queue');
+        $queueIsRequired = $queueNames !== [];
+        if (!$queueIsRequired) {
+            $queueNames = $this->queueProvider->getNames();
+        }
+
         $queues = [];
         /** @var string $queue */
-        foreach ($input->getArgument('queue') as $queue) {
-            $queues[] = $this->queueProvider->get($queue);
+        foreach ($queueNames as $queue) {
+            $queueConsumer = $this->getQueueConsumer($queue, $queueIsRequired);
+            if ($queueConsumer !== null) {
+                $queues[] = $queueConsumer;
+            }
+        }
+
+        if ($queues === []) {
+            return Command::SUCCESS;
         }
 
         $pauseSeconds = (int) $input->getOption('pause');
         if ($pauseSeconds < 0) {
             $pauseSeconds = 1;
         }
 
@@ -86,4 +105,26 @@
 
         return 0;
     }
+
+    private function getQueueConsumer(string $name, bool $required): ?QueueConsumerInterface
+    {
+        $queue = $this->queueProvider->get($name);
+
+        if (!$queue instanceof QueueConsumerInterface) {
+            if (!$required) {
+                return null;
+            }
+
+            throw new InvalidQueueConfigException(
+                sprintf(
+                    'Queue "%s" must implement "%s" to consume messages. Got "%s" instead.',
+                    $name,
+                    QueueConsumerInterface::class,
+                    get_debug_type($queue),
+                ),
+            );
+        }
+
+        return $queue;
+    }
 }

src/Command/ListenCommand.php

@@ -9,7 +9,12 @@
 use Symfony\Component\Console\Input\InputArgument;
 use Symfony\Component\Console\Input\InputInterface;
 use Symfony\Component\Console\Output\OutputInterface;
+use Yiisoft\Queue\Provider\InvalidQueueConfigException;
 use Yiisoft\Queue\Provider\QueueProviderInterface;
+use Yiisoft\Queue\QueueConsumerInterface;
+
+use function get_debug_type;
+use function sprintf;
 
 #[AsCommand(
     'queue:listen',
@@ -37,8 +42,26 @@ protected function execute(InputInterface $input, OutputInterface $output): int
     {
         $queueName = (string) $input->getArgument('queue');
 
-        $this->queueProvider->get($queueName)->listen();
+        $this->getQueueConsumer($queueName)->listen();
 
         return Command::SUCCESS;
     }
+
+    private function getQueueConsumer(string $name): QueueConsumerInterface
+    {
+        $queue = $this->queueProvider->get($name);
+
+        if (!$queue instanceof QueueConsumerInterface) {
+            throw new InvalidQueueConfigException(
+                sprintf(
+                    'Queue "%s" must implement "%s" to consume messages. Got "%s" instead.',
+                    $name,
+                    QueueConsumerInterface::class,
+                    get_debug_type($queue),
+                ),
+            );
+        }
+
+        return $queue;
+    }
 }

src/Command/RunCommand.php

@@ -10,7 +10,12 @@
 use Symfony\Component\Console\Input\InputInterface;
 use Symfony\Component\Console\Input\InputOption;
 use Symfony\Component\Console\Output\OutputInterface;
+use Yiisoft\Queue\Provider\InvalidQueueConfigException;
 use Yiisoft\Queue\Provider\QueueProviderInterface;
+use Yiisoft\Queue\QueueConsumerInterface;
+
+use function get_debug_type;
+use function sprintf;
 
 #[AsCommand(
     'queue:run',
@@ -30,7 +35,7 @@ public function configure(): void
             'queue',
             InputArgument::OPTIONAL | InputArgument::IS_ARRAY,
             'Queue name list to connect to.',
-            $this->queueProvider->getNames(),
+            [],
         )
             ->addOption(
                 'limit',
@@ -44,16 +49,48 @@ public function configure(): void
 
     protected function execute(InputInterface $input, OutputInterface $output): int
     {
+        /** @var string[] $queueNames */
+        $queueNames = $input->getArgument('queue');
+        $queueIsRequired = $queueNames !== [];
+        if (!$queueIsRequired) {
+            $queueNames = $this->queueProvider->getNames();
+        }
+
         /** @var string $queue */
-        foreach ($input->getArgument('queue') as $queue) {
+        foreach ($queueNames as $queue) {
+            $queueConsumer = $this->getQueueConsumer($queue, $queueIsRequired);
+            if ($queueConsumer === null) {
+                continue;
+            }
+
             $output->write("Processing queue $queue... ");
-            $count = $this->queueProvider
-                ->get($queue)
-                ->run((int) $input->getOption('limit'));
+            $count = $queueConsumer->run((int) $input->getOption('limit'));
 
             $output->writeln("Messages processed: $count.");
         }
 
         return 0;
     }
+
+    private function getQueueConsumer(string $name, bool $required): ?QueueConsumerInterface
+    {
+        $queue = $this->queueProvider->get($name);
+
+        if (!$queue instanceof QueueConsumerInterface) {
+            if (!$required) {
+                return null;
+            }
+
+            throw new InvalidQueueConfigException(
+                sprintf(
+                    'Queue "%s" must implement "%s" to consume messages. Got "%s" instead.',
+                    $name,
+                    QueueConsumerInterface::class,
+                    get_debug_type($queue),
+                ),
+            );
+        }
+
+        return $queue;
+    }
 }

src/Debug/QueueConsumerDecorator.php

@@ -0,0 +1,47 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Yiisoft\Queue\Debug;
+
+use Yiisoft\Queue\MessageStatus;
+use Yiisoft\Queue\Message\MessageInterface;
+use Yiisoft\Queue\QueueConsumerInterface;
+use Yiisoft\Queue\QueueInterface;
+
+final class QueueConsumerDecorator implements QueueInterface, QueueConsumerInterface
+{
+    private readonly QueueDecorator $queueDecorator;
+
+    public function __construct(
+        private readonly QueueInterface&QueueConsumerInterface $queue,
+        QueueCollector $collector,
+    ) {
+        $this->queueDecorator = new QueueDecorator($queue, $collector);
+    }
+
+    public function status(string|int $id): MessageStatus
+    {
+        return $this->queueDecorator->status($id);
+    }
+
+    public function push(MessageInterface $message): MessageInterface
+    {
+        return $this->queueDecorator->push($message);
+    }
+
+    public function run(int $max = 0): int
+    {
+        return $this->queue->run($max);
+    }
+
+    public function listen(): void
+    {
+        $this->queue->listen();
+    }
+
+    public function getName(): string
+    {
+        return $this->queueDecorator->getName();
+    }
+}

src/Debug/QueueDecorator.php

@@ -36,16 +36,6 @@ public function push(MessageInterface $message): MessageInterface
         return $message;
     }
 
-    public function run(int $max = 0): int
-    {
-        return $this->queue->run($max);
-    }
-
-    public function listen(): void
-    {
-        $this->queue->listen();
-    }
-
     public function getName(): string
     {
         return $this->queue->getName();

src/Debug/QueueProviderInterfaceProxy.php

@@ -6,6 +6,7 @@
 
 use BackedEnum;
 use Yiisoft\Queue\Provider\QueueProviderInterface;
+use Yiisoft\Queue\QueueConsumerInterface;
 use Yiisoft\Queue\QueueInterface;
 
 final class QueueProviderInterfaceProxy implements QueueProviderInterface
@@ -19,7 +20,9 @@ public function get(string|BackedEnum $name): QueueInterface
     {
         $queue = $this->queueProvider->get($name);
 
-        return new QueueDecorator($queue, $this->collector);
+        return $queue instanceof QueueConsumerInterface
+            ? new QueueConsumerDecorator($queue, $this->collector)
+            : new QueueDecorator($queue, $this->collector);
     }
 
     public function has(string|BackedEnum $name): bool

src/Queue.php

@@ -17,7 +17,7 @@
 use Yiisoft\Queue\Message\IdEnvelope;
 use Yiisoft\Queue\Provider\QueueProviderInterface;
 
-final class Queue implements QueueInterface
+final class Queue implements QueueInterface, QueueConsumerInterface
 {
     private string $name;