From daface1d36ca4a9fb48121de04af663a5f116b70 Mon Sep 17 00:00:00 2001 From: Alexander Makarov Date: Mon, 20 Jul 2026 12:43:17 +0300 Subject: [PATCH 1/3] Split queue consumer contract --- README.md | 4 ++ docs/guide/en/configuration-manual.md | 9 ++++ docs/guide/en/queue-names-advanced.md | 1 + src/Command/ListenAllCommand.php | 25 +++++++++- src/Command/ListenCommand.php | 25 +++++++++- src/Command/RunCommand.php | 27 +++++++++-- src/Debug/QueueConsumerDecorator.php | 47 +++++++++++++++++++ src/Debug/QueueDecorator.php | 10 ---- src/Debug/QueueProviderInterfaceProxy.php | 5 +- src/Queue.php | 2 +- src/QueueConsumerInterface.php | 20 ++++++++ src/QueueInterface.php | 12 ----- stubs/StubQueue.php | 3 +- tests/Benchmark/QueueBench.php | 3 +- tests/Unit/Command/ListenAllCommandTest.php | 23 ++++++++- tests/Unit/Command/ListenCommandTest.php | 20 ++++++-- tests/Unit/Command/RunCommandTest.php | 25 ++++++++-- tests/Unit/Debug/QueueDecoratorTest.php | 10 ++-- .../Debug/QueueProviderInterfaceProxyTest.php | 16 +++++++ tests/Unit/QueueTest.php | 12 +++++ 20 files changed, 253 insertions(+), 46 deletions(-) create mode 100644 src/Debug/QueueConsumerDecorator.php create mode 100644 src/QueueConsumerInterface.php diff --git a/README.md b/README.md index 942e4155..a06c58c8 100644 --- a/README.md +++ b/README.md @@ -138,6 +138,8 @@ For setting up all classes manually, see the [Manual configuration](docs/guide/e To send a message to the queue, get the queue instance and call `push()`. Typically the queue is injected as a dependency: ```php +use Yiisoft\Queue\QueueInterface; + final readonly class Foo { public function __construct(private QueueInterface $queue) {} @@ -164,6 +166,8 @@ By default, Yii Framework uses [yiisoft/yii-console](https://github.com/yiisoft/ See [Console commands](docs/guide/en/console-commands.md) for more details. +Consumers are represented by `Yiisoft\Queue\QueueConsumerInterface`; it contains `run()` and `listen()`. + > In case you're running the queue in synchronous mode (no adapter), `queue:listen` logs an info message and exits. The messages are processed immediately when pushed. ## Documentation diff --git a/docs/guide/en/configuration-manual.md b/docs/guide/en/configuration-manual.md index 181874e7..acf81055 100644 --- a/docs/guide/en/configuration-manual.md +++ b/docs/guide/en/configuration-manual.md @@ -98,9 +98,15 @@ $provider = new PredefinedQueueProvider([ ## Running the queue +Message consumption methods are available on `Yiisoft\Queue\QueueConsumerInterface`. +The built-in `Queue` implements both `QueueInterface` for producing messages and `QueueConsumerInterface` for consuming them. + ### Processing existing messages ```php +use Yiisoft\Queue\QueueConsumerInterface; + +/** @var QueueConsumerInterface $queue */ $queue->run(); // Process all messages $queue->run(10); // Process up to 10 messages ``` @@ -108,6 +114,9 @@ $queue->run(10); // Process up to 10 messages ### Listening for new messages ```php +use Yiisoft\Queue\QueueConsumerInterface; + +/** @var QueueConsumerInterface $queue */ $queue->listen(); // Run indefinitely ``` diff --git a/docs/guide/en/queue-names-advanced.md b/docs/guide/en/queue-names-advanced.md index 21cdd8cc..8ccc9a0b 100644 --- a/docs/guide/en/queue-names-advanced.md +++ b/docs/guide/en/queue-names-advanced.md @@ -7,6 +7,7 @@ Use this reference when you need to understand how queue names map to adapters, - A queue name (string or `BackedEnum`) is passed to `Yiisoft\Queue\Provider\QueueProviderInterface::get($queueName)`. - The provider returns a `Yiisoft\Queue\QueueInterface` instance configured for that name. - `QueueInterface::getName()` can be used for introspection; it returns the logical name the queue was created with. +- Queues that can consume messages also implement `Yiisoft\Queue\QueueConsumerInterface`. ## Provider implementations diff --git a/src/Command/ListenAllCommand.php b/src/Command/ListenAllCommand.php index 51b55009..b57fbadd 100644 --- a/src/Command/ListenAllCommand.php +++ b/src/Command/ListenAllCommand.php @@ -11,7 +11,12 @@ use Symfony\Component\Console\Input\InputOption; use Symfony\Component\Console\Output\OutputInterface; use Yiisoft\Queue\Cli\LoopInterface; +use Yiisoft\Queue\Provider\InvalidQueueConfigException; use Yiisoft\Queue\Provider\QueueProviderInterface; +use Yiisoft\Queue\QueueConsumerInterface; + +use function get_debug_type; +use function sprintf; #[AsCommand( 'queue:listen-all', @@ -64,7 +69,7 @@ protected function execute(InputInterface $input, OutputInterface $output): int $queues = []; /** @var string $queue */ foreach ($input->getArgument('queue') as $queue) { - $queues[] = $this->queueProvider->get($queue); + $queues[] = $this->getQueueConsumer($queue); } $pauseSeconds = (int) $input->getOption('pause'); @@ -86,4 +91,22 @@ protected function execute(InputInterface $input, OutputInterface $output): int return 0; } + + private function getQueueConsumer(string $name): QueueConsumerInterface + { + $queue = $this->queueProvider->get($name); + + if (!$queue instanceof QueueConsumerInterface) { + throw new InvalidQueueConfigException( + sprintf( + 'Queue "%s" must implement "%s" to consume messages. Got "%s" instead.', + $name, + QueueConsumerInterface::class, + get_debug_type($queue), + ), + ); + } + + return $queue; + } } diff --git a/src/Command/ListenCommand.php b/src/Command/ListenCommand.php index cfc4c9db..c1cdb39c 100644 --- a/src/Command/ListenCommand.php +++ b/src/Command/ListenCommand.php @@ -9,7 +9,12 @@ use Symfony\Component\Console\Input\InputArgument; use Symfony\Component\Console\Input\InputInterface; use Symfony\Component\Console\Output\OutputInterface; +use Yiisoft\Queue\Provider\InvalidQueueConfigException; use Yiisoft\Queue\Provider\QueueProviderInterface; +use Yiisoft\Queue\QueueConsumerInterface; + +use function get_debug_type; +use function sprintf; #[AsCommand( 'queue:listen', @@ -37,8 +42,26 @@ protected function execute(InputInterface $input, OutputInterface $output): int { $queueName = (string) $input->getArgument('queue'); - $this->queueProvider->get($queueName)->listen(); + $this->getQueueConsumer($queueName)->listen(); return Command::SUCCESS; } + + private function getQueueConsumer(string $name): QueueConsumerInterface + { + $queue = $this->queueProvider->get($name); + + if (!$queue instanceof QueueConsumerInterface) { + throw new InvalidQueueConfigException( + sprintf( + 'Queue "%s" must implement "%s" to consume messages. Got "%s" instead.', + $name, + QueueConsumerInterface::class, + get_debug_type($queue), + ), + ); + } + + return $queue; + } } diff --git a/src/Command/RunCommand.php b/src/Command/RunCommand.php index c9c0baa8..a4d9792c 100644 --- a/src/Command/RunCommand.php +++ b/src/Command/RunCommand.php @@ -10,7 +10,12 @@ use Symfony\Component\Console\Input\InputInterface; use Symfony\Component\Console\Input\InputOption; use Symfony\Component\Console\Output\OutputInterface; +use Yiisoft\Queue\Provider\InvalidQueueConfigException; use Yiisoft\Queue\Provider\QueueProviderInterface; +use Yiisoft\Queue\QueueConsumerInterface; + +use function get_debug_type; +use function sprintf; #[AsCommand( 'queue:run', @@ -47,13 +52,29 @@ protected function execute(InputInterface $input, OutputInterface $output): int /** @var string $queue */ foreach ($input->getArgument('queue') as $queue) { $output->write("Processing queue $queue... "); - $count = $this->queueProvider - ->get($queue) - ->run((int) $input->getOption('limit')); + $count = $this->getQueueConsumer($queue)->run((int) $input->getOption('limit')); $output->writeln("Messages processed: $count."); } return 0; } + + private function getQueueConsumer(string $name): QueueConsumerInterface + { + $queue = $this->queueProvider->get($name); + + if (!$queue instanceof QueueConsumerInterface) { + throw new InvalidQueueConfigException( + sprintf( + 'Queue "%s" must implement "%s" to consume messages. Got "%s" instead.', + $name, + QueueConsumerInterface::class, + get_debug_type($queue), + ), + ); + } + + return $queue; + } } diff --git a/src/Debug/QueueConsumerDecorator.php b/src/Debug/QueueConsumerDecorator.php new file mode 100644 index 00000000..b6223ab0 --- /dev/null +++ b/src/Debug/QueueConsumerDecorator.php @@ -0,0 +1,47 @@ +queueDecorator = new QueueDecorator($queue, $collector); + } + + public function status(string|int $id): MessageStatus + { + return $this->queueDecorator->status($id); + } + + public function push(MessageInterface $message): MessageInterface + { + return $this->queueDecorator->push($message); + } + + public function run(int $max = 0): int + { + return $this->queue->run($max); + } + + public function listen(): void + { + $this->queue->listen(); + } + + public function getName(): string + { + return $this->queueDecorator->getName(); + } +} diff --git a/src/Debug/QueueDecorator.php b/src/Debug/QueueDecorator.php index de0e901b..3b3f9617 100644 --- a/src/Debug/QueueDecorator.php +++ b/src/Debug/QueueDecorator.php @@ -36,16 +36,6 @@ public function push(MessageInterface $message): MessageInterface return $message; } - public function run(int $max = 0): int - { - return $this->queue->run($max); - } - - public function listen(): void - { - $this->queue->listen(); - } - public function getName(): string { return $this->queue->getName(); diff --git a/src/Debug/QueueProviderInterfaceProxy.php b/src/Debug/QueueProviderInterfaceProxy.php index c28d1b20..b40a0b56 100644 --- a/src/Debug/QueueProviderInterfaceProxy.php +++ b/src/Debug/QueueProviderInterfaceProxy.php @@ -6,6 +6,7 @@ use BackedEnum; use Yiisoft\Queue\Provider\QueueProviderInterface; +use Yiisoft\Queue\QueueConsumerInterface; use Yiisoft\Queue\QueueInterface; final class QueueProviderInterfaceProxy implements QueueProviderInterface @@ -19,7 +20,9 @@ public function get(string|BackedEnum $name): QueueInterface { $queue = $this->queueProvider->get($name); - return new QueueDecorator($queue, $this->collector); + return $queue instanceof QueueConsumerInterface + ? new QueueConsumerDecorator($queue, $this->collector) + : new QueueDecorator($queue, $this->collector); } public function has(string|BackedEnum $name): bool diff --git a/src/Queue.php b/src/Queue.php index b9bc5d48..bb2c6d5a 100644 --- a/src/Queue.php +++ b/src/Queue.php @@ -17,7 +17,7 @@ use Yiisoft\Queue\Message\IdEnvelope; use Yiisoft\Queue\Provider\QueueProviderInterface; -final class Queue implements QueueInterface +final class Queue implements QueueInterface, QueueConsumerInterface { private string $name; diff --git a/src/QueueConsumerInterface.php b/src/QueueConsumerInterface.php new file mode 100644 index 00000000..3e33d5ed --- /dev/null +++ b/src/QueueConsumerInterface.php @@ -0,0 +1,20 @@ +createMock(QueueInterface::class); + $queue1 = $this->createMockForIntersectionOfInterfaces([QueueInterface::class, QueueConsumerInterface::class]); $queue1->expects($this->once())->method('run'); - $queue2 = $this->createMock(QueueInterface::class); + $queue2 = $this->createMockForIntersectionOfInterfaces([QueueInterface::class, QueueConsumerInterface::class]); $queue2->expects($this->once())->method('run'); $queueFactory = new PredefinedQueueProvider([ @@ -39,4 +41,21 @@ public function testExecute(): void $this->assertEquals(0, $exitCode); } + + public function testExecuteRequiresConsumerQueues(): void + { + $queueFactory = new PredefinedQueueProvider([ + 'producer-only' => $this->createMock(QueueInterface::class), + ]); + $loop = $this->createMock(LoopInterface::class); + + $this->expectException(InvalidQueueConfigException::class); + $this->expectExceptionMessage('Queue "producer-only" must implement'); + + $command = new ListenAllCommand( + $queueFactory, + $loop, + ); + $command->run(new ArrayInput(['queue' => ['producer-only']], $command->getNativeDefinition()), $this->createMock(OutputInterface::class)); + } } diff --git a/tests/Unit/Command/ListenCommandTest.php b/tests/Unit/Command/ListenCommandTest.php index 46dda71f..ccd77210 100644 --- a/tests/Unit/Command/ListenCommandTest.php +++ b/tests/Unit/Command/ListenCommandTest.php @@ -8,14 +8,16 @@ use Symfony\Component\Console\Input\StringInput; use Symfony\Component\Console\Output\OutputInterface; use Yiisoft\Queue\Command\ListenCommand; +use Yiisoft\Queue\Provider\InvalidQueueConfigException; use Yiisoft\Queue\Provider\QueueProviderInterface; +use Yiisoft\Queue\QueueConsumerInterface; use Yiisoft\Queue\QueueInterface; final class ListenCommandTest extends TestCase { public function testExecuteWithDefaultQueue(): void { - $queue = $this->createMock(QueueInterface::class); + $queue = $this->createMockForIntersectionOfInterfaces([QueueInterface::class, QueueConsumerInterface::class]); $queue->expects($this->once()) ->method('listen'); @@ -34,7 +36,7 @@ public function testExecuteWithDefaultQueue(): void public function testExecuteWithCustomQueue(): void { - $queue = $this->createMock(QueueInterface::class); + $queue = $this->createMockForIntersectionOfInterfaces([QueueInterface::class, QueueConsumerInterface::class]); $queue->expects($this->once()) ->method('listen'); @@ -53,7 +55,7 @@ public function testExecuteWithCustomQueue(): void public function testExecuteReturnsZero(): void { - $queue = $this->createMock(QueueInterface::class); + $queue = $this->createMockForIntersectionOfInterfaces([QueueInterface::class, QueueConsumerInterface::class]); $queue->expects($this->once()) ->method('listen'); @@ -66,4 +68,16 @@ public function testExecuteReturnsZero(): void $this->assertSame(0, $exitCode); } + + public function testExecuteRequiresConsumerQueue(): void + { + $queueProvider = $this->createMock(QueueProviderInterface::class); + $queueProvider->method('get')->willReturn($this->createMock(QueueInterface::class)); + + $this->expectException(InvalidQueueConfigException::class); + $this->expectExceptionMessage('Queue "producer-only" must implement'); + + $command = new ListenCommand($queueProvider); + $command->run(new StringInput('producer-only'), $this->createMock(OutputInterface::class)); + } } diff --git a/tests/Unit/Command/RunCommandTest.php b/tests/Unit/Command/RunCommandTest.php index 1ddc02dc..7e7abbc4 100644 --- a/tests/Unit/Command/RunCommandTest.php +++ b/tests/Unit/Command/RunCommandTest.php @@ -8,15 +8,17 @@ use Symfony\Component\Console\Input\StringInput; use Symfony\Component\Console\Output\OutputInterface; use Yiisoft\Queue\Command\RunCommand; +use Yiisoft\Queue\Provider\InvalidQueueConfigException; use Yiisoft\Queue\Provider\PredefinedQueueProvider; use Yiisoft\Queue\Provider\QueueProviderInterface; +use Yiisoft\Queue\QueueConsumerInterface; use Yiisoft\Queue\QueueInterface; final class RunCommandTest extends TestCase { public function testExecuteWithSingleQueue(): void { - $queue = $this->createMock(QueueInterface::class); + $queue = $this->createMockForIntersectionOfInterfaces([QueueInterface::class, QueueConsumerInterface::class]); $queue->expects($this->once()) ->method('run') ->with($this->equalTo(0)) @@ -43,13 +45,13 @@ public function testExecuteWithSingleQueue(): void public function testExecuteWithMultipleQueues(): void { - $queue1 = $this->createMock(QueueInterface::class); + $queue1 = $this->createMockForIntersectionOfInterfaces([QueueInterface::class, QueueConsumerInterface::class]); $queue1->expects($this->once()) ->method('run') ->with($this->equalTo(0)) ->willReturn(3); - $queue2 = $this->createMock(QueueInterface::class); + $queue2 = $this->createMockForIntersectionOfInterfaces([QueueInterface::class, QueueConsumerInterface::class]); $queue2->expects($this->once()) ->method('run') ->with($this->equalTo(0)) @@ -75,7 +77,7 @@ public function testExecuteWithMultipleQueues(): void public function testExecuteWithLimitOption(): void { - $queue = $this->createMock(QueueInterface::class); + $queue = $this->createMockForIntersectionOfInterfaces([QueueInterface::class, QueueConsumerInterface::class]); $queue->expects($this->once()) ->method('run') ->with($this->equalTo(100)) @@ -102,7 +104,7 @@ public function testExecuteWithLimitOption(): void public function testExecuteWithDefaultQueues(): void { - $queue = $this->createMock(QueueInterface::class); + $queue = $this->createMockForIntersectionOfInterfaces([QueueInterface::class, QueueConsumerInterface::class]); $queue->expects($this->once()) ->method('run') ->with($this->equalTo(0)) @@ -126,4 +128,17 @@ public function testExecuteWithDefaultQueues(): void $this->assertEquals(0, $exitCode); } + + public function testExecuteRequiresConsumerQueue(): void + { + $queueProvider = new PredefinedQueueProvider([ + 'producer-only' => $this->createMock(QueueInterface::class), + ]); + + $this->expectException(InvalidQueueConfigException::class); + $this->expectExceptionMessage('Queue "producer-only" must implement'); + + $command = new RunCommand($queueProvider); + $command->run(new StringInput('producer-only'), $this->createMock(OutputInterface::class)); + } } diff --git a/tests/Unit/Debug/QueueDecoratorTest.php b/tests/Unit/Debug/QueueDecoratorTest.php index 394f4cb7..b095885f 100644 --- a/tests/Unit/Debug/QueueDecoratorTest.php +++ b/tests/Unit/Debug/QueueDecoratorTest.php @@ -6,9 +6,11 @@ use PHPUnit\Framework\TestCase; use Yiisoft\Queue\Debug\QueueCollector; +use Yiisoft\Queue\Debug\QueueConsumerDecorator; use Yiisoft\Queue\Debug\QueueDecorator; use Yiisoft\Queue\MessageStatus; use Yiisoft\Queue\Message\MessageInterface; +use Yiisoft\Queue\QueueConsumerInterface; use Yiisoft\Queue\QueueInterface; final class QueueDecoratorTest extends TestCase @@ -88,10 +90,10 @@ public function testStatusCollectsCallLocation(): void public function testRun(): void { - $queue = $this->createMock(QueueInterface::class); + $queue = $this->createMockForIntersectionOfInterfaces([QueueInterface::class, QueueConsumerInterface::class]); $queue->expects($this->once())->method('run'); $collector = new QueueCollector(); - $decorator = new QueueDecorator( + $decorator = new QueueConsumerDecorator( $queue, $collector, ); @@ -101,10 +103,10 @@ public function testRun(): void public function testListen(): void { - $queue = $this->createMock(QueueInterface::class); + $queue = $this->createMockForIntersectionOfInterfaces([QueueInterface::class, QueueConsumerInterface::class]); $queue->expects($this->once())->method('listen'); $collector = new QueueCollector(); - $decorator = new QueueDecorator( + $decorator = new QueueConsumerDecorator( $queue, $collector, ); diff --git a/tests/Unit/Debug/QueueProviderInterfaceProxyTest.php b/tests/Unit/Debug/QueueProviderInterfaceProxyTest.php index 2ba2ae42..24915332 100644 --- a/tests/Unit/Debug/QueueProviderInterfaceProxyTest.php +++ b/tests/Unit/Debug/QueueProviderInterfaceProxyTest.php @@ -6,9 +6,11 @@ use PHPUnit\Framework\TestCase; use Yiisoft\Queue\Debug\QueueCollector; +use Yiisoft\Queue\Debug\QueueConsumerDecorator; use Yiisoft\Queue\Debug\QueueDecorator; use Yiisoft\Queue\Debug\QueueProviderInterfaceProxy; use Yiisoft\Queue\Provider\QueueProviderInterface; +use Yiisoft\Queue\QueueConsumerInterface; use Yiisoft\Queue\QueueInterface; final class QueueProviderInterfaceProxyTest extends TestCase @@ -24,6 +26,20 @@ public function testGet(): void $this->assertInstanceOf(QueueDecorator::class, $factory->get('test')); } + public function testGetConsumerQueue(): void + { + $queueFactory = $this->createMock(QueueProviderInterface::class); + $queue = $this->createMockForIntersectionOfInterfaces([QueueInterface::class, QueueConsumerInterface::class]); + $queueFactory->expects($this->once())->method('get')->willReturn($queue); + $collector = new QueueCollector(); + $factory = new QueueProviderInterfaceProxy($queueFactory, $collector); + + $decoratedQueue = $factory->get('test'); + + $this->assertInstanceOf(QueueConsumerDecorator::class, $decoratedQueue); + $this->assertInstanceOf(QueueConsumerInterface::class, $decoratedQueue); + } + public function testHas(): void { $queueFactory = $this->createMock(QueueProviderInterface::class); diff --git a/tests/Unit/QueueTest.php b/tests/Unit/QueueTest.php index 254c8076..eecfc8a3 100644 --- a/tests/Unit/QueueTest.php +++ b/tests/Unit/QueueTest.php @@ -8,6 +8,8 @@ use Yiisoft\Queue\Message\IdEnvelope; use Yiisoft\Queue\Message\GenericMessage; use Yiisoft\Queue\MessageStatus; +use Yiisoft\Queue\QueueConsumerInterface; +use Yiisoft\Queue\QueueInterface; use Yiisoft\Queue\Stubs\InMemoryAdapter; use Yiisoft\Queue\Tests\TestCase; @@ -22,6 +24,16 @@ enum TestQueue: string final class QueueTest extends TestCase { + public function testContracts(): void + { + $queue = $this->createQueue(); + + self::assertInstanceOf(QueueInterface::class, $queue); + self::assertInstanceOf(QueueConsumerInterface::class, $queue); + self::assertFalse(method_exists(QueueInterface::class, 'run')); + self::assertFalse(method_exists(QueueInterface::class, 'listen')); + } + public function testPushSuccessful(): void { $adapter = new InMemoryAdapter(); From 5b408415fb8a557a3450e4936be4503151682afc Mon Sep 17 00:00:00 2001 From: Alexander Makarov Date: Mon, 20 Jul 2026 12:47:02 +0300 Subject: [PATCH 2/3] Document queue interface roles --- src/QueueConsumerInterface.php | 6 ++++++ src/QueueInterface.php | 5 +++++ 2 files changed, 11 insertions(+) diff --git a/src/QueueConsumerInterface.php b/src/QueueConsumerInterface.php index 3e33d5ed..320f1b77 100644 --- a/src/QueueConsumerInterface.php +++ b/src/QueueConsumerInterface.php @@ -4,6 +4,12 @@ namespace Yiisoft\Queue; +/** + * Consumes messages from a queue. + * + * This interface complements {@see QueueInterface}, which is responsible for producing messages and checking their + * status. + */ interface QueueConsumerInterface { /** diff --git a/src/QueueInterface.php b/src/QueueInterface.php index 0aa223d8..66a70b39 100644 --- a/src/QueueInterface.php +++ b/src/QueueInterface.php @@ -6,6 +6,11 @@ use Yiisoft\Queue\Message\MessageInterface; +/** + * Produces messages by pushing them into a queue. + * + * For consuming messages from a queue, use {@see QueueConsumerInterface}. + */ interface QueueInterface { /** From b27031d73b01011a7384cddf23e08574048971f5 Mon Sep 17 00:00:00 2001 From: Alexander Makarov Date: Mon, 20 Jul 2026 16:33:38 +0300 Subject: [PATCH 3/3] Address queue consumer review comments --- README.md | 2 +- docs/guide/en/console-commands.md | 6 ++--- src/Command/ListenAllCommand.php | 28 +++++++++++++++++---- src/Command/RunCommand.php | 24 +++++++++++++++--- tests/Benchmark/QueueBench.php | 4 ++- tests/Unit/Command/ListenAllCommandTest.php | 24 ++++++++++++++++++ tests/Unit/Command/RunCommandTest.php | 28 +++++++++++++++++++++ 7 files changed, 102 insertions(+), 14 deletions(-) diff --git a/README.md b/README.md index a06c58c8..2997080c 100644 --- a/README.md +++ b/README.md @@ -161,7 +161,7 @@ By default, Yii Framework uses [yiisoft/yii-console](https://github.com/yiisoft/ ```bash ./yii queue:run # Handle all existing messages in the queue ./yii queue:listen [queueName] # Start a daemon listening for new messages permanently from the specified queue -./yii queue:listen-all [queueName [queueName2 [...]]] # Start a daemon listening for new messages permanently from all queues or specified list of queues (use with caution in production, recommended for dev only) +./yii queue:listen-all [queueName [queueName2 [...]]] # Start a daemon listening for new messages permanently from all consumer-capable queues or specified list of queues (use with caution in production, recommended for dev only) ``` See [Console commands](docs/guide/en/console-commands.md) for more details. diff --git a/docs/guide/en/console-commands.md b/docs/guide/en/console-commands.md index 0157e7f5..c3f5443e 100644 --- a/docs/guide/en/console-commands.md +++ b/docs/guide/en/console-commands.md @@ -6,7 +6,7 @@ If you are using [yiisoft/config](https://github.com/yiisoft/config) and [yiisof If you are using [symfony/console](https://github.com/symfony/console) directly, you should register the commands manually. -> **Note:** The default queue name list (used when no queue names are passed to a command) is only available when using [yiisoft/config](https://github.com/yiisoft/config) and [yiisoft/yii-console](https://github.com/yiisoft/yii-console). Without them, you must pass the queue name list explicitly to the command constructor. +> **Note:** When no queue names are passed to `queue:run` or `queue:listen-all`, the commands use the queue names provided by `QueueProviderInterface::getNames()` and skip queues that don't implement `QueueConsumerInterface`. Explicitly passed queue names must be consumer-capable. In [yiisoft/app](https://github.com/yiisoft/app) the `yii` console binary is provided out of the box. If you are using [yiisoft/yii-console](https://github.com/yiisoft/yii-console) or `symfony/console` without that template, invoke these commands the same way you invoke other console commands in your application. @@ -17,7 +17,7 @@ The command `queue:run` obtains and handles messages until the queue is empty, t You can also narrow the scope of processed messages by specifying queue name(s) and maximum number of messages to process: -- Specify one or more queue names to process. Messages from other queues will be ignored. Defaults to all registered queue names. +- Specify one or more queue names to process. Messages from other queues will be ignored. Defaults to all registered consumer-capable queue names. - Use `--limit` to limit the number of messages processed. When set, command will exit either when all the messages are processed or when the maximum count is reached. The full command signature is: @@ -39,7 +39,7 @@ yii queue:listen [queueName] The following command iterates through multiple queues and is meant to be used in development environment only, as it consumes a lot of CPU for iterating through queues. You can pass to it: -- `queueName` argument(s). Specify one or more queue names to process. Messages from other queues will be ignored. Defaults to all registered queue names. +- `queueName` argument(s). Specify one or more queue names to process. Messages from other queues will be ignored. Defaults to all registered consumer-capable queue names. - `--limit` option to limit the number of messages processed before switching to another queue. E.g. you set `--limit` to 500 and right now you have 1000 messages in `queue1`. This command will consume only 500 of them, then it will switch to `queue2` to see if there are any messages there. Defaults to `0` (no limit). - `--pause` option to specify the number of seconds to pause between checking queues when no messages are found. Defaults to `1`. diff --git a/src/Command/ListenAllCommand.php b/src/Command/ListenAllCommand.php index b57fbadd..e9904a67 100644 --- a/src/Command/ListenAllCommand.php +++ b/src/Command/ListenAllCommand.php @@ -22,7 +22,7 @@ 'queue:listen-all', 'Listens the all the given queues and executes messages as they come. ' . 'Meant to be used in development environment only. ' - . 'Listens all configured queues by default in case you\'re using yiisoft/config. ' + . 'Listens all consumer-capable configured queues by default. ' . 'Needs to be stopped manually.', )] final class ListenAllCommand extends Command @@ -43,7 +43,7 @@ public function configure(): void 'queue', InputArgument::OPTIONAL | InputArgument::IS_ARRAY, 'Queue name list to connect to', - $this->queueProvider->getNames(), + [], ) ->addOption( 'pause', @@ -66,10 +66,24 @@ public function configure(): void protected function execute(InputInterface $input, OutputInterface $output): int { + /** @var string[] $queueNames */ + $queueNames = $input->getArgument('queue'); + $queueIsRequired = $queueNames !== []; + if (!$queueIsRequired) { + $queueNames = $this->queueProvider->getNames(); + } + $queues = []; /** @var string $queue */ - foreach ($input->getArgument('queue') as $queue) { - $queues[] = $this->getQueueConsumer($queue); + foreach ($queueNames as $queue) { + $queueConsumer = $this->getQueueConsumer($queue, $queueIsRequired); + if ($queueConsumer !== null) { + $queues[] = $queueConsumer; + } + } + + if ($queues === []) { + return Command::SUCCESS; } $pauseSeconds = (int) $input->getOption('pause'); @@ -92,11 +106,15 @@ protected function execute(InputInterface $input, OutputInterface $output): int return 0; } - private function getQueueConsumer(string $name): QueueConsumerInterface + private function getQueueConsumer(string $name, bool $required): ?QueueConsumerInterface { $queue = $this->queueProvider->get($name); if (!$queue instanceof QueueConsumerInterface) { + if (!$required) { + return null; + } + throw new InvalidQueueConfigException( sprintf( 'Queue "%s" must implement "%s" to consume messages. Got "%s" instead.', diff --git a/src/Command/RunCommand.php b/src/Command/RunCommand.php index a4d9792c..09747ad8 100644 --- a/src/Command/RunCommand.php +++ b/src/Command/RunCommand.php @@ -35,7 +35,7 @@ public function configure(): void 'queue', InputArgument::OPTIONAL | InputArgument::IS_ARRAY, 'Queue name list to connect to.', - $this->queueProvider->getNames(), + [], ) ->addOption( 'limit', @@ -49,10 +49,22 @@ public function configure(): void protected function execute(InputInterface $input, OutputInterface $output): int { + /** @var string[] $queueNames */ + $queueNames = $input->getArgument('queue'); + $queueIsRequired = $queueNames !== []; + if (!$queueIsRequired) { + $queueNames = $this->queueProvider->getNames(); + } + /** @var string $queue */ - foreach ($input->getArgument('queue') as $queue) { + foreach ($queueNames as $queue) { + $queueConsumer = $this->getQueueConsumer($queue, $queueIsRequired); + if ($queueConsumer === null) { + continue; + } + $output->write("Processing queue $queue... "); - $count = $this->getQueueConsumer($queue)->run((int) $input->getOption('limit')); + $count = $queueConsumer->run((int) $input->getOption('limit')); $output->writeln("Messages processed: $count."); } @@ -60,11 +72,15 @@ protected function execute(InputInterface $input, OutputInterface $output): int return 0; } - private function getQueueConsumer(string $name): QueueConsumerInterface + private function getQueueConsumer(string $name, bool $required): ?QueueConsumerInterface { $queue = $this->queueProvider->get($name); if (!$queue instanceof QueueConsumerInterface) { + if (!$required) { + return null; + } + throw new InvalidQueueConfigException( sprintf( 'Queue "%s" must implement "%s" to consume messages. Got "%s" instead.', diff --git a/tests/Benchmark/QueueBench.php b/tests/Benchmark/QueueBench.php index e5a5f550..24ca03bf 100644 --- a/tests/Benchmark/QueueBench.php +++ b/tests/Benchmark/QueueBench.php @@ -22,13 +22,15 @@ use Yiisoft\Queue\Middleware\Push\PushMiddlewareConfig; use Yiisoft\Queue\Middleware\Push\PushMiddlewareFactory; use Yiisoft\Queue\Queue; +use Yiisoft\Queue\QueueConsumerInterface; +use Yiisoft\Queue\QueueInterface; use Yiisoft\Queue\Tests\Benchmark\Support\VoidAdapter; use Yiisoft\Queue\Worker\Worker; use Yiisoft\Test\Support\Container\SimpleContainer; final class QueueBench { - private readonly Queue $queue; + private readonly QueueInterface&QueueConsumerInterface $queue; private readonly MessageSerializer $serializer; private readonly VoidAdapter $adapter; diff --git a/tests/Unit/Command/ListenAllCommandTest.php b/tests/Unit/Command/ListenAllCommandTest.php index 3c51c62c..98224873 100644 --- a/tests/Unit/Command/ListenAllCommandTest.php +++ b/tests/Unit/Command/ListenAllCommandTest.php @@ -42,6 +42,30 @@ public function testExecute(): void $this->assertEquals(0, $exitCode); } + public function testExecuteSkipsProducerOnlyQueuesByDefault(): void + { + $queue = $this->createMockForIntersectionOfInterfaces([QueueInterface::class, QueueConsumerInterface::class]); + $queue->expects($this->once())->method('run'); + + $queueFactory = new PredefinedQueueProvider([ + 'producer-only' => $this->createMock(QueueInterface::class), + 'consumer' => $queue, + ]); + + $loop = $this->createMock(LoopInterface::class); + $loop->method('canContinue')->willReturn(true, false); + + $command = new ListenAllCommand( + $queueFactory, + $loop, + ); + $input = new ArrayInput([], $command->getNativeDefinition()); + $input->setOption('pause', 0); + $exitCode = $command->run($input, $this->createMock(OutputInterface::class)); + + $this->assertEquals(0, $exitCode); + } + public function testExecuteRequiresConsumerQueues(): void { $queueFactory = new PredefinedQueueProvider([ diff --git a/tests/Unit/Command/RunCommandTest.php b/tests/Unit/Command/RunCommandTest.php index 7e7abbc4..e2fe4625 100644 --- a/tests/Unit/Command/RunCommandTest.php +++ b/tests/Unit/Command/RunCommandTest.php @@ -129,6 +129,34 @@ public function testExecuteWithDefaultQueues(): void $this->assertEquals(0, $exitCode); } + public function testExecuteWithDefaultQueuesSkipsProducerOnlyQueues(): void + { + $queue = $this->createMockForIntersectionOfInterfaces([QueueInterface::class, QueueConsumerInterface::class]); + $queue->expects($this->once()) + ->method('run') + ->with($this->equalTo(0)) + ->willReturn(2); + + $queueProvider = new PredefinedQueueProvider([ + 'producer-only' => $this->createMock(QueueInterface::class), + 'consumer' => $queue, + ]); + + $input = new StringInput(''); + $output = $this->createMock(OutputInterface::class); + $output->expects($this->once()) + ->method('write') + ->with($this->equalTo('Processing queue consumer... ')); + $output->expects($this->once()) + ->method('writeln') + ->with($this->equalTo('Messages processed: 2.')); + + $command = new RunCommand($queueProvider); + $exitCode = $command->run($input, $output); + + $this->assertEquals(0, $exitCode); + } + public function testExecuteRequiresConsumerQueue(): void { $queueProvider = new PredefinedQueueProvider([