From d6a33351d95f7d9966d15c2f0aceb0096c2884d7 Mon Sep 17 00:00:00 2001 From: Alexander Makarov Date: Sun, 1 Nov 2026 18:46:30 +0300 Subject: [PATCH 1/4] Replace queue commands with consume #350 --- CHANGELOG.md | 1 + README.md | 10 +-- config/params.php | 8 +- docs/guide/en/best-practices.md | 2 +- docs/guide/en/console-commands.md | 32 +++++-- docs/guide/en/loops.md | 8 +- docs/guide/en/process-managers.md | 10 +-- docs/guide/en/queue-names.md | 2 +- docs/guide/en/usage.md | 6 +- docs/guide/en/worker.md | 2 +- src/Cli/ScopedLoopInterface.php | 2 +- ...istenAllCommand.php => ConsumeCommand.php} | 85 +++++++++++-------- src/Command/ListenCommand.php | 45 ---------- src/Command/RunCommand.php | 81 ------------------ tests/Integration/ConsumeCommandTest.php | 62 ++++++++++++++ tests/Integration/QueueProviderTest.php | 7 +- ...mandTest.php => ConsumeAllCommandTest.php} | 66 +++++++++++--- ...andTest.php => ConsumeOnceCommandTest.php} | 15 ++-- .../Unit/Command/ConsumeSingleCommandTest.php | 62 ++++++++++++++ tests/Unit/Command/ListenCommandTest.php | 23 ----- tests/Unit/SignalHandlingTest.php | 25 +++--- 21 files changed, 303 insertions(+), 251 deletions(-) rename src/Command/{ListenAllCommand.php => ConsumeCommand.php} (50%) delete mode 100644 src/Command/ListenCommand.php delete mode 100644 src/Command/RunCommand.php create mode 100644 tests/Integration/ConsumeCommandTest.php rename tests/Unit/Command/{ListenAllCommandTest.php => ConsumeAllCommandTest.php} (50%) rename tests/Unit/Command/{RunCommandTest.php => ConsumeOnceCommandTest.php} (68%) create mode 100644 tests/Unit/Command/ConsumeSingleCommandTest.php delete mode 100644 tests/Unit/Command/ListenCommandTest.php diff --git a/CHANGELOG.md b/CHANGELOG.md index 58b0792e..102ff2b6 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -3,4 +3,5 @@ ## 1.0.0 under development - Initial release. +- Chg #350: Replace `queue:run`, `queue:listen`, and `queue:listen-all` with `queue:consume` using `--once` and `--all` options (@samdark) - 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) diff --git a/README.md b/README.md index 8f39156e..b276ac76 100644 --- a/README.md +++ b/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 diff --git a/config/params.php b/config/params.php index e502d23a..ca1ca1ce 100644 --- a/config/params.php +++ b/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' => [ diff --git a/docs/guide/en/best-practices.md b/docs/guide/en/best-practices.md index 8b67f9f7..58a53b62 100644 --- a/docs/guide/en/best-practices.md +++ b/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 diff --git a/docs/guide/en/console-commands.md b/docs/guide/en/console-commands.md index 9e214e9c..fc9e04ac 100644 --- a/docs/guide/en/console-commands.md +++ b/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,7 +23,7 @@ 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 @@ -30,7 +31,7 @@ yii queue:run [queueName1 [queueName2 [...]]] --limit=100 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. @@ -43,10 +44,25 @@ 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`. + 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, the default `SignalLoop` handles signals such as `SIGTERM`/`SIGINT`. + +## Migrating existing commands + +Update process manager and cron configurations as follows: + +| Previous command | Replacement | +| --- | --- | +| `queue:run` | `queue:consume --once` | +| `queue:listen [queueName]` | `queue:consume [queueName]` | +| `queue:listen-all [queueNames...]` | `queue:consume --all [queueNames...]` | + +The previous command names are no longer registered. For manual registration, replace `RunCommand`, `ListenCommand`, +and `ListenAllCommand` with `ConsumeCommand`, passing the consumer provider and loop to its constructor. diff --git a/docs/guide/en/loops.md b/docs/guide/en/loops.md index 0e925c89..8192e306 100644 --- a/docs/guide/en/loops.md +++ b/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: diff --git a/docs/guide/en/process-managers.md b/docs/guide/en/process-managers.md index e5b3654b..4e9f09a7 100644 --- a/docs/guide/en/process-managers.md +++ b/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. diff --git a/docs/guide/en/queue-names.md b/docs/guide/en/queue-names.md index bee1d1bc..15798f5b 100644 --- a/docs/guide/en/queue-names.md +++ b/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). diff --git a/docs/guide/en/usage.md b/docs/guide/en/usage.md index ba4de48c..b39c48f5 100644 --- a/docs/guide/en/usage.md +++ b/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 [...]]] ``` diff --git a/docs/guide/en/worker.md b/docs/guide/en/worker.md index b4e48a2a..8d894565 100644 --- a/docs/guide/en/worker.md +++ b/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). diff --git a/src/Cli/ScopedLoopInterface.php b/src/Cli/ScopedLoopInterface.php index 53985eb4..69a2ba6e 100644 --- a/src/Cli/ScopedLoopInterface.php +++ b/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 diff --git a/src/Command/ListenAllCommand.php b/src/Command/ConsumeCommand.php similarity index 50% rename from src/Command/ListenAllCommand.php rename to src/Command/ConsumeCommand.php index f3bdde4d..8bf4e85d 100644 --- a/src/Command/ListenAllCommand.php +++ b/src/Command/ConsumeCommand.php @@ -12,16 +12,13 @@ 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; -#[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 consumer-capable configured queues by default. ' - . 'Needs to be stopped manually.', -)] -final class ListenAllCommand extends Command +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, @@ -30,48 +27,56 @@ public function __construct( parent::__construct(); } - /** - * @codeCoverageIgnore - */ public function configure(): void { - $this->addArgument( - 'queue', - InputArgument::OPTIONAL | InputArgument::IS_ARRAY, - 'Queue name list to connect to', - [], - ) + $this + ->addArgument('queue', InputArgument::OPTIONAL | InputArgument::IS_ARRAY, 'Queue names to consume.') + ->addOption('once', null, InputOption::VALUE_NONE, 'Process existing messages and exit when queues are empty.') ->addOption( - 'pause', - 'p', - InputOption::VALUE_REQUIRED, - 'Pause between queue iterations in seconds. May save some CPU. Default: 1', - 1, + 'all', + null, + InputOption::VALUE_NONE, + 'Poll all consumer queues, or the specified queues. Recommended for development only.', ) ->addOption( 'limit', 'm', InputOption::VALUE_REQUIRED, - 'Maximum number of messages to process in each queue before switching to another queue. ' - . 'Default is 0 (no limits).', + 'Maximum messages per queue in each pass with --once or --all. 0 means no limit.', 0, - ); - - $this->addUsage('[queue1 [queue2 [...]]] [--pause=] [--limit=]'); + ) + ->addOption('pause', 'p', InputOption::VALUE_REQUIRED, 'Seconds between idle passes with --all.', 1) + ->addUsage('[queue]') + ->addUsage('--once [queue1 [queue2 [...]]] --limit=100') + ->addUsage('--all [queue1 [queue2 [...]]] --pause=1 --limit=100'); } 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'); + + if (!$once && !$all) { + if (count($queueNames) > 1) { + $output->writeln('Use --all or --once to consume multiple queues.'); + + return Command::INVALID; + } + + $this->queueProvider->getConsumer($queueNames[0] ?? DefaultQueue::NAME)->listen(); + + return Command::SUCCESS; + } + if ($queueNames === []) { $queueNames = $this->queueProvider->getConsumerQueueNames(); } $consumers = []; - /** @var string $queueName */ foreach ($queueNames as $queueName) { - $consumers[] = $this->queueProvider->getConsumer($queueName); + $consumers[$queueName] = $this->queueProvider->getConsumer($queueName); } if ($consumers === []) { @@ -80,22 +85,34 @@ protected function execute(InputInterface $input, OutputInterface $output): int return Command::SUCCESS; } + $limit = (int) $input->getOption('limit'); $pauseSeconds = (int) $input->getOption('pause'); if ($pauseSeconds < 0) { $pauseSeconds = 1; } - $limit = (int) $input->getOption('limit'); - if ($this->loop instanceof ScopedLoopInterface) { $this->loop->start(); } try { - while ($this->loop->canContinue()) { + while ($once || $this->loop->canContinue()) { $hasMessages = false; - foreach ($consumers as $consumer) { - $hasMessages = $consumer->run($limit) > 0 || $hasMessages; + 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()) { diff --git a/src/Command/ListenCommand.php b/src/Command/ListenCommand.php deleted file mode 100644 index 14b9e04c..00000000 --- a/src/Command/ListenCommand.php +++ /dev/null @@ -1,45 +0,0 @@ -addArgument( - 'queue', - InputArgument::OPTIONAL, - 'Queue name to connect to', - DefaultQueue::NAME, - ); - } - - protected function execute(InputInterface $input, OutputInterface $output): int - { - $queueName = (string) $input->getArgument('queue'); - - $this->queueProvider->getConsumer($queueName)->listen(); - - return Command::SUCCESS; - } -} diff --git a/src/Command/RunCommand.php b/src/Command/RunCommand.php deleted file mode 100644 index b57d676d..00000000 --- a/src/Command/RunCommand.php +++ /dev/null @@ -1,81 +0,0 @@ -addArgument( - 'queue', - InputArgument::OPTIONAL | InputArgument::IS_ARRAY, - 'Queue name list to connect to.', - [], - ) - ->addOption( - 'limit', - 'm', - InputOption::VALUE_REQUIRED, - 'Maximum number of messages to process in each queue. Default is 0 (no limits).', - 0, - ) - ->addUsage('[queue1 [queue2 [...]]] --limit 100'); - } - - protected function execute(InputInterface $input, OutputInterface $output): int - { - /** @var string[] $queueNames */ - $queueNames = $input->getArgument('queue'); - if ($queueNames === []) { - $queueNames = $this->queueProvider->getConsumerQueueNames(); - } - - $limit = (int) $input->getOption('limit'); - - // One scope for all queues, so that a signal received between two of them doesn't reach the default handler. - if ($this->loop instanceof ScopedLoopInterface) { - $this->loop->start(); - } - - try { - /** @var string $queueName */ - foreach ($queueNames as $queueName) { - $queueConsumer = $this->queueProvider->getConsumer($queueName); - - $output->write("Processing queue $queueName... "); - $count = $queueConsumer->run($limit); - - $output->writeln("Messages processed: $count."); - } - } finally { - if ($this->loop instanceof ScopedLoopInterface) { - $this->loop->finish(); - } - } - - return Command::SUCCESS; - } -} diff --git a/tests/Integration/ConsumeCommandTest.php b/tests/Integration/ConsumeCommandTest.php new file mode 100644 index 00000000..1076f923 --- /dev/null +++ b/tests/Integration/ConsumeCommandTest.php @@ -0,0 +1,62 @@ +createQueue($adapter, $name); + $messages[$name] = [ + IdEnvelope::fromMessage($producer->push(new GenericMessage('simple', null))), + IdEnvelope::fromMessage($producer->push(new GenericMessage('simple', null))), + ]; + $queues[$name] = ['consumer' => new QueueConsumer( + $this->getWorker(), + $this->getLoop(), + new NullLogger(), + $adapter, + $name, + )]; + } + $command = new ConsumeCommand(new PredefinedQueueProvider($queues), $this->getLoop()); + $output = new BufferedOutput(); + + $result = $command->run(new StringInput('--once first second --limit=1'), $output); + + self::assertSame(Command::SUCCESS, $result); + self::assertSame(2, $this->executionTimes); + foreach (['first', 'second'] as $name) { + self::assertSame(MessageStatus::DONE, $adapters[$name]->status($messages[$name][0]->getId())); + self::assertSame(MessageStatus::WAITING, $adapters[$name]->status($messages[$name][1]->getId())); + } + foreach ($messages['other'] as $message) { + self::assertSame(MessageStatus::WAITING, $adapters['other']->status($message->getId())); + } + self::assertSame( + "Processing queue first... Messages processed: 1.\nProcessing queue second... Messages processed: 1.\n", + $output->fetch(), + ); + } +} diff --git a/tests/Integration/QueueProviderTest.php b/tests/Integration/QueueProviderTest.php index cd9e5e31..799d9ccd 100644 --- a/tests/Integration/QueueProviderTest.php +++ b/tests/Integration/QueueProviderTest.php @@ -8,7 +8,7 @@ use Symfony\Component\Console\Input\StringInput; use Symfony\Component\Console\Output\NullOutput; use Yiisoft\Definitions\Reference; -use Yiisoft\Queue\Command\ListenCommand; +use Yiisoft\Queue\Command\ConsumeCommand; use Yiisoft\Queue\Debug\QueueCollector; use Yiisoft\Queue\Debug\QueueConsumerDecorator; use Yiisoft\Queue\Debug\QueueConsumerProviderProxy; @@ -20,6 +20,7 @@ use Yiisoft\Queue\Provider\QueueNotFoundException; use Yiisoft\Queue\QueueConsumerInterface; use Yiisoft\Queue\QueueProducerInterface; +use Yiisoft\Queue\Stubs\StubLoop; use Yiisoft\Queue\Stubs\StubQueueConsumer; use Yiisoft\Queue\Stubs\StubQueueProducer; use Yiisoft\Test\Support\Container\SimpleContainer; @@ -54,7 +55,7 @@ public function testFactoryRoleMapsResolveThroughContainerAndKeepCapabilitiesSep $provider->getConsumer('producer-only'); } - public function testPredefinedRoleMapsAndListenCommandUseConsumerOnlyService(): void + public function testPredefinedRoleMapsAndConsumeCommandUseConsumerOnlyService(): void { $consumer = $this->createMock(QueueConsumerInterface::class); $consumer->expects(self::once())->method('listen'); @@ -71,7 +72,7 @@ public function testPredefinedRoleMapsAndListenCommandUseConsumerOnlyService(): self::assertFalse($provider->hasConsumer('producer-only')); self::assertFalse($provider->hasProducer('consumer-only')); - self::assertSame(0, (new ListenCommand($provider))->run(new StringInput('consumer-only'), new NullOutput())); + self::assertSame(0, (new ConsumeCommand($provider, new StubLoop()))->run(new StringInput('consumer-only'), new NullOutput())); try { $provider->getProducer('consumer-only'); diff --git a/tests/Unit/Command/ListenAllCommandTest.php b/tests/Unit/Command/ConsumeAllCommandTest.php similarity index 50% rename from tests/Unit/Command/ListenAllCommandTest.php rename to tests/Unit/Command/ConsumeAllCommandTest.php index a2d6b214..367c7172 100644 --- a/tests/Unit/Command/ListenAllCommandTest.php +++ b/tests/Unit/Command/ConsumeAllCommandTest.php @@ -7,22 +7,68 @@ use PHPUnit\Framework\TestCase; use Symfony\Component\Console\Command\Command; use Symfony\Component\Console\Input\ArrayInput; +use Symfony\Component\Console\Input\StringInput; use Symfony\Component\Console\Output\BufferedOutput; use Symfony\Component\Console\Output\OutputInterface; use Yiisoft\Queue\Cli\LoopInterface; -use Yiisoft\Queue\Command\ListenAllCommand; +use Yiisoft\Queue\Command\ConsumeCommand; use Yiisoft\Queue\Provider\PredefinedQueueProvider; use Yiisoft\Queue\QueueConsumerInterface; use Yiisoft\Queue\Stubs\StubQueueProducer; -final class ListenAllCommandTest extends TestCase +final class ConsumeAllCommandTest extends TestCase { + public function testPollsSelectedQueuesRepeatedlyWithPerQueueLimit(): void + { + $calls = []; + $first = $this->createMock(QueueConsumerInterface::class); + $first->expects($this->exactly(2))->method('run')->with(3)->willReturnCallback( + static function () use (&$calls): int { + $calls[] = 'first'; + return 1; + }, + ); + $second = $this->createMock(QueueConsumerInterface::class); + $second->expects($this->exactly(2))->method('run')->with(3)->willReturnCallback( + static function () use (&$calls): int { + $calls[] = 'second'; + return 0; + }, + ); + $other = $this->createMock(QueueConsumerInterface::class); + $other->expects($this->never())->method('run'); + $command = new ConsumeCommand(new PredefinedQueueProvider([ + 'first' => ['consumer' => $first], + 'second' => ['consumer' => $second], + 'other' => ['consumer' => $other], + ]), $this->createLoop(2)); + + $result = $command->run(new StringInput('--all first second --limit=3'), new BufferedOutput()); + + self::assertSame(Command::SUCCESS, $result); + self::assertSame(['first', 'second', 'first', 'second'], $calls); + } + + public function testOnceWithAllExitsAfterOnePass(): void + { + $consumer = $this->createMock(QueueConsumerInterface::class); + $consumer->expects($this->once())->method('run')->willReturn(2); + $consumer->expects($this->never())->method('listen'); + $command = new ConsumeCommand(new PredefinedQueueProvider([ + 'consumer' => ['consumer' => $consumer], + ]), $this->createLoop(2)); + + $result = $command->run(new StringInput('--all --once'), new BufferedOutput()); + + self::assertSame(Command::SUCCESS, $result); + } + public function testReportsWhenNoConsumersAreConfigured(): void { - $command = new ListenAllCommand(new PredefinedQueueProvider([]), $this->createMock(LoopInterface::class)); + $command = new ConsumeCommand(new PredefinedQueueProvider([]), $this->createMock(LoopInterface::class)); $output = new BufferedOutput(); - self::assertSame(Command::SUCCESS, $command->run(new ArrayInput([], $command->getNativeDefinition()), $output)); + self::assertSame(Command::SUCCESS, $command->run(new ArrayInput(['--all' => true], $command->getNativeDefinition()), $output)); self::assertSame("No consumers are configured.\n", $output->fetch()); } @@ -30,11 +76,11 @@ public function testRunsOnlyConsumerRolesByDefault(): void { $consumer = $this->createMock(QueueConsumerInterface::class); $consumer->expects($this->once())->method('run')->willReturn(0); - $command = new ListenAllCommand(new PredefinedQueueProvider([ + $command = new ConsumeCommand(new PredefinedQueueProvider([ 'producer' => ['producer' => new StubQueueProducer()], 'consumer' => ['consumer' => $consumer], ]), $this->createLoop(1)); - $input = new ArrayInput(['--pause' => 0], $command->getNativeDefinition()); + $input = new ArrayInput(['--all' => true, '--pause' => 0], $command->getNativeDefinition()); self::assertSame(Command::SUCCESS, $command->run($input, $this->createMock(OutputInterface::class))); } @@ -42,10 +88,10 @@ public function testDoesNotPauseWhenTheLoopCannotContinue(): void { $consumer = $this->createMock(QueueConsumerInterface::class); $consumer->method('run')->willReturn(0); - $command = new ListenAllCommand(new PredefinedQueueProvider([ + $command = new ConsumeCommand(new PredefinedQueueProvider([ 'consumer' => ['consumer' => $consumer], ]), $this->createLoop(1)); - $input = new ArrayInput(['--pause' => 3], $command->getNativeDefinition()); + $input = new ArrayInput(['--all' => true, '--pause' => 3], $command->getNativeDefinition()); $start = hrtime(true); $command->run($input, $this->createMock(OutputInterface::class)); @@ -57,10 +103,10 @@ public function testPausesWhenThereAreNoMessages(): void { $consumer = $this->createMock(QueueConsumerInterface::class); $consumer->method('run')->willReturn(0); - $command = new ListenAllCommand(new PredefinedQueueProvider([ + $command = new ConsumeCommand(new PredefinedQueueProvider([ 'consumer' => ['consumer' => $consumer], ]), $this->createLoop(2)); - $input = new ArrayInput(['--pause' => 1], $command->getNativeDefinition()); + $input = new ArrayInput(['--all' => true, '--pause' => 1], $command->getNativeDefinition()); $start = hrtime(true); $command->run($input, $this->createMock(OutputInterface::class)); diff --git a/tests/Unit/Command/RunCommandTest.php b/tests/Unit/Command/ConsumeOnceCommandTest.php similarity index 68% rename from tests/Unit/Command/RunCommandTest.php rename to tests/Unit/Command/ConsumeOnceCommandTest.php index 133cdd83..d6370316 100644 --- a/tests/Unit/Command/RunCommandTest.php +++ b/tests/Unit/Command/ConsumeOnceCommandTest.php @@ -7,33 +7,34 @@ use PHPUnit\Framework\TestCase; use Symfony\Component\Console\Input\StringInput; use Symfony\Component\Console\Output\OutputInterface; -use Yiisoft\Queue\Command\RunCommand; +use Yiisoft\Queue\Command\ConsumeCommand; use Yiisoft\Queue\DefaultQueue; +use Yiisoft\Queue\Stubs\StubLoop; use Yiisoft\Queue\Provider\PredefinedQueueProvider; use Yiisoft\Queue\QueueConsumerInterface; use Yiisoft\Queue\Stubs\StubQueueProducer; -final class RunCommandTest extends TestCase +final class ConsumeOnceCommandTest extends TestCase { public function testRunsSelectedConsumer(): void { $consumer = $this->createMock(QueueConsumerInterface::class); $consumer->expects($this->once())->method('run')->with(5)->willReturn(3); - $command = new RunCommand(new PredefinedQueueProvider(['queue' => ['consumer' => $consumer]])); + $command = new ConsumeCommand(new PredefinedQueueProvider(['queue' => ['consumer' => $consumer]]), new StubLoop()); $output = $this->createMock(OutputInterface::class); $output->expects($this->once())->method('write')->with('Processing queue queue... '); $output->expects($this->once())->method('writeln')->with('Messages processed: 3.'); - self::assertSame(0, $command->run(new StringInput('queue --limit=5'), $output)); + self::assertSame(0, $command->run(new StringInput('--once queue --limit=5'), $output)); } public function testDefaultRunSkipsProducerOnlyQueues(): void { $consumer = $this->createMock(QueueConsumerInterface::class); $consumer->expects($this->once())->method('run')->willReturn(0); - $command = new RunCommand(new PredefinedQueueProvider([ + $command = new ConsumeCommand(new PredefinedQueueProvider([ 'producer' => ['producer' => new StubQueueProducer()], DefaultQueue::NAME => ['consumer' => $consumer], - ])); - self::assertSame(0, $command->run(new StringInput(''), $this->createMock(OutputInterface::class))); + ]), new StubLoop()); + self::assertSame(0, $command->run(new StringInput('--once'), $this->createMock(OutputInterface::class))); } } diff --git a/tests/Unit/Command/ConsumeSingleCommandTest.php b/tests/Unit/Command/ConsumeSingleCommandTest.php new file mode 100644 index 00000000..63e8ddbc --- /dev/null +++ b/tests/Unit/Command/ConsumeSingleCommandTest.php @@ -0,0 +1,62 @@ +createMock(QueueConsumerInterface::class); + $consumer->expects($this->once())->method('listen'); + $consumer->expects($this->never())->method('run'); + $command = new ConsumeCommand(new PredefinedQueueProvider([ + DefaultQueue::NAME => ['consumer' => $consumer], + ]), new StubLoop()); + $application = new Application(); + $application->setAutoExit(false); + $application->add($command); + + $result = $application->run(new StringInput('queue:consume'), new BufferedOutput()); + + self::assertSame(Command::SUCCESS, $result); + } + + public function testListensSelectedConsumer(): void + { + $consumer = $this->createMock(QueueConsumerInterface::class); + $consumer->expects($this->once())->method('listen'); + $command = new ConsumeCommand(new PredefinedQueueProvider(['queue' => ['consumer' => $consumer]]), new StubLoop()); + self::assertSame(0, $command->run(new StringInput('queue'), $this->createMock(OutputInterface::class))); + } + + public function testRejectsMultipleQueuesWithoutAMode(): void + { + $consumer = $this->createMock(QueueConsumerInterface::class); + $consumer->expects($this->never())->method('listen'); + $consumer->expects($this->never())->method('run'); + $command = new ConsumeCommand(new PredefinedQueueProvider([ + 'first' => ['consumer' => $consumer], + 'second' => ['consumer' => $consumer], + ]), new StubLoop()); + $output = new BufferedOutput(); + + $result = $command->run(new StringInput('first second'), $output); + + self::assertSame(Command::INVALID, $result); + self::assertStringContainsString('Use --all or --once to consume multiple queues.', $output->fetch()); + } +} diff --git a/tests/Unit/Command/ListenCommandTest.php b/tests/Unit/Command/ListenCommandTest.php deleted file mode 100644 index fedceca5..00000000 --- a/tests/Unit/Command/ListenCommandTest.php +++ /dev/null @@ -1,23 +0,0 @@ -createMock(QueueConsumerInterface::class); - $consumer->expects($this->once())->method('listen'); - $command = new ListenCommand(new PredefinedQueueProvider(['queue' => ['consumer' => $consumer]])); - self::assertSame(0, $command->run(new StringInput('queue'), $this->createMock(OutputInterface::class))); - } -} diff --git a/tests/Unit/SignalHandlingTest.php b/tests/Unit/SignalHandlingTest.php index 32572709..9c6a6e9b 100644 --- a/tests/Unit/SignalHandlingTest.php +++ b/tests/Unit/SignalHandlingTest.php @@ -11,8 +11,7 @@ use Symfony\Component\Console\Input\ArrayInput; use Symfony\Component\Console\Output\BufferedOutput; use Yiisoft\Queue\Cli\SignalLoop; -use Yiisoft\Queue\Command\ListenAllCommand; -use Yiisoft\Queue\Command\RunCommand; +use Yiisoft\Queue\Command\ConsumeCommand; use Yiisoft\Queue\Message\GenericMessage; use Yiisoft\Queue\Message\IdEnvelope; use Yiisoft\Queue\MessageStatus; @@ -120,10 +119,10 @@ public static function consumerMethodProvider(): iterable } /** - * @param class-string|class-string $commandClass + * @param array $options */ #[DataProvider('commandProvider')] - public function testSignalBetweenQueuesStopsBeforeTheNextMessage(string $commandClass): void + public function testSignalBetweenQueuesStopsBeforeTheNextMessage(array $options): void { $applicationSignals = []; pcntl_signal(SIGTERM, static function (int $signal) use (&$applicationSignals): void { @@ -154,8 +153,8 @@ public function testSignalBetweenQueuesStopsBeforeTheNextMessage(string $command 'first' => ['consumer' => $consumerAtQueueBoundary], 'second' => ['consumer' => new QueueConsumer($secondWorker, $loop, new NullLogger(), $secondAdapter)], ]); - $command = new $commandClass($provider, $loop); - $input = new ArrayInput([], $command->getNativeDefinition()); + $command = new ConsumeCommand($provider, $loop); + $input = new ArrayInput($options, $command->getNativeDefinition()); $result = $command->run($input, new BufferedOutput()); posix_kill(getmypid(), SIGTERM); @@ -167,10 +166,10 @@ public function testSignalBetweenQueuesStopsBeforeTheNextMessage(string $command } /** - * @param class-string|class-string $commandClass + * @param array $options */ #[DataProvider('commandProvider')] - public function testCommandFailureRestoresApplicationSignalHandling(string $commandClass): void + public function testCommandFailureRestoresApplicationSignalHandling(array $options): void { $applicationSignals = []; pcntl_signal(SIGTERM, static function (int $signal) use (&$applicationSignals): void { @@ -183,8 +182,8 @@ public function testCommandFailureRestoresApplicationSignalHandling(string $comm $worker->method('process')->willThrowException($failure); $loop = new SignalLoop(); $consumer = new QueueConsumer($worker, $loop, new NullLogger(), $adapter); - $command = new $commandClass(new PredefinedQueueProvider(['queue' => ['consumer' => $consumer]]), $loop); - $input = new ArrayInput([], $command->getNativeDefinition()); + $command = new ConsumeCommand(new PredefinedQueueProvider(['queue' => ['consumer' => $consumer]]), $loop); + $input = new ArrayInput($options, $command->getNativeDefinition()); try { $command->run($input, new BufferedOutput()); @@ -198,11 +197,11 @@ public function testCommandFailureRestoresApplicationSignalHandling(string $comm } /** - * @return iterable|class-string}> + * @return iterable}> */ public static function commandProvider(): iterable { - yield 'queue:run' => [RunCommand::class]; - yield 'queue:listen-all' => [ListenAllCommand::class]; + yield 'queue:consume --once' => [['--once' => true]]; + yield 'queue:consume --all' => [['--all' => true]]; } } From 9dd20ed95bcb37e8c64a567e16aafefea97a2a7e Mon Sep 17 00:00:00 2001 From: Alexander Makarov Date: Sun, 1 Nov 2026 19:14:44 +0300 Subject: [PATCH 2/4] Cover negative consume pause fallback #350 --- tests/Unit/Command/ConsumeAllCommandTest.php | 16 ++++++++++++++++ 1 file changed, 16 insertions(+) diff --git a/tests/Unit/Command/ConsumeAllCommandTest.php b/tests/Unit/Command/ConsumeAllCommandTest.php index 367c7172..75455af6 100644 --- a/tests/Unit/Command/ConsumeAllCommandTest.php +++ b/tests/Unit/Command/ConsumeAllCommandTest.php @@ -114,6 +114,22 @@ public function testPausesWhenThereAreNoMessages(): void self::assertGreaterThanOrEqual(1.0, (hrtime(true) - $start) / 1_000_000_000, 'An empty queue is polled once per pause.'); } + public function testNegativePauseUsesDefaultIdleDelay(): void + { + $consumer = $this->createMock(QueueConsumerInterface::class); + $consumer->expects($this->once())->method('run')->willReturn(0); + $command = new ConsumeCommand(new PredefinedQueueProvider([ + 'consumer' => ['consumer' => $consumer], + ]), $this->createLoop(2)); + $input = new ArrayInput(['--all' => true, '--pause' => -1], $command->getNativeDefinition()); + + $start = hrtime(true); + $result = $command->run($input, new BufferedOutput()); + + self::assertSame(Command::SUCCESS, $result); + self::assertGreaterThanOrEqual(1.0, (hrtime(true) - $start) / 1_000_000_000); + } + /** * @return LoopInterface A loop which allows the given number of checks and then asks to stop. */ From e6bda3753fab051b4205f1dd33e6edd840d40a4a Mon Sep 17 00:00:00 2001 From: Alexander Makarov Date: Sun, 1 Nov 2026 20:10:36 +0300 Subject: [PATCH 3/4] Remove command migration section #350 --- docs/guide/en/console-commands.md | 13 ------------- 1 file changed, 13 deletions(-) diff --git a/docs/guide/en/console-commands.md b/docs/guide/en/console-commands.md index fc9e04ac..d46d07ae 100644 --- a/docs/guide/en/console-commands.md +++ b/docs/guide/en/console-commands.md @@ -53,16 +53,3 @@ 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, the default `SignalLoop` handles signals such as `SIGTERM`/`SIGINT`. - -## Migrating existing commands - -Update process manager and cron configurations as follows: - -| Previous command | Replacement | -| --- | --- | -| `queue:run` | `queue:consume --once` | -| `queue:listen [queueName]` | `queue:consume [queueName]` | -| `queue:listen-all [queueNames...]` | `queue:consume --all [queueNames...]` | - -The previous command names are no longer registered. For manual registration, replace `RunCommand`, `ListenCommand`, -and `ListenAllCommand` with `ConsumeCommand`, passing the consumer provider and loop to its constructor. From 3957762c3fc8dd71840d2105132e3e7daff835bf Mon Sep 17 00:00:00 2001 From: Alexander Makarov Date: Sun, 1 Nov 2026 20:20:29 +0300 Subject: [PATCH 4/4] Reject ignored consume options and clarify help --- docs/guide/en/console-commands.md | 4 + src/Command/ConsumeCommand.php | 45 +++++-- tests/Unit/Command/ConsumeAllCommandTest.php | 2 +- .../Command/ConsumeCommandOptionsTest.php | 122 ++++++++++++++++++ tests/Unit/Command/ConsumeOnceCommandTest.php | 2 +- 5 files changed, 165 insertions(+), 10 deletions(-) create mode 100644 tests/Unit/Command/ConsumeCommandOptionsTest.php diff --git a/docs/guide/en/console-commands.md b/docs/guide/en/console-commands.md index d46d07ae..a8383f5a 100644 --- a/docs/guide/en/console-commands.md +++ b/docs/guide/en/console-commands.md @@ -36,6 +36,9 @@ 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: @@ -45,6 +48,7 @@ The following command iterates through multiple queues and is meant to be used i - `--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 diff --git a/src/Command/ConsumeCommand.php b/src/Command/ConsumeCommand.php index 8bf4e85d..70e09913 100644 --- a/src/Command/ConsumeCommand.php +++ b/src/Command/ConsumeCommand.php @@ -31,24 +31,39 @@ 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 and exit when queues are empty.') + ->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, - 'Poll all consumer queues, or the specified queues. Recommended for development only.', + '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. 0 means no limit.', - 0, + '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.', ) - ->addOption('pause', 'p', InputOption::VALUE_REQUIRED, 'Seconds between idle passes with --all.', 1) ->addUsage('[queue]') ->addUsage('--once [queue1 [queue2 [...]]] --limit=100') - ->addUsage('--all [queue1 [queue2 [...]]] --pause=1 --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 @@ -57,6 +72,20 @@ protected function execute(InputInterface $input, OutputInterface $output): int $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('--limit requires --once or --all.'); + + return Command::INVALID; + } + + if ((!$all || $once) && $pauseSeconds !== null) { + $output->writeln('--pause requires --all without --once.'); + + return Command::INVALID; + } if (!$once && !$all) { if (count($queueNames) > 1) { @@ -85,8 +114,8 @@ protected function execute(InputInterface $input, OutputInterface $output): int return Command::SUCCESS; } - $limit = (int) $input->getOption('limit'); - $pauseSeconds = (int) $input->getOption('pause'); + $limit = (int) $limit; + $pauseSeconds = (int) ($pauseSeconds ?? 1); if ($pauseSeconds < 0) { $pauseSeconds = 1; } diff --git a/tests/Unit/Command/ConsumeAllCommandTest.php b/tests/Unit/Command/ConsumeAllCommandTest.php index 75455af6..cec4b108 100644 --- a/tests/Unit/Command/ConsumeAllCommandTest.php +++ b/tests/Unit/Command/ConsumeAllCommandTest.php @@ -106,7 +106,7 @@ public function testPausesWhenThereAreNoMessages(): void $command = new ConsumeCommand(new PredefinedQueueProvider([ 'consumer' => ['consumer' => $consumer], ]), $this->createLoop(2)); - $input = new ArrayInput(['--all' => true, '--pause' => 1], $command->getNativeDefinition()); + $input = new ArrayInput(['--all' => true], $command->getNativeDefinition()); $start = hrtime(true); $command->run($input, $this->createMock(OutputInterface::class)); diff --git a/tests/Unit/Command/ConsumeCommandOptionsTest.php b/tests/Unit/Command/ConsumeCommandOptionsTest.php new file mode 100644 index 00000000..76675bbe --- /dev/null +++ b/tests/Unit/Command/ConsumeCommandOptionsTest.php @@ -0,0 +1,122 @@ +createMock(QueueConsumerProviderInterface::class); + $provider->expects($this->never())->method('getConsumer'); + $provider->expects($this->never())->method('getConsumerQueueNames'); + $command = new ConsumeCommand($provider, new StubLoop()); + $input = is_array($arguments) ? new ArrayInput($arguments) : new StringInput($arguments); + $output = new BufferedOutput(); + + $result = $command->run($input, $output); + + self::assertSame(Command::INVALID, $result); + self::assertStringContainsString($message, $output->fetch()); + } + + public static function ignoredOptionsProvider(): iterable + { + foreach (['--limit=100', '-m100', '--limit=0', '-m 0', '--limit=0 -- --all'] as $arguments) { + yield $arguments => [$arguments, '--limit requires --once or --all.']; + } + foreach ([ + '--pause=1', '-p1', '--pause=0', '-p 0', '--pause=1 -- --all', + '--once --pause=1', '--once -p0', '--all --once --pause=0', '--all --once -p 1', + ] as $arguments) { + yield $arguments => [$arguments, '--pause requires --all without --once.']; + } + yield 'array limit zero' => [['--limit' => 0], '--limit requires --once or --all.']; + yield 'array short limit zero' => [['-m' => 0], '--limit requires --once or --all.']; + yield 'array pause default' => [['--pause' => 1], '--pause requires --all without --once.']; + yield 'array short pause zero' => [['-p' => 0], '--pause requires --all without --once.']; + } + + #[DataProvider('queueNamesAfterSeparatorProvider')] + public function testOptionsAfterSeparatorAreQueueNames(string $name): void + { + $consumer = $this->createMock(QueueConsumerInterface::class); + $consumer->expects($this->once())->method('listen'); + $command = new ConsumeCommand(new PredefinedQueueProvider([ + $name => ['consumer' => $consumer], + ]), new StubLoop()); + + $result = $command->run(new StringInput('-- ' . $name), new BufferedOutput()); + + self::assertSame(Command::SUCCESS, $result); + } + + public static function queueNamesAfterSeparatorProvider(): iterable + { + yield ['--limit=100']; + yield ['-m100']; + yield ['--pause=1']; + yield ['-p1']; + } + + public function testRejectsIgnoredLimitInGroupedApplicationOptions(): void + { + $provider = $this->createMock(QueueConsumerProviderInterface::class); + $provider->expects($this->never())->method('getConsumer'); + $application = new Application(); + $application->setAutoExit(false); + $application->add(new ConsumeCommand($provider, new StubLoop())); + + $result = $application->run(new StringInput('queue:consume -vm0'), new BufferedOutput()); + + self::assertSame(Command::INVALID, $result); + } + + public function testCombinedModesAcceptLimitAndPerformOnePass(): void + { + $consumer = $this->createMock(QueueConsumerInterface::class); + $consumer->expects($this->once())->method('run')->with(2)->willReturn(2); + $consumer->expects($this->never())->method('listen'); + $command = new ConsumeCommand(new PredefinedQueueProvider([ + 'queue' => ['consumer' => $consumer], + ]), new StubLoop()); + + $result = $command->run(new StringInput('--all --once -m2'), new BufferedOutput()); + + self::assertSame(Command::SUCCESS, $result); + } + + public function testHelpExplainsQueueDefaultsAndOncePrecedence(): void + { + $application = new Application(); + $application->setAutoExit(false); + $application->add(new ConsumeCommand(new PredefinedQueueProvider([]), new StubLoop())); + $output = new BufferedOutput(); + + $result = $application->run(new StringInput('queue:consume --help'), $output); + + $help = preg_replace('/\s+/', ' ', $output->fetch()); + self::assertSame(Command::SUCCESS, $result); + self::assertStringContainsString('Without queue names, listen to yii-queue.', $help); + self::assertStringContainsString('--once and --all instead use all configured consumer queues.', $help); + self::assertStringContainsString('--once takes precedence over --all.', $help); + } +} diff --git a/tests/Unit/Command/ConsumeOnceCommandTest.php b/tests/Unit/Command/ConsumeOnceCommandTest.php index d6370316..2e41b075 100644 --- a/tests/Unit/Command/ConsumeOnceCommandTest.php +++ b/tests/Unit/Command/ConsumeOnceCommandTest.php @@ -30,7 +30,7 @@ public function testRunsSelectedConsumer(): void public function testDefaultRunSkipsProducerOnlyQueues(): void { $consumer = $this->createMock(QueueConsumerInterface::class); - $consumer->expects($this->once())->method('run')->willReturn(0); + $consumer->expects($this->once())->method('run')->with(0)->willReturn(0); $command = new ConsumeCommand(new PredefinedQueueProvider([ 'producer' => ['producer' => new StubQueueProducer()], DefaultQueue::NAME => ['consumer' => $consumer],