From 5f3843b8f867569009d4cd6cbba35b3c7a33ac63 Mon Sep 17 00:00:00 2001
From: Wolfgang Klinger <wk@plan2.net>
Date: Thu, 12 Dec 2024 10:44:13 +0100
Subject: [PATCH] [Messenger] Filter out non-consumable receivers when
 registering `ConsumeMessagesCommand`

---
 Command/ConsumeMessagesCommand.php            |  6 ++++
 DependencyInjection/MessengerPass.php         |  6 +++-
 .../DependencyInjection/MessengerPassTest.php | 29 +++++++++++++++++++
 3 files changed, 40 insertions(+), 1 deletion(-)

diff --git a/Command/ConsumeMessagesCommand.php b/Command/ConsumeMessagesCommand.php
index a959c2ba..7aa8752f 100644
--- a/Command/ConsumeMessagesCommand.php
+++ b/Command/ConsumeMessagesCommand.php
@@ -136,6 +136,12 @@ protected function interact(InputInterface $input, OutputInterface $output)
         $io = new SymfonyStyle($input, $output instanceof ConsoleOutputInterface ? $output->getErrorOutput() : $output);
 
         if ($this->receiverNames && !$input->getArgument('receivers')) {
+            if (1 === \count($this->receiverNames)) {
+                $input->setArgument('receivers', $this->receiverNames);
+
+                return;
+            }
+
             $io->block('Which transports/receivers do you want to consume?', null, 'fg=white;bg=blue', ' ', true);
 
             $io->writeln('Choose which receivers you want to consume messages from in order of priority.');
diff --git a/DependencyInjection/MessengerPass.php b/DependencyInjection/MessengerPass.php
index 032ec76e..98ad2058 100644
--- a/DependencyInjection/MessengerPass.php
+++ b/DependencyInjection/MessengerPass.php
@@ -274,6 +274,7 @@ private function registerReceivers(ContainerBuilder $container, array $busIds):
             }
         }
 
+        $consumableReceiverNames = [];
         foreach ($container->findTaggedServiceIds('messenger.receiver') as $id => $tags) {
             $receiverClass = $this->getServiceClass($container, $id);
             if (!is_subclass_of($receiverClass, ReceiverInterface::class)) {
@@ -289,6 +290,9 @@ private function registerReceivers(ContainerBuilder $container, array $busIds):
                         $failureTransportsMap[$tag['alias']] = $receiverMapping[$id];
                     }
                 }
+                if (!isset($tag['is_consumable']) || $tag['is_consumable'] !== false) {
+                    $consumableReceiverNames[] = $tag['alias'] ?? $id;
+                }
             }
         }
 
@@ -314,7 +318,7 @@ private function registerReceivers(ContainerBuilder $container, array $busIds):
                 $consumeCommandDefinition->replaceArgument(0, new Reference('messenger.routable_message_bus'));
             }
 
-            $consumeCommandDefinition->replaceArgument(4, array_values($receiverNames));
+            $consumeCommandDefinition->replaceArgument(4, $consumableReceiverNames);
             try {
                 $consumeCommandDefinition->replaceArgument(6, $busIds);
             } catch (OutOfBoundsException) {
diff --git a/Tests/DependencyInjection/MessengerPassTest.php b/Tests/DependencyInjection/MessengerPassTest.php
index 13d18993..e75117e5 100644
--- a/Tests/DependencyInjection/MessengerPassTest.php
+++ b/Tests/DependencyInjection/MessengerPassTest.php
@@ -13,6 +13,7 @@
 
 use PHPUnit\Framework\TestCase;
 use Symfony\Bridge\PhpUnit\ExpectDeprecationTrait;
+use Symfony\Component\Console\Command\Command;
 use Symfony\Component\DependencyInjection\ChildDefinition;
 use Symfony\Component\DependencyInjection\Compiler\AttributeAutoconfigurationPass;
 use Symfony\Component\DependencyInjection\Compiler\ResolveChildDefinitionsPass;
@@ -506,6 +507,34 @@ public function testItSetsTheReceiverNamesOnTheSetupTransportsCommand()
         $this->assertSame(['amqp', 'dummy'], $container->getDefinition('console.command.messenger_setup_transports')->getArgument(1));
     }
 
+    public function testOnlyConsumableTransportsAreAddedToConsumeCommand()
+    {
+        $container = new ContainerBuilder();
+
+        $container->register('messenger.transport.async', DummyReceiver::class)
+            ->addTag('messenger.receiver', ['alias' => 'async']);
+        $container->register('messenger.transport.sync', DummyReceiver::class)
+            ->addTag('messenger.receiver', ['alias' => 'sync', 'is_consumable' => false]);
+        $container->register('messenger.receiver_locator', ServiceLocator::class)
+            ->setArguments([[]]);
+
+        $container->register('console.command.messenger_consume_messages', Command::class)
+            ->setArguments([
+                null,
+                null,
+                null,
+                null,
+                [],
+            ]);
+
+        (new MessengerPass())->process($container);
+
+        $this->assertSame(
+            ['async'],
+            $container->getDefinition('console.command.messenger_consume_messages')->getArgument(4)
+        );
+    }
+
     /**
      * @group legacy
      */