diff --git a/CHANGELOG.md b/CHANGELOG.md
index 9999835c..0e4da67d 100644
--- a/CHANGELOG.md
+++ b/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.
\ No newline at end of file
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..a8383f5a 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,11 +31,14 @@ 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.
+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,
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/ConsumeCommand.php b/src/Command/ConsumeCommand.php
new file mode 100644
index 00000000..70e09913
--- /dev/null
+++ b/src/Command/ConsumeCommand.php
@@ -0,0 +1,160 @@
+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('--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) {
+ $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 = [];
+ 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;
+ }
+}
diff --git a/src/Command/ListenAllCommand.php b/src/Command/ListenAllCommand.php
deleted file mode 100644
index f3bdde4d..00000000
--- a/src/Command/ListenAllCommand.php
+++ /dev/null
@@ -1,114 +0,0 @@
-addArgument(
- 'queue',
- InputArgument::OPTIONAL | InputArgument::IS_ARRAY,
- 'Queue name list to connect to',
- [],
- )
- ->addOption(
- 'pause',
- 'p',
- InputOption::VALUE_REQUIRED,
- 'Pause between queue iterations in seconds. May save some CPU. Default: 1',
- 1,
- )
- ->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).',
- 0,
- );
-
- $this->addUsage('[queue1 [queue2 [...]]] [--pause=] [--limit=]');
- }
-
- protected function execute(InputInterface $input, OutputInterface $output): int
- {
- /** @var string[] $queueNames */
- $queueNames = $input->getArgument('queue');
- if ($queueNames === []) {
- $queueNames = $this->queueProvider->getConsumerQueueNames();
- }
-
- $consumers = [];
- /** @var string $queueName */
- foreach ($queueNames as $queueName) {
- $consumers[] = $this->queueProvider->getConsumer($queueName);
- }
-
- if ($consumers === []) {
- $output->writeln('No consumers are configured.');
-
- return Command::SUCCESS;
- }
-
- $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()) {
- $hasMessages = false;
- foreach ($consumers as $consumer) {
- $hasMessages = $consumer->run($limit) > 0 || $hasMessages;
- }
-
- 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;
- }
-}
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/ConsumeAllCommandTest.php b/tests/Unit/Command/ConsumeAllCommandTest.php
new file mode 100644
index 00000000..cec4b108
--- /dev/null
+++ b/tests/Unit/Command/ConsumeAllCommandTest.php
@@ -0,0 +1,148 @@
+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 ConsumeCommand(new PredefinedQueueProvider([]), $this->createMock(LoopInterface::class));
+ $output = new BufferedOutput();
+
+ self::assertSame(Command::SUCCESS, $command->run(new ArrayInput(['--all' => true], $command->getNativeDefinition()), $output));
+ self::assertSame("No consumers are configured.\n", $output->fetch());
+ }
+
+ public function testRunsOnlyConsumerRolesByDefault(): void
+ {
+ $consumer = $this->createMock(QueueConsumerInterface::class);
+ $consumer->expects($this->once())->method('run')->willReturn(0);
+ $command = new ConsumeCommand(new PredefinedQueueProvider([
+ 'producer' => ['producer' => new StubQueueProducer()],
+ 'consumer' => ['consumer' => $consumer],
+ ]), $this->createLoop(1));
+ $input = new ArrayInput(['--all' => true, '--pause' => 0], $command->getNativeDefinition());
+ self::assertSame(Command::SUCCESS, $command->run($input, $this->createMock(OutputInterface::class)));
+ }
+
+ public function testDoesNotPauseWhenTheLoopCannotContinue(): void
+ {
+ $consumer = $this->createMock(QueueConsumerInterface::class);
+ $consumer->method('run')->willReturn(0);
+ $command = new ConsumeCommand(new PredefinedQueueProvider([
+ 'consumer' => ['consumer' => $consumer],
+ ]), $this->createLoop(1));
+ $input = new ArrayInput(['--all' => true, '--pause' => 3], $command->getNativeDefinition());
+
+ $start = hrtime(true);
+ $command->run($input, $this->createMock(OutputInterface::class));
+
+ self::assertLessThan(1.0, (hrtime(true) - $start) / 1_000_000_000, 'The pause is skipped once the loop asked to stop.');
+ }
+
+ public function testPausesWhenThereAreNoMessages(): void
+ {
+ $consumer = $this->createMock(QueueConsumerInterface::class);
+ $consumer->method('run')->willReturn(0);
+ $command = new ConsumeCommand(new PredefinedQueueProvider([
+ 'consumer' => ['consumer' => $consumer],
+ ]), $this->createLoop(2));
+ $input = new ArrayInput(['--all' => true], $command->getNativeDefinition());
+
+ $start = hrtime(true);
+ $command->run($input, $this->createMock(OutputInterface::class));
+
+ 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.
+ */
+ private function createLoop(int $continueTimes): LoopInterface
+ {
+ $loop = $this->createMock(LoopInterface::class);
+ $loop->method('canContinue')->willReturnCallback(static function () use (&$continueTimes): bool {
+ $canContinue = $continueTimes > 0;
+ $continueTimes--;
+
+ return $canContinue;
+ });
+
+ return $loop;
+ }
+}
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/RunCommandTest.php b/tests/Unit/Command/ConsumeOnceCommandTest.php
similarity index 63%
rename from tests/Unit/Command/RunCommandTest.php
rename to tests/Unit/Command/ConsumeOnceCommandTest.php
index 133cdd83..2e41b075 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([
+ $consumer->expects($this->once())->method('run')->with(0)->willReturn(0);
+ $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/ListenAllCommandTest.php b/tests/Unit/Command/ListenAllCommandTest.php
deleted file mode 100644
index a2d6b214..00000000
--- a/tests/Unit/Command/ListenAllCommandTest.php
+++ /dev/null
@@ -1,86 +0,0 @@
-createMock(LoopInterface::class));
- $output = new BufferedOutput();
-
- self::assertSame(Command::SUCCESS, $command->run(new ArrayInput([], $command->getNativeDefinition()), $output));
- self::assertSame("No consumers are configured.\n", $output->fetch());
- }
-
- public function testRunsOnlyConsumerRolesByDefault(): void
- {
- $consumer = $this->createMock(QueueConsumerInterface::class);
- $consumer->expects($this->once())->method('run')->willReturn(0);
- $command = new ListenAllCommand(new PredefinedQueueProvider([
- 'producer' => ['producer' => new StubQueueProducer()],
- 'consumer' => ['consumer' => $consumer],
- ]), $this->createLoop(1));
- $input = new ArrayInput(['--pause' => 0], $command->getNativeDefinition());
- self::assertSame(Command::SUCCESS, $command->run($input, $this->createMock(OutputInterface::class)));
- }
-
- public function testDoesNotPauseWhenTheLoopCannotContinue(): void
- {
- $consumer = $this->createMock(QueueConsumerInterface::class);
- $consumer->method('run')->willReturn(0);
- $command = new ListenAllCommand(new PredefinedQueueProvider([
- 'consumer' => ['consumer' => $consumer],
- ]), $this->createLoop(1));
- $input = new ArrayInput(['--pause' => 3], $command->getNativeDefinition());
-
- $start = hrtime(true);
- $command->run($input, $this->createMock(OutputInterface::class));
-
- self::assertLessThan(1.0, (hrtime(true) - $start) / 1_000_000_000, 'The pause is skipped once the loop asked to stop.');
- }
-
- public function testPausesWhenThereAreNoMessages(): void
- {
- $consumer = $this->createMock(QueueConsumerInterface::class);
- $consumer->method('run')->willReturn(0);
- $command = new ListenAllCommand(new PredefinedQueueProvider([
- 'consumer' => ['consumer' => $consumer],
- ]), $this->createLoop(2));
- $input = new ArrayInput(['--pause' => 1], $command->getNativeDefinition());
-
- $start = hrtime(true);
- $command->run($input, $this->createMock(OutputInterface::class));
-
- self::assertGreaterThanOrEqual(1.0, (hrtime(true) - $start) / 1_000_000_000, 'An empty queue is polled once per pause.');
- }
-
- /**
- * @return LoopInterface A loop which allows the given number of checks and then asks to stop.
- */
- private function createLoop(int $continueTimes): LoopInterface
- {
- $loop = $this->createMock(LoopInterface::class);
- $loop->method('canContinue')->willReturnCallback(static function () use (&$continueTimes): bool {
- $canContinue = $continueTimes > 0;
- $continueTimes--;
-
- return $canContinue;
- });
-
- return $loop;
- }
-}
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]];
}
}