Skip to content

Commit 9d9a9d8

Browse files
jbdelhommeaufabpot
jbdelhommeau
authored and
fabpot
committed
chore: add exclude-receivers consume parameters
1 parent 0f1cbf3 commit 9d9a9d8

File tree

3 files changed

+187
-8
lines changed

3 files changed

+187
-8
lines changed

‎src/Symfony/Component/Messenger/CHANGELOG.md‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -4,6 +4,7 @@ CHANGELOG
44
7.4
55
---
66

7+
* Add `--exclude-receivers` option to the `messenger:consume command`
78
* Allow any `ServiceResetterInterface` implementation in `ResetServicesListener`
89

910
7.3

‎src/Symfony/Component/Messenger/Command/ConsumeMessagesCommand.php‎

Lines changed: 34 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -77,6 +77,7 @@ protected function configure(): void
7777
new InputOption('queues', null, InputOption::VALUE_REQUIRED | InputOption::VALUE_IS_ARRAY, 'Limit receivers to only consume from the specified queues'),
7878
new InputOption('no-reset', null, InputOption::VALUE_NONE, 'Do not reset container services after each message'),
7979
new InputOption('all', null, InputOption::VALUE_NONE, 'Consume messages from all receivers'),
80+
new InputOption('exclude-receivers', null, InputOption::VALUE_REQUIRED | InputOption::VALUE_IS_ARRAY, 'Exclude specific receivers/transports from consumption (can only be used with --all)'),
8081
new InputOption('keepalive', null, InputOption::VALUE_OPTIONAL, 'Whether to use the transport\'s keepalive mechanism if implemented', self::DEFAULT_KEEPALIVE_INTERVAL),
8182
])
8283
->setHelp(<<<'EOF'
@@ -88,40 +89,44 @@ protected function configure(): void
8889
8990
<info>php %command.full_name% receiver1 receiver2</info>
9091
91-
Use the --limit option to limit the number of messages received:
92+
Use the <info>--limit</info> option to limit the number of messages received:
9293
9394
<info>php %command.full_name% <receiver-name> --limit=10</info>
9495
95-
Use the --failure-limit option to stop the worker when the given number of failed messages is reached:
96+
Use the <info>--failure-limit</info> option to stop the worker when the given number of failed messages is reached:
9697
9798
<info>php %command.full_name% <receiver-name> --failure-limit=2</info>
9899
99-
Use the --memory-limit option to stop the worker if it exceeds a given memory usage limit. You can use shorthand byte values [K, M or G]:
100+
Use the <info>--memory-limit</info> option to stop the worker if it exceeds a given memory usage limit. You can use shorthand byte values [K, M or G]:
100101
101102
<info>php %command.full_name% <receiver-name> --memory-limit=128M</info>
102103
103-
Use the --time-limit option to stop the worker when the given time limit (in seconds) is reached.
104+
Use the <info>--time-limit</info> option to stop the worker when the given time limit (in seconds) is reached.
104105
If a message is being handled, the worker will stop after the processing is finished:
105106
106107
<info>php %command.full_name% <receiver-name> --time-limit=3600</info>
107108
108-
Use the --bus option to specify the message bus to dispatch received messages
109+
Use the <info>--bus</info> option to specify the message bus to dispatch received messages
109110
to instead of trying to determine it automatically. This is required if the
110111
messages didn't originate from Messenger:
111112
112113
<info>php %command.full_name% <receiver-name> --bus=event_bus</info>
113114
114-
Use the --queues option to limit a receiver to only certain queues (only supported by some receivers):
115+
Use the <info>--queues</info> option to limit a receiver to only certain queues (only supported by some receivers):
115116
116117
<info>php %command.full_name% <receiver-name> --queues=fasttrack</info>
117118
118-
Use the --no-reset option to prevent services resetting after each message (may lead to leaking services' state between messages):
119+
Use the <info>--no-reset</info> option to prevent services resetting after each message (may lead to leaking services' state between messages):
119120
120121
<info>php %command.full_name% <receiver-name> --no-reset</info>
121122
122-
Use the --all option to consume from all receivers:
123+
Use the <info>--all</info> option to consume from all receivers:
123124
124125
<info>php %command.full_name% --all</info>
126+
127+
Use the <info>--exclude-receivers</info> option to exclude specific receivers/transports from consumption (can only be used with <info>--all</info>):
128+
129+
<info>php %command.full_name% --all --exclude-receivers=<receiver-name></info>
125130
EOF
126131
)
127132
;
@@ -132,6 +137,10 @@ protected function initialize(InputInterface $input, OutputInterface $output): v
132137
if ($input->hasParameterOption('--keepalive')) {
133138
$this->getApplication()->setAlarmInterval((int) ($input->getOption('keepalive') ?? self::DEFAULT_KEEPALIVE_INTERVAL));
134139
}
140+
141+
if ($input->getOption('exclude-receivers') && !$input->getOption('all')) {
142+
throw new InvalidOptionException('The "--exclude-receivers" option can only be used with the "--all" option.');
143+
}
135144
}
136145

137146
protected function interact(InputInterface $input, OutputInterface $output): void
@@ -169,9 +178,22 @@ protected function interact(InputInterface $input, OutputInterface $output): voi
169178

170179
protected function execute(InputInterface $input, OutputInterface $output): int
171180
{
181+
if ($input->getOption('exclude-receivers') && !$input->getOption('all')) {
182+
throw new InvalidOptionException('The "--exclude-receivers" option can only be used with the "--all" option.');
183+
}
184+
172185
$receivers = [];
173186
$rateLimiters = [];
174187
$receiverNames = $input->getOption('all') ? $this->receiverNames : $input->getArgument('receivers');
188+
189+
if ($input->getOption('all') && $excludedTransports = $input->getOption('exclude-receivers')) {
190+
$receiverNames = array_diff($receiverNames, $excludedTransports);
191+
192+
if (!$receiverNames) {
193+
throw new RuntimeException('All transports/receivers have been excluded, please specify at least one to consume from.');
194+
}
195+
}
196+
175197
foreach ($receiverNames as $receiverName) {
176198
if (!$this->receiverLocator->has($receiverName)) {
177199
$message = \sprintf('The receiver "%s" does not exist.', $receiverName);
@@ -276,6 +298,10 @@ public function complete(CompletionInput $input, CompletionSuggestions $suggesti
276298
if ($input->mustSuggestOptionValuesFor('bus')) {
277299
$suggestions->suggestValues($this->busIds);
278300
}
301+
302+
if ($input->mustSuggestOptionValuesFor('exclude-receivers')) {
303+
$suggestions->suggestValues($this->receiverNames);
304+
}
279305
}
280306

281307
public function getSubscribedSignals(): array

‎src/Symfony/Component/Messenger/Tests/Command/ConsumeMessagesCommandTest.php‎

Lines changed: 152 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -329,4 +329,156 @@ public static function provideCompletionSuggestions()
329329
yield 'receiver (no repeat)' => [['async', ''], ['async_high', 'failed']];
330330
yield 'option --bus' => [['--bus', ''], ['messenger.bus.default']];
331331
}
332+
333+
public function testRunWithExcludeReceiversOption()
334+
{
335+
$envelope1 = new Envelope(new \stdClass(), [new BusNameStamp('dummy-bus')]);
336+
$envelope2 = new Envelope(new \stdClass(), [new BusNameStamp('dummy-bus')]);
337+
$envelope3 = new Envelope(new \stdClass(), [new BusNameStamp('dummy-bus')]);
338+
339+
$receiver1 = $this->createMock(ReceiverInterface::class);
340+
$receiver1->method('get')->willReturn([$envelope1]);
341+
$receiver2 = $this->createMock(ReceiverInterface::class);
342+
$receiver2->method('get')->willReturn([$envelope2]);
343+
$receiver3 = $this->createMock(ReceiverInterface::class);
344+
$receiver3->method('get')->willReturn([$envelope3]);
345+
346+
$receiverLocator = new Container();
347+
$receiverLocator->set('dummy-receiver1', $receiver1);
348+
$receiverLocator->set('dummy-receiver2', $receiver2);
349+
$receiverLocator->set('dummy-receiver3', $receiver3);
350+
351+
$bus = $this->createMock(MessageBusInterface::class);
352+
// Only 2 messages should be dispatched (receiver1 and receiver3, receiver2 is excluded)
353+
$bus->expects($this->exactly(2))->method('dispatch');
354+
355+
$busLocator = new Container();
356+
$busLocator->set('dummy-bus', $bus);
357+
358+
$command = new ConsumeMessagesCommand(
359+
new RoutableMessageBus($busLocator),
360+
$receiverLocator, new EventDispatcher(),
361+
receiverNames: ['dummy-receiver1', 'dummy-receiver2', 'dummy-receiver3']
362+
);
363+
364+
$application = new Application();
365+
if (method_exists($application, 'addCommand')) {
366+
$application->addCommand($command);
367+
} else {
368+
$application->add($command);
369+
}
370+
$tester = new CommandTester($application->get('messenger:consume'));
371+
$tester->execute([
372+
'--all' => true,
373+
'--exclude-receivers' => ['dummy-receiver2'],
374+
'--limit' => 2,
375+
]);
376+
377+
$tester->assertCommandIsSuccessful();
378+
$this->assertStringContainsString('[OK] Consuming messages from transports "dummy-receiver1, dummy-receiver3"', $tester->getDisplay());
379+
}
380+
381+
public function testRunWithExcludeReceiversMultipleQueues()
382+
{
383+
$envelope1 = new Envelope(new \stdClass(), [new BusNameStamp('dummy-bus')]);
384+
$envelope2 = new Envelope(new \stdClass(), [new BusNameStamp('dummy-bus')]);
385+
$envelope3 = new Envelope(new \stdClass(), [new BusNameStamp('dummy-bus')]);
386+
$envelope4 = new Envelope(new \stdClass(), [new BusNameStamp('dummy-bus')]);
387+
388+
$receiver1 = $this->createMock(ReceiverInterface::class);
389+
$receiver1->method('get')->willReturn([$envelope1]);
390+
$receiver2 = $this->createMock(ReceiverInterface::class);
391+
$receiver2->method('get')->willReturn([$envelope2]);
392+
$receiver3 = $this->createMock(ReceiverInterface::class);
393+
$receiver3->method('get')->willReturn([$envelope3]);
394+
$receiver4 = $this->createMock(ReceiverInterface::class);
395+
$receiver4->method('get')->willReturn([$envelope4]);
396+
397+
$receiverLocator = new Container();
398+
$receiverLocator->set('dummy-receiver1', $receiver1);
399+
$receiverLocator->set('dummy-receiver2', $receiver2);
400+
$receiverLocator->set('dummy-receiver3', $receiver3);
401+
$receiverLocator->set('dummy-receiver4', $receiver4);
402+
403+
$bus = $this->createMock(MessageBusInterface::class);
404+
// Only 2 messages should be dispatched (receiver1 and receiver4, receiver2 and receiver3 are excluded)
405+
$bus->expects($this->exactly(2))->method('dispatch');
406+
407+
$busLocator = new Container();
408+
$busLocator->set('dummy-bus', $bus);
409+
410+
$command = new ConsumeMessagesCommand(
411+
new RoutableMessageBus($busLocator),
412+
$receiverLocator, new EventDispatcher(),
413+
receiverNames: ['dummy-receiver1', 'dummy-receiver2', 'dummy-receiver3', 'dummy-receiver4']
414+
);
415+
416+
$application = new Application();
417+
if (method_exists($application, 'addCommand')) {
418+
$application->addCommand($command);
419+
} else {
420+
$application->add($command);
421+
}
422+
$tester = new CommandTester($application->get('messenger:consume'));
423+
$tester->execute([
424+
'--all' => true,
425+
'--exclude-receivers' => ['dummy-receiver2', 'dummy-receiver3'],
426+
'--limit' => 2,
427+
]);
428+
429+
$tester->assertCommandIsSuccessful();
430+
$this->assertStringContainsString('[OK] Consuming messages from transports "dummy-receiver1, dummy-receiver4"', $tester->getDisplay());
431+
}
432+
433+
public function testExcludeReceiverssWithoutAllOptionThrowsException()
434+
{
435+
$receiverLocator = new Container();
436+
$receiverLocator->set('dummy-receiver', new \stdClass());
437+
438+
$command = new ConsumeMessagesCommand(new RoutableMessageBus(new Container()), $receiverLocator, new EventDispatcher());
439+
440+
$application = new Application();
441+
if (method_exists($application, 'addCommand')) {
442+
$application->addCommand($command);
443+
} else {
444+
$application->add($command);
445+
}
446+
$tester = new CommandTester($application->get('messenger:consume'));
447+
448+
$this->expectException(InvalidOptionException::class);
449+
$this->expectExceptionMessage('The "--exclude-receivers" option can only be used with the "--all" option.');
450+
$tester->execute([
451+
'receivers' => ['dummy-receiver'],
452+
'--exclude-receivers' => ['dummy-receiver'],
453+
]);
454+
}
455+
456+
public function testExcludeReceiversWithAllQueuesExcludedThrowsException()
457+
{
458+
$receiverLocator = new Container();
459+
$receiverLocator->set('dummy-receiver1', new \stdClass());
460+
$receiverLocator->set('dummy-receiver2', new \stdClass());
461+
462+
$command = new ConsumeMessagesCommand(
463+
new RoutableMessageBus(new Container()),
464+
$receiverLocator,
465+
new EventDispatcher(),
466+
receiverNames: ['dummy-receiver1', 'dummy-receiver2']
467+
);
468+
469+
$application = new Application();
470+
if (method_exists($application, 'addCommand')) {
471+
$application->addCommand($command);
472+
} else {
473+
$application->add($command);
474+
}
475+
$tester = new CommandTester($application->get('messenger:consume'));
476+
477+
$this->expectException(\Symfony\Component\Console\Exception\RuntimeException::class);
478+
$this->expectExceptionMessage('All transports/receivers have been excluded, please specify at least one to consume from.');
479+
$tester->execute([
480+
'--all' => true,
481+
'--exclude-receivers' => ['dummy-receiver1', 'dummy-receiver2'],
482+
]);
483+
}
332484
}

0 commit comments

Comments
 (0)