bug symfony#59198 [Messenger] Filter out non-consumable receivers when registering ConsumeMessagesCommand (wazum)

This PR was squashed before being merged into the 6.4 branch.

Discussion
----------

[Messenger] Filter out non-consumable receivers when registering `ConsumeMessagesCommand`

| Q             | A
| ------------- | ---
| Branch?       | 6.4
| Bug fix?      | yes
| New feature?  | no
| Deprecations? | no
| Issues        | Fix symfony#51556
| License       | MIT

This pull request enhances the `MessengerPass` in the Symfony Framework Bundle to automatically filter out non-consumable receivers when registering the `ConsumeMessagesCommand`. The key changes are:

The FrameworkExtension has been updated to mark the "sync" transport as non-consumable by default, without requiring the user to explicitly set the `is_consumable` attribute.

The `MessengerPass` then checks the `is_consumable` attribute on the `messenger.receiver` tags. If the attribute is set to `false`, the receiver is considered non-consumable and will not be added to the list of receivers for the consume command.
If there is only one consumable receiver, the consume command will automatically use that receiver without prompting the user to select one (I added this as a separate commit if this is too implicit).

These changes improve the user experience by removing unnecessary (and non-functioning) options.

This is my first pull request for the _Symfony_ project, so please let me know what I can improve.

Commits
-------

34c7e6f [Messenger] Filter out non-consumable receivers when registering `ConsumeMessagesCommand`

Author: fabpot

src/Symfony/Bundle/FrameworkBundle/DependencyInjection/FrameworkExtension.php

@@ -2282,13 +2282,17 @@ private function registerMessengerConfiguration(array $config, ContainerBuilder
         $transportRateLimiterReferences = [];
         foreach ($config['transports'] as $name => $transport) {
             $serializerId = $transport['serializer'] ?? 'messenger.default_serializer';
+            $tags = [
+                'alias' => $name,
+                'is_failure_transport' => \in_array($name, $failureTransports),
+            ];
+            if (str_starts_with($transport['dsn'], 'sync://')) {
+                $tags['is_consumable'] = false;
+            }
             $transportDefinition = (new Definition(TransportInterface::class))
                 ->setFactory([new Reference('messenger.transport_factory'), 'createTransport'])
                 ->setArguments([$transport['dsn'], $transport['options'] + ['transport_name' => $name], new Reference($serializerId)])
-                ->addTag('messenger.receiver', [
-                    'alias' => $name,
-                    'is_failure_transport' => \in_array($name, $failureTransports),
-                ])
+                ->addTag('messenger.receiver', $tags)
             ;
             $container->setDefinition($transportId = 'messenger.transport.'.$name, $transportDefinition);
             $senderAliases[$name] = $transportId;

src/Symfony/Component/Messenger/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.');

src/Symfony/Component/Messenger/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) {

src/Symfony/Component/Messenger/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
      */