Replace queue commands with queue:consume

CHANGELOG.md

@@ -2,6 +2,4 @@
 
 ## 1.0.0 under development
 
-- Initial release.
-- Bug #341: Pass the message a handler was registered for to that handler instead of the envelope an adapter wrapped it in, so a handler no longer reads properties off an `IdEnvelope`, `DelayEnvelope` or `FailureEnvelope` and works on empty values; what the envelope carried comes with the message as meta, and consume middleware still receive the envelope itself (@KalimeroMK)
-- Bug #347: `SignalLoop` installs signal handlers only while messages are consumed and restores the previous ones afterwards, instead of replacing them when it is created (@samdark)
+- Initial release.

README.md

@@ -156,19 +156,19 @@ final readonly class Foo
 
 ### 5. Handle queued messages
 
-By default, Yii Framework uses [yiisoft/yii-console](https://github.com/yiisoft/yii-console) to run CLI commands. If you installed [yiisoft/app](https://github.com/yiisoft/app) or [yiisoft/app-api](https://github.com/yiisoft/app-api), you can run the queue worker with one of these two commands:
+By default, Yii Framework uses [yiisoft/yii-console](https://github.com/yiisoft/yii-console) to run CLI commands. If you installed [yiisoft/app](https://github.com/yiisoft/app) or [yiisoft/app-api](https://github.com/yiisoft/app-api), you can run the queue worker with `queue:consume`:
 
 ```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 consumer-capable queues or specified list of queues (use with caution in production, recommended for dev only)
+./yii queue:consume --once # Handle all existing messages in the queue
+./yii queue:consume [queueName] # Start a daemon listening for new messages permanently from the specified queue
+./yii queue:consume --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.
 
 Producers use `Yiisoft\Queue\QueueProducerInterface` (`push()`, `status()`, `getQueueName()`); consumers use `Yiisoft\Queue\QueueConsumerInterface` (`run()`, `listen()`). See [capability configuration](docs/guide/en/queue-capabilities.md) for the strict role map used when named queues are configured.
 
-> 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.
+> In case you're running the queue in synchronous mode (no adapter), `queue:consume` logs an info message and exits. The messages are processed immediately when pushed.
 
 ## Documentation
 

config/params.php

@@ -2,9 +2,7 @@
 
 declare(strict_types=1);
 
-use Yiisoft\Queue\Command\ListenAllCommand;
-use Yiisoft\Queue\Command\ListenCommand;
-use Yiisoft\Queue\Command\RunCommand;
+use Yiisoft\Queue\Command\ConsumeCommand;
 use Yiisoft\Queue\Debug\QueueCollector;
 use Yiisoft\Queue\Debug\QueueConsumerProviderProxy;
 use Yiisoft\Queue\Debug\QueueProducerProviderProxy;
@@ -18,9 +16,7 @@
 return [
     'yiisoft/yii-console' => [
         'commands' => [
-            'queue:run' => RunCommand::class,
-            'queue:listen' => ListenCommand::class,
-            'queue:listen-all' => ListenAllCommand::class,
+            'queue:consume' => ConsumeCommand::class,
         ],
     ],
     'yiisoft/queue' => [

docs/guide/en/best-practices.md

@@ -410,7 +410,7 @@ return [
 
  ```bash
  # Running worker manually without supervision in production
- php yii queue:listen
+ php yii queue:consume
  ```
 
 #### Good

docs/guide/en/console-commands.md

@@ -1,19 +1,20 @@
 # Console commands
 
-Yii Queue provides several console commands for processing queued messages.
+Yii Queue provides the `queue:consume` console command for processing queued messages.
 
-If you are using [yiisoft/config](https://github.com/yiisoft/config) and [yiisoft/yii-console](https://github.com/yiisoft/yii-console), the commands are registered automatically.
+If you are using [yiisoft/config](https://github.com/yiisoft/config) and [yiisoft/yii-console](https://github.com/yiisoft/yii-console), the command is registered automatically.
 
-If you are using [symfony/console](https://github.com/symfony/console) directly, you should register the commands manually.
+If you are using [symfony/console](https://github.com/symfony/console) directly, you should register `Yiisoft\Queue\Command\ConsumeCommand` manually, passing a consumer provider and the same loop instance used by its consumers.
 
-> **Note:** `queue:run` and `queue:listen-all` use `QueueConsumerProviderInterface::getConsumerQueueNames()` when no queue names are passed. Explicitly passed names are resolved with `getConsumer()` and must have a consumer role.
+> **Note:** `queue:consume --once` and `queue:consume --all` use `QueueConsumerProviderInterface::getConsumerQueueNames()` when no queue names are passed. Explicitly passed names are resolved with `getConsumer()` and must have a consumer role.
 
 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.
 
 ## 1. Run queued messages and exit
 
-The command `queue:run` obtains and handles messages until the queue is empty, then exits.  
+`queue:consume --once` handles existing messages until the queues are empty, then exits.
+`--once` means one pass through the selected queues, not one message.
 
 You can also narrow the scope of processed messages by specifying queue name(s) and maximum number of messages to process:
 
@@ -22,19 +23,22 @@ You can also narrow the scope of processed messages by specifying queue name(s)
 
 The full command signature is:
 ```sh
-yii queue:run [queueName1 [queueName2 [...]]] --limit=100
+yii queue:consume --once [queueName1 [queueName2 [...]]] --limit=100
 ```
 
 ## 2. Listen for queued messages and process them continuously
 
 The following command launches a daemon, which infinitely consumes messages from a single queue. This command receives an optional `queueName` argument to specify which queue to listen to, defaults to the queue name `yii-queue`.
 
 ```sh
-yii queue:listen [queueName]
+yii queue:consume [queueName]
 ```
 
 > **Note:** If the queue is not configured with an adapter (synchronous mode), the command logs an info message and exits gracefully.
 
+The `--limit` and `--pause` options are not accepted in this single-queue listening mode.
+Use `--once --limit=100` to process up to 100 existing messages per queue and exit.
+
 ## 3. Listen to multiple queues
 
 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:
@@ -43,9 +47,12 @@ The following command iterates through multiple queues and is meant to be used i
 - `--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`.
 
+Use `--all --once` to process those queues for one pass and exit. Passing several queue names requires `--all` or `--once`.
+`--once` takes precedence over `--all`. The `--pause` option is accepted only with `--all` without `--once`.
+
 The full command signature is:
 ```sh
-yii queue:listen-all [queueName1 [queueName2 [...]]] --pause=1 --limit=0
+yii queue:consume --all [queueName1 [queueName2 [...]]] --pause=1 --limit=0
 ```
 
 For long-running processes, graceful shutdown is controlled by `LoopInterface`. When `ext-pcntl` is available,

docs/guide/en/loops.md

@@ -6,7 +6,7 @@ The loop is evaluated to determine whether it can continue:
 
 - Before each message in `QueueConsumer::run()` and `QueueConsumer::listen()`. A message taken from the queue after the
   loop stopped is not processed and goes back to the queue.
-- On each iteration of `queue:listen-all`.
+- On each iteration of `queue:consume --all`.
 
 When `canContinue()` returns `false`, consuming stops gracefully (as soon as the current message is finished).
 
@@ -104,13 +104,13 @@ use Yiisoft\Queue\Cli\SignalLoop;
 $loop = new SignalLoop(memorySoftLimit: 256 * 1024 * 1024);
 ```
 
-When creating `RunCommand` or `ListenAllCommand` manually, pass the same loop instance to the command and its consumers.
+When creating `ConsumeCommand` manually, pass the same loop instance to the command and its consumers.
 This keeps graceful shutdown active between queues. For example, with your configured consumer provider:
 
 ```php
-use Yiisoft\Queue\Command\RunCommand;
+use Yiisoft\Queue\Command\ConsumeCommand;
 
-$command = new RunCommand($consumerProvider, $loop);
+$command = new ConsumeCommand($consumerProvider, $loop);
 ```
 
 For your own execution loop, use `start()` and `finish()` to restore application signal handlers when work ends:

docs/guide/en/process-managers.md

@@ -44,7 +44,7 @@ PartOf=yii-queue.target
 Type=simple
 User=www-data
 WorkingDirectory=/var/www/app
-ExecStart=/usr/bin/php /var/www/app/yii queue:listen
+ExecStart=/usr/bin/php /var/www/app/yii queue:consume
 Restart=always
 RestartSec=5
 
@@ -145,7 +145,7 @@ Supervisor manages multiple worker processes using a single configuration file a
 # /etc/supervisor/conf.d/yii-queue.conf
 [program:yii-queue-worker]
 process_name=%(program_name)s_%(process_num)02d
-command=/usr/bin/php /var/www/app/yii queue:listen
+command=/usr/bin/php /var/www/app/yii queue:consume
 autostart=true
 autorestart=true
 user=www-data
@@ -199,10 +199,10 @@ sudo supervisorctl start yii-queue-worker:*
 
 ## Cron
 
-For simple workloads without a persistent process manager, you can run `queue:run` via cron. It processes all pending messages and exits.
+For simple workloads without a persistent process manager, you can run `queue:consume --once` via cron. It processes all pending messages and exits.
 
 ```cron
-* * * * * /usr/bin/php /var/www/app/yii queue:run
+* * * * * /usr/bin/php /var/www/app/yii queue:consume --once
 ```
 
-This starts a worker every minute. Use this only when the message volume is low and a processing delay of a minute or more is acceptable. For high-throughput or latency-sensitive queues, use `systemd` or Supervisor with `queue:listen` instead.
+This starts a worker every minute. Use this only when the message volume is low and a processing delay of a minute or more is acceptable. For high-throughput or latency-sensitive queues, use `systemd` or Supervisor with `queue:consume` instead.

docs/guide/en/queue-names.md

@@ -82,6 +82,6 @@ Both typed providers accept strings and `BackedEnum` values. Use `getProducerQue
 
 ## Running workers
 
-`queue:run` and `queue:listen-all` use all configured consumer names when no names are supplied. `queue:listen` and explicitly named commands resolve that name through `QueueConsumerProviderInterface`, so a producer-only name cannot be consumed. See [Console commands](console-commands.md) for details.
+`queue:consume --once` and `queue:consume --all` use all configured consumer names when no names are supplied. `queue:consume` and explicitly named commands resolve that name through `QueueConsumerProviderInterface`, so a producer-only name cannot be consumed. See [Console commands](console-commands.md) for details.
 
 For provider implementations and manually constructed role maps, see [Advanced queue name internals](queue-names-advanced.md). For migration details, see [Producer and consumer capabilities](queue-capabilities.md).

docs/guide/en/usage.md

@@ -73,9 +73,9 @@ Most adapters can be consumed using [console commands](./console-commands.md) re
 If you configured multiple [queue names](./queue-names.md), you can choose which queue to consume with console commands:
 
 ```sh
-yii queue:listen [queueName]
-yii queue:run [queueName1 [queueName2 [...]]]
-yii queue:listen-all [queueName1 [queueName2 [...]]]
+yii queue:consume [queueName]
+yii queue:consume --once [queueName1 [queueName2 [...]]]
+yii queue:consume --all [queueName1 [queueName2 [...]]]
 ```
 
 

docs/guide/en/worker.md

@@ -4,6 +4,6 @@ To use a worker, you should resolve the worker's dependencies (e.g., through DI
 
 ## Starting Workers
 
-To start a worker, you should run the console commands such as `queue:run`, `queue:listen`, and `queue:listen-all`. See [Console commands](console-commands.md) for details.
+To start a worker, you should run the `queue:consume` console command. See [Console commands](console-commands.md) for details.
 
 For production-grade process management with `systemd`, Supervisor, or cron, see [Running workers in production](process-managers.md).

src/Cli/ScopedLoopInterface.php

@@ -8,7 +8,7 @@
  * A loop that changes process-wide state, such as signal handlers, only while messages are consumed.
  *
  * {@see ScopedLoopInterface::start()} is called before consuming starts and {@see ScopedLoopInterface::finish()} after it ends. Calls may be nested, for example
- * when `queue:listen-all` runs consumers that share the same loop: the state is changed by the outermost
+ * when `queue:consume --all` runs consumers that share the same loop: the state is changed by the outermost
  * {@see ScopedLoopInterface::start()} and restored by the matching {@see ScopedLoopInterface::finish()}.
  */
 interface ScopedLoopInterface extends LoopInterface

src/Command/ConsumeCommand.php

@@ -0,0 +1,160 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Yiisoft\Queue\Command;
+
+use Symfony\Component\Console\Attribute\AsCommand;
+use Symfony\Component\Console\Command\Command;
+use Symfony\Component\Console\Input\InputArgument;
+use Symfony\Component\Console\Input\InputInterface;
+use Symfony\Component\Console\Input\InputOption;
+use Symfony\Component\Console\Output\OutputInterface;
+use Yiisoft\Queue\Cli\LoopInterface;
+use Yiisoft\Queue\Cli\ScopedLoopInterface;
+use Yiisoft\Queue\DefaultQueue;
+use Yiisoft\Queue\Provider\QueueConsumerProviderInterface;
+
+use function count;
+
+#[AsCommand('queue:consume', 'Consumes queued messages. Listens continuously unless --once is passed.')]
+final class ConsumeCommand extends Command
+{
+    public function __construct(
+        private readonly QueueConsumerProviderInterface $queueProvider,
+        private readonly LoopInterface $loop,
+    ) {
+        parent::__construct();
+    }
+
+    public function configure(): void
+    {
+        $this
+            ->addArgument('queue', InputArgument::OPTIONAL | InputArgument::IS_ARRAY, 'Queue names to consume.')
+            ->addOption(
+                'once',
+                null,
+                InputOption::VALUE_NONE,
+                'Process existing messages in one pass and exit. Without names, use all configured consumer queues.',
+            )
+            ->addOption(
+                'all',
+                null,
+                InputOption::VALUE_NONE,
+                'Continuously poll queues. Without names, use all configured consumer queues. '
+                    . 'Recommended for development only.',
+            )
+            ->addOption(
+                'limit',
+                'm',
+                InputOption::VALUE_REQUIRED,
+                'Maximum messages per queue in each pass with --once or --all. Default: 0 (no limit).',
+            )
+            ->addOption(
+                'pause',
+                'p',
+                InputOption::VALUE_REQUIRED,
+                'Seconds between idle passes with --all without --once. Default: 1.',
+            )
+            ->addUsage('[queue]')
+            ->addUsage('--once [queue1 [queue2 [...]]] --limit=100')
+            ->addUsage('--all [queue1 [queue2 [...]]] --pause=1 --limit=100')
+            ->setHelp(
+                'Without queue names, listen to ' . DefaultQueue::NAME . '. '
+                    . '--once and --all instead use all configured consumer queues. '
+                    . '--once takes precedence over --all.',
+            );
+    }
+
+    protected function execute(InputInterface $input, OutputInterface $output): int
+    {
+        /** @var string[] $queueNames */
+        $queueNames = $input->getArgument('queue');
+        $once = (bool) $input->getOption('once');
+        $all = (bool) $input->getOption('all');
+        $limit = $input->getOption('limit');
+        $pauseSeconds = $input->getOption('pause');
+
+        if (!$once && !$all && $limit !== null) {
+            $output->writeln('<error>--limit requires --once or --all.</error>');
+
+            return Command::INVALID;
+        }
+
+        if ((!$all || $once) && $pauseSeconds !== null) {
+            $output->writeln('<error>--pause requires --all without --once.</error>');
+
+            return Command::INVALID;
+        }
+
+        if (!$once && !$all) {
+            if (count($queueNames) > 1) {
+                $output->writeln('<error>Use --all or --once to consume multiple queues.</error>');
+
+                return Command::INVALID;
+            }
+
+            $this->queueProvider->getConsumer($queueNames[0] ?? DefaultQueue::NAME)->listen();
+
+            return Command::SUCCESS;
+        }
+
+        if ($queueNames === []) {
+            $queueNames = $this->queueProvider->getConsumerQueueNames();
+        }
+
+        $consumers = [];
+        foreach ($queueNames as $queueName) {
+            $consumers[$queueName] = $this->queueProvider->getConsumer($queueName);
+        }
+
+        if ($consumers === []) {
+            $output->writeln('No consumers are configured.');
+
+            return Command::SUCCESS;
+        }
+
+        $limit = (int) $limit;
+        $pauseSeconds = (int) ($pauseSeconds ?? 1);
+        if ($pauseSeconds < 0) {
+            $pauseSeconds = 1;
+        }
+
+        if ($this->loop instanceof ScopedLoopInterface) {
+            $this->loop->start();
+        }
+
+        try {
+            while ($once || $this->loop->canContinue()) {
+                $hasMessages = false;
+                foreach ($consumers as $queueName => $consumer) {
+                    if ($once) {
+                        $output->write("Processing queue $queueName... ");
+                    }
+
+                    $count = $consumer->run($limit);
+                    $hasMessages = $count > 0 || $hasMessages;
+
+                    if ($once) {
+                        $output->writeln("Messages processed: $count.");
+                    }
+                }
+
+                if ($once) {
+                    break;
+                }
+
+                if (!$hasMessages && $this->loop->canContinue()) {
+                    /** @psalm-var 0|positive-int $pauseSeconds */
+                    sleep($pauseSeconds);
+                }
+            }
+        } finally {
+            if ($this->loop instanceof ScopedLoopInterface) {
+                $this->loop->finish();
+            }
+        }
+
+        return Command::SUCCESS;
+    }
+}