Uses a service locator to discover receivers

Author: sroze

src/Symfony/Bundle/FrameworkBundle/Command/MessageConsumeCommand.php

@@ -11,12 +11,12 @@
 
 namespace Symfony\Bundle\FrameworkBundle\Command;
 
+use Psr\Container\ContainerInterface;
 use Symfony\Component\Console\Command\Command;
 use Symfony\Component\Console\Input\InputArgument;
 use Symfony\Component\Console\Input\InputInterface;
 use Symfony\Component\Console\Input\InputOption;
 use Symfony\Component\Console\Output\OutputInterface;
-use Symfony\Component\DependencyInjection\ContainerInterface;
 use Symfony\Component\Message\MessageBusInterface;
 use Symfony\Component\Message\Transport\Enhancers\MaximumCountReceiver;
 use Symfony\Component\Message\Transport\ReceiverInterface;
@@ -29,6 +29,17 @@ class MessageConsumeCommand extends Command
 {
     protected static $defaultName = 'message:consume';
 
+    private $bus;
+    private $receiverLocator;
+
+    public function __construct(MessageBusInterface $bus, ContainerInterface $receiverLocator)
+    {
+        parent::__construct();
+
+        $this->bus = $bus;
+        $this->receiverLocator = $receiverLocator;
+    }
+
     /**
      * {@inheritdoc}
      */
@@ -37,7 +48,6 @@ protected function configure()
         $this
             ->setDefinition(array(
                 new InputArgument('receiver', InputArgument::REQUIRED, 'Name of the receiver'),
-                new InputOption('bus', 'b', InputOption::VALUE_REQUIRED, 'Name of the bus to dispatch the messages to', 'message_bus'),
                 new InputOption('limit', 'l', InputOption::VALUE_REQUIRED, 'Limit the number of received messages'),
             ))
             ->setDescription('Consumes a message')
@@ -55,30 +65,19 @@ protected function configure()
      */
     protected function execute(InputInterface $input, OutputInterface $output)
     {
-        /** @var ContainerInterface $container */
-        $container = $this->getApplication()->getKernel()->getContainer();
-
-        if (!$container->has($receiverName = $input->getArgument('receiver'))) {
+        if (!$this->receiverLocator->has($receiverName = $input->getArgument('receiver'))) {
             throw new \RuntimeException(sprintf('Receiver "%s" does not exist.', $receiverName));
         }
 
-        if (!($receiver = $container->get($receiverName)) instanceof ReceiverInterface) {
+        if (!($receiver = $this->receiverLocator->get($receiverName)) instanceof ReceiverInterface) {
             throw new \RuntimeException(sprintf('Receiver "%s" is not a valid message consumer. It must implement the "%s" interface.', $receiverName, ReceiverInterface::class));
         }
 
-        if (!$container->has($busName = $input->getOption('bus'))) {
-            throw new \RuntimeException(sprintf('Bus "%s" does not exist.', $busName));
-        }
-
-        if (!($messageBus = $container->get($busName)) instanceof MessageBusInterface) {
-            throw new \RuntimeException(sprintf('Bus "%s" is not a valid message bus. It must implement the "%s" interface.', $busName, MessageBusInterface::class));
-        }
-
         if ($limit = $input->getOption('limit')) {
             $receiver = new MaximumCountReceiver($receiver, $limit);
         }
 
-        $worker = new Worker($receiver, $messageBus);
+        $worker = new Worker($receiver, $this->bus);
         $worker->run();
     }
 }

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

@@ -42,6 +42,7 @@
 use Symfony\Component\DependencyInjection\ParameterBag\ContainerBagInterface;
 use Symfony\Component\DependencyInjection\ParameterBag\ParameterBagInterface;
 use Symfony\Component\DependencyInjection\Reference;
+use Symfony\Component\DependencyInjection\ServiceLocator;
 use Symfony\Component\DependencyInjection\ServiceSubscriberInterface;
 use Symfony\Component\EventDispatcher\EventSubscriberInterface;
 use Symfony\Component\ExpressionLanguage\ExpressionLanguage;
@@ -58,6 +59,8 @@
 use Symfony\Component\Lock\LockInterface;
 use Symfony\Component\Lock\Store\StoreFactory;
 use Symfony\Component\Lock\StoreInterface;
+use Symfony\Component\Message\Transport\ReceiverInterface;
+use Symfony\Component\Message\Transport\SenderInterface;
 use Symfony\Component\PropertyAccess\PropertyAccessor;
 use Symfony\Component\PropertyInfo\PropertyAccessExtractorInterface;
 use Symfony\Component\PropertyInfo\PropertyDescriptionExtractorInterface;
@@ -267,6 +270,8 @@ public function load(array $configs, ContainerBuilder $container)
 
         if ($this->isConfigEnabled($container, $config['message'])) {
             $this->registerMessageConfiguration($config['message'], $container, $loader);
+        } else {
+            $container->removeDefinition('console.command.message_consume');
         }
 
         if ($this->isConfigEnabled($container, $config['web_link'])) {
@@ -334,6 +339,10 @@ public function load(array $configs, ContainerBuilder $container)
             ->addTag('validator.constraint_validator');
         $container->registerForAutoconfiguration(ObjectInitializerInterface::class)
             ->addTag('validator.initializer');
+        $container->registerForAutoconfiguration(ReceiverInterface::class)
+            ->addTag('message.receiver');
+        $container->registerForAutoconfiguration(SenderInterface::class)
+            ->addTag('message.sender');
 
         if (!$container->getParameter('kernel.debug')) {
             // remove tagged iterator argument for resource checkers

src/Symfony/Bundle/FrameworkBundle/Resources/config/console.xml

@@ -65,6 +65,9 @@
         </service>
 
         <service id="console.command.message_consume" class="Symfony\Bundle\FrameworkBundle\Command\MessageConsumeCommand">
+            <argument type="service" id="message_bus" />
+            <argument type="service" id="message.receiver_locator" />
+
             <tag name="console.command" command="message:consume" />
         </service>
 

src/Symfony/Bundle/FrameworkBundle/Resources/config/message.xml

@@ -55,5 +55,11 @@
             <tag name="data_collector" template="@WebProfiler/Collector/messages.html.twig" id="messages" priority="100" />
             <tag name="message_middleware" />
         </service>
+
+        <!-- Discovery -->
+        <service id="message.receiver_locator">
+            <tag name="container.service_locator" />
+            <argument type="collection" />
+        </service>
     </services>
 </container>

src/Symfony/Component/Message/DependencyInjection/MessagePass.php

@@ -47,6 +47,7 @@ public function process(ContainerBuilder $container)
             return;
         }
 
+        $this->registerReceivers($container);
         $this->registerHandlers($container);
     }
 
@@ -113,4 +114,16 @@ private function guessHandledClass(ContainerBuilder $container, string $serviceI
 
         return $parameter->getClass()->getName();
     }
+
+    private function registerReceivers(ContainerBuilder $container)
+    {
+        $receiverMapping = [];
+        foreach ($container->findTaggedServiceIds('message.receiver') as $id => $tags) {
+            foreach ($tags as $tag) {
+                $receiverMapping[$tag['id'] ?? $id] = new Reference($id);
+            }
+        }
+
+        $container->getDefinition('message.receiver_locator')->replaceArgument(0, $receiverMapping);
+    }
 }

src/Symfony/Component/Message/Handler/ChainHandler.php

@@ -37,7 +37,7 @@ public function __construct(array $handlers)
 
     public function __invoke($message)
     {
-        $results = [];
+        $results = array();
 
         foreach ($this->handlers as $handler) {
             $results[] = $handler($message);