feature #27275 [Messenger] Allow to scope handlers per bus (ogizanagi, sroze)

This PR was merged into the 4.1 branch.

Discussion
----------

[Messenger] Allow to scope handlers per bus

| Q             | A
| ------------- | ---
| Branch?       | 4.1 <!-- see below -->
| Bug fix?      | no
| New feature?  | yes <!-- don't forget to update src/**/CHANGELOG.md files -->
| BC breaks?    | no     <!-- see https://symfony.com/bc -->
| Deprecations? | no <!-- don't forget to update UPGRADE-*.md and src/**/CHANGELOG.md files -->
| Tests pass?   | yes    <!-- please add some, will be required by reviewers -->
| Fixed tickets | N/A   <!-- #-prefixed issue number(s), if any -->
| License       | MIT
| Doc PR        | todo

<img width="970" alt="screenshot 2018-05-15 a 18 34 11" src="https://user-images.githubusercontent.com/2211145/40070551-b36d1f50-586e-11e8-8a44-c6a1503312dd.PNG">

This prevents from dispatching a message in the wrong bus. It works especially great with PSR-4 discovery in DI:

```yaml
command_handlers:
    namespace: App\Message\
    resource: '%kernel.project_dir%/src/Message/*CommandHandler.php'
    tags:
        - { name: messenger.message_handler, bus: command_bus }

query_handlers:
    namespace: App\Message\
    resource: '%kernel.project_dir%/src/Message/*QueryHandler.php'
    tags:
        - { name: messenger.message_handler, bus: query_bus }
```

<details>

<summary>(or in some layered architecture)</summary>

```yaml
command_handlers:
    namespace: App\Application\
    resource: '%kernel.project_dir%/src/Application/**/Handler/*CommandHandler.php'
    tags:
        - { name: messenger.message_handler, bus: command_bus }

query_handlers:
    namespace: App\Application\
    resource: '%kernel.project_dir%/src/Application/**/Handler/*QueryHandler.php'
    tags:
        - { name: messenger.message_handler, bus: query_bus }
```

</details>

---
It also updates (and add tests for) the `DebugCommand`, so we can inspect easily where is supposed to be handled each message.
It's totally optional, and if the bus isn't provided, then it stays available in every bus.

Commits
-------

fd810cdee0 Uses `protected` for test functions
9d658e915e [Messenger] Allow to scope handlers per bus

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

@@ -1483,7 +1483,7 @@ class FrameworkExtension extends Extension
             }
 
             $container->setParameter($busId.'.middleware', $middleware);
-            $container->setDefinition($busId, (new Definition(MessageBus::class, array(array())))->addTag('messenger.bus'));
+            $container->register($busId, MessageBus::class)->addArgument(array())->addTag('messenger.bus');
 
             if ($busId === $config['default_bus']) {
                 $container->setAlias('message_bus', $busId);

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

@@ -7,11 +7,6 @@
     <services>
         <defaults public="false" />
 
-        <!-- Handlers -->
-        <service id="messenger.handler_resolver" class="Symfony\Component\Messenger\Handler\Locator\ContainerHandlerLocator">
-            <argument type="service" id="service_container"/>
-        </service>
-
         <!-- Asynchronous -->
         <service id="messenger.asynchronous.routing.sender_locator" class="Symfony\Component\Messenger\Asynchronous\Routing\SenderLocator">
             <argument type="service" id="messenger.sender_locator" />
@@ -32,7 +27,7 @@
         <!-- Middleware -->
         <service id="messenger.middleware.allow_no_handler" class="Symfony\Component\Messenger\Middleware\AllowNoHandlerMiddleware" abstract="true" />
         <service id="messenger.middleware.call_message_handler" class="Symfony\Component\Messenger\Middleware\HandleMessageMiddleware" abstract="true">
-            <argument type="service" id="messenger.handler_resolver" />
+            <argument /> <!-- Bus handler resolver -->
         </service>
 
         <service id="messenger.middleware.validation" class="Symfony\Component\Messenger\Middleware\ValidationMiddleware" abstract="true">

src/Symfony/Component/Messenger/Command/DebugCommand.php

@@ -12,6 +12,8 @@
 namespace Symfony\Component\Messenger\Command;
 
 use Symfony\Component\Console\Command\Command;
+use Symfony\Component\Console\Exception\RuntimeException;
+use Symfony\Component\Console\Input\InputArgument;
 use Symfony\Component\Console\Input\InputInterface;
 use Symfony\Component\Console\Output\OutputInterface;
 use Symfony\Component\Console\Style\SymfonyStyle;
@@ -31,9 +33,9 @@ class DebugCommand extends Command
 
     public function __construct(array $mapping)
     {
-        parent::__construct();
-
         $this->mapping = $mapping;
+
+        parent::__construct();
     }
 
     /**
@@ -42,13 +44,18 @@ class DebugCommand extends Command
     protected function configure()
     {
         $this
-            ->setDescription('Lists messages you can dispatch using the message bus')
+            ->addArgument('bus', InputArgument::OPTIONAL, sprintf('The bus id (one of %s)', implode(', ', array_keys($this->mapping))), null)
+            ->setDescription('Lists messages you can dispatch using the message buses')
             ->setHelp(<<<'EOF'
 The <info>%command.name%</info> command displays all messages that can be
-dispatched using the message bus:
+dispatched using the message buses:
 
   <info>php %command.full_name%</info>
 
+Or for a specific bus only:
+
+  <info>php %command.full_name% command_bus</info>
+
 EOF
             )
         ;
@@ -61,21 +68,33 @@ EOF
     {
         $io = new SymfonyStyle($input, $output);
         $io->title('Messenger');
-        $io->text('The following messages can be dispatched:');
-        $io->newLine();
 
-        $tableRows = array();
-        foreach ($this->mapping as $message => $handlers) {
-            $tableRows[] = array(sprintf('<fg=cyan>%s</fg=cyan>', $message));
-            foreach ($handlers as $handler) {
-                $tableRows[] = array(sprintf('    handled by %s', $handler));
+        $mapping = $this->mapping;
+        if ($bus = $input->getArgument('bus')) {
+            if (!isset($mapping[$bus])) {
+                throw new RuntimeException(sprintf('Bus "%s" does not exist. Known buses are %s.', $bus, implode(', ', array_keys($this->mapping))));
             }
+            $mapping = array($bus => $mapping[$bus]);
         }
 
-        if ($tableRows) {
-            $io->table(array(), $tableRows);
-        } else {
-            $io->text('No messages were found that have valid handlers.');
+        foreach ($mapping as $bus => $handlersByMessage) {
+            $io->section($bus);
+
+            $tableRows = array();
+            foreach ($handlersByMessage as $message => $handlers) {
+                $tableRows[] = array(sprintf('<fg=cyan>%s</fg=cyan>', $message));
+                foreach ($handlers as $handler) {
+                    $tableRows[] = array(sprintf('    handled by <info>%s</>', $handler));
+                }
+            }
+
+            if ($tableRows) {
+                $io->text('The following messages can be dispatched:');
+                $io->newLine();
+                $io->table(array(), $tableRows);
+            } else {
+                $io->warning(sprintf('No handled message found in bus "%s".', $bus));
+            }
         }
     }
 }

src/Symfony/Component/Messenger/DependencyInjection/MessengerPass.php

@@ -20,6 +20,7 @@ use Symfony\Component\DependencyInjection\Definition;
 use Symfony\Component\DependencyInjection\Exception\RuntimeException;
 use Symfony\Component\DependencyInjection\Reference;
 use Symfony\Component\Messenger\Handler\ChainHandler;
+use Symfony\Component\Messenger\Handler\Locator\ContainerHandlerLocator;
 use Symfony\Component\Messenger\Handler\MessageHandlerInterface;
 use Symfony\Component\Messenger\Handler\MessageSubscriberInterface;
 use Symfony\Component\Messenger\TraceableMessageBus;
@@ -51,11 +52,13 @@ class MessengerPass implements CompilerPassInterface
      */
     public function process(ContainerBuilder $container)
     {
-        if (!$container->hasDefinition('messenger.handler_resolver')) {
+        if (!$container->has('message_bus')) {
             return;
         }
 
+        $busIds = array();
         foreach ($container->findTaggedServiceIds($this->busTag) as $busId => $tags) {
+            $busIds[] = $busId;
             if ($container->hasParameter($busMiddlewareParameter = $busId.'.middleware')) {
                 $this->registerBusMiddleware($container, $busId, $container->getParameter($busMiddlewareParameter));
 
@@ -69,16 +72,20 @@ class MessengerPass implements CompilerPassInterface
 
         $this->registerReceivers($container);
         $this->registerSenders($container);
-        $this->registerHandlers($container);
+        $this->registerHandlers($container, $busIds);
     }
 
-    private function registerHandlers(ContainerBuilder $container)
+    private function registerHandlers(ContainerBuilder $container, array $busIds)
     {
         $definitions = array();
-        $handlersByMessage = array();
+        $handlersByBusAndMessage = array();
 
         foreach ($container->findTaggedServiceIds($this->handlerTag, true) as $serviceId => $tags) {
             foreach ($tags as $tag) {
+                if (isset($tag['bus']) && !\in_array($tag['bus'], $busIds, true)) {
+                    throw new RuntimeException(sprintf('Invalid handler service "%s": bus "%s" specified on the tag "%s" does not exist (known ones are: %s).', $serviceId, $tag['bus'], $this->handlerTag, implode(', ', $busIds)));
+                }
+
                 $r = $container->getReflectionClass($container->getDefinition($serviceId)->getClass());
 
                 if (isset($tag['handles'])) {
@@ -88,6 +95,7 @@ class MessengerPass implements CompilerPassInterface
                 }
 
                 $priority = $tag['priority'] ?? 0;
+                $handlerBuses = (array) ($tag['bus'] ?? $busIds);
 
                 foreach ($handles as $messageClass => $method) {
                     if (\is_int($messageClass)) {
@@ -123,38 +131,57 @@ class MessengerPass implements CompilerPassInterface
                         $definitions[$serviceId = '.messenger.method_on_object_wrapper.'.ContainerBuilder::hash($messageClass.':'.$messagePriority.':'.$serviceId.':'.$method)] = $wrapperDefinition;
                     }
 
-                    $handlersByMessage[$messageClass][$messagePriority][] = new Reference($serviceId);
+                    foreach ($handlerBuses as $handlerBus) {
+                        $handlersByBusAndMessage[$handlerBus][$messageClass][$messagePriority][] = $serviceId;
+                    }
                 }
             }
         }
 
-        foreach ($handlersByMessage as $message => $handlers) {
-            krsort($handlersByMessage[$message]);
-            $handlersByMessage[$message] = array_merge(...$handlersByMessage[$message]);
+        foreach ($handlersByBusAndMessage as $bus => $handlersByMessage) {
+            foreach ($handlersByMessage as $message => $handlersByPriority) {
+                krsort($handlersByPriority);
+                $handlersByBusAndMessage[$bus][$message] = array_unique(array_merge(...$handlersByPriority));
+            }
         }
 
-        $handlersLocatorMapping = array();
-        foreach ($handlersByMessage as $message => $handlers) {
-            if (1 === \count($handlers)) {
-                $handlersLocatorMapping['handler.'.$message] = current($handlers);
-            } else {
-                $d = new Definition(ChainHandler::class, array($handlers));
-                $d->setPrivate(true);
-                $serviceId = hash('sha1', $message);
-                $definitions[$serviceId] = $d;
-                $handlersLocatorMapping['handler.'.$message] = new Reference($serviceId);
+        $handlersLocatorMappingByBus = array();
+        foreach ($handlersByBusAndMessage as $bus => $handlersByMessage) {
+            foreach ($handlersByMessage as $message => $handlersIds) {
+                if (1 === \count($handlersIds)) {
+                    $handlersLocatorMappingByBus[$bus]['handler.'.$message] = new Reference(current($handlersIds));
+                } else {
+                    $chainHandler = new Definition(ChainHandler::class, array(array_map(function (string $handlerId): Reference {
+                        return new Reference($handlerId);
+                    }, $handlersIds)));
+                    $chainHandler->setPrivate(true);
+                    $serviceId = '.messenger.chain_handler.'.ContainerBuilder::hash($bus.$message);
+                    $definitions[$serviceId] = $chainHandler;
+                    $handlersLocatorMappingByBus[$bus]['handler.'.$message] = new Reference($serviceId);
+                }
             }
         }
         $container->addDefinitions($definitions);
 
-        $handlerResolver = $container->getDefinition('messenger.handler_resolver');
-        $handlerResolver->replaceArgument(0, ServiceLocatorTagPass::register($container, $handlersLocatorMapping));
+        foreach ($busIds as $bus) {
+            $container->register($resolverName = "$bus.messenger.handler_resolver", ContainerHandlerLocator::class)
+                ->setArgument(0, ServiceLocatorTagPass::register($container, $handlersLocatorMappingByBus[$bus] ?? array()))
+            ;
+            if ($container->has($callMessageHandlerId = "$bus.middleware.call_message_handler")) {
+                $container->getDefinition($callMessageHandlerId)
+                    ->replaceArgument(0, new Reference($resolverName))
+                ;
+            }
+        }
 
         if ($container->hasDefinition('console.command.messenger_debug')) {
-            $container->getDefinition('console.command.messenger_debug')
-                ->replaceArgument(0, array_map(function (array $handlers): array {
-                    return array_map('strval', $handlers);
-                }, $handlersByMessage));
+            $debugCommandMapping = $handlersByBusAndMessage;
+            foreach ($busIds as $bus) {
+                if (!isset($debugCommandMapping[$bus])) {
+                    $debugCommandMapping[$bus] = array();
+                }
+            }
+            $container->getDefinition('console.command.messenger_debug')->replaceArgument(0, $debugCommandMapping);
         }
     }
 

src/Symfony/Component/Messenger/Tests/Command/DebugCommandTest.php

@@ -0,0 +1,155 @@
+<?php
+
+/*
+ * This file is part of the Symfony package.
+ *
+ * (c) Fabien Potencier <fabien@symfony.com>
+ *
+ * For the full copyright and license information, please view the LICENSE
+ * file that was distributed with this source code.
+ */
+
+namespace Symfony\Component\Messenger\Tests\Command;
+
+use PHPUnit\Framework\TestCase;
+use Symfony\Component\Console\Tester\CommandTester;
+use Symfony\Component\Messenger\Command\DebugCommand;
+use Symfony\Component\Messenger\Tests\Fixtures\DummyCommand;
+use Symfony\Component\Messenger\Tests\Fixtures\DummyCommandHandler;
+use Symfony\Component\Messenger\Tests\Fixtures\DummyQuery;
+use Symfony\Component\Messenger\Tests\Fixtures\DummyQueryHandler;
+use Symfony\Component\Messenger\Tests\Fixtures\MultipleBusesMessage;
+use Symfony\Component\Messenger\Tests\Fixtures\MultipleBusesMessageHandler;
+
+/**
+ * @author Maxime Steinhausser <maxime.steinhausser@gmail.com>
+ */
+class DebugCommandTest extends TestCase
+{
+    protected function setUp()
+    {
+        putenv('COLUMNS='.(119 + strlen(PHP_EOL)));
+    }
+
+    protected function tearDown()
+    {
+        putenv('COLUMNS=');
+    }
+
+    public function testOutput()
+    {
+        $command = new DebugCommand(
+            array(
+                'command_bus' => array(
+                    DummyCommand::class => array(DummyCommandHandler::class),
+                    MultipleBusesMessage::class => array(MultipleBusesMessageHandler::class),
+                ),
+                'query_bus' => array(
+                    DummyQuery::class => array(DummyQueryHandler::class),
+                    MultipleBusesMessage::class => array(MultipleBusesMessageHandler::class),
+                ),
+            )
+        );
+
+        $tester = new CommandTester($command);
+        $tester->execute(array(), array('decorated' => false));
+
+        $this->assertSame(<<<TXT
+
+Messenger
+=========
+
+command_bus
+-----------
+
+ The following messages can be dispatched:
+
+ --------------------------------------------------------------------------------------- 
+  Symfony\Component\Messenger\Tests\Fixtures\DummyCommand                                
+      handled by Symfony\Component\Messenger\Tests\Fixtures\DummyCommandHandler          
+  Symfony\Component\Messenger\Tests\Fixtures\MultipleBusesMessage                        
+      handled by Symfony\Component\Messenger\Tests\Fixtures\MultipleBusesMessageHandler  
+ --------------------------------------------------------------------------------------- 
+
+query_bus
+---------
+
+ The following messages can be dispatched:
+
+ --------------------------------------------------------------------------------------- 
+  Symfony\Component\Messenger\Tests\Fixtures\DummyQuery                                  
+      handled by Symfony\Component\Messenger\Tests\Fixtures\DummyQueryHandler            
+  Symfony\Component\Messenger\Tests\Fixtures\MultipleBusesMessage                        
+      handled by Symfony\Component\Messenger\Tests\Fixtures\MultipleBusesMessageHandler  
+ --------------------------------------------------------------------------------------- 
+
+
+TXT
+            , $tester->getDisplay(true)
+        );
+
+        $tester->execute(array('bus' => 'query_bus'), array('decorated' => false));
+
+        $this->assertSame(<<<TXT
+
+Messenger
+=========
+
+query_bus
+---------
+
+ The following messages can be dispatched:
+
+ --------------------------------------------------------------------------------------- 
+  Symfony\Component\Messenger\Tests\Fixtures\DummyQuery                                  
+      handled by Symfony\Component\Messenger\Tests\Fixtures\DummyQueryHandler            
+  Symfony\Component\Messenger\Tests\Fixtures\MultipleBusesMessage                        
+      handled by Symfony\Component\Messenger\Tests\Fixtures\MultipleBusesMessageHandler  
+ --------------------------------------------------------------------------------------- 
+
+
+TXT
+            , $tester->getDisplay(true)
+        );
+    }
+
+    public function testOutputWithoutMessages()
+    {
+        $command = new DebugCommand(array('command_bus' => array(), 'query_bus' => array()));
+
+        $tester = new CommandTester($command);
+        $tester->execute(array(), array('decorated' => false));
+
+        $this->assertSame(<<<TXT
+
+Messenger
+=========
+
+command_bus
+-----------
+
+ [WARNING] No handled message found in bus "command_bus".                                                               
+
+query_bus
+---------
+
+ [WARNING] No handled message found in bus "query_bus".                                                                 
+
+
+TXT
+            , $tester->getDisplay(true)
+        );
+    }
+
+    /**
+     * @expectedException \Symfony\Component\Console\Exception\RuntimeException
+     * @expectedExceptionMessage Bus "unknown_bus" does not exist. Known buses are command_bus, query_bus.
+     */
+    public function testExceptionOnUnknownBusArgument()
+    {
+        $command = new DebugCommand(array('command_bus' => array(), 'query_bus' => array()));
+
+        $tester = new CommandTester($command);
+        $tester->execute(array('bus' => 'unknown_bus'), array('decorated' => false));
+    }
+}

src/Symfony/Component/Messenger/Tests/DependencyInjection/MessengerPassTest.php

@@ -14,30 +14,38 @@ namespace Symfony\Component\Messenger\Tests\DependencyInjection;
 use PHPUnit\Framework\TestCase;
 use Symfony\Component\DependencyInjection\Argument\ServiceClosureArgument;
 use Symfony\Component\DependencyInjection\Compiler\ResolveChildDefinitionsPass;
+use Symfony\Component\DependencyInjection\Compiler\ResolveClassPass;
 use Symfony\Component\DependencyInjection\ContainerBuilder;
 use Symfony\Component\DependencyInjection\Reference;
 use Symfony\Component\DependencyInjection\ServiceLocator;
 use Symfony\Component\Messenger\Command\ConsumeMessagesCommand;
-use Symfony\Component\Messenger\Envelope;
-use Symfony\Component\Messenger\Transport\AmqpExt\AmqpReceiver;
-use Symfony\Component\Messenger\Transport\AmqpExt\AmqpSender;
-use Symfony\Component\Messenger\Handler\Locator\ContainerHandlerLocator;
+use Symfony\Component\Messenger\Command\DebugCommand;
 use Symfony\Component\Messenger\DataCollector\MessengerDataCollector;
 use Symfony\Component\Messenger\DependencyInjection\MessengerPass;
+use Symfony\Component\Messenger\Envelope;
 use Symfony\Component\Messenger\Handler\ChainHandler;
 use Symfony\Component\Messenger\Handler\MessageSubscriberInterface;
 use Symfony\Component\Messenger\MessageBusInterface;
 use Symfony\Component\Messenger\Middleware\AllowNoHandlerMiddleware;
+use Symfony\Component\Messenger\Middleware\HandleMessageMiddleware;
 use Symfony\Component\Messenger\Middleware\MiddlewareInterface;
+use Symfony\Component\Messenger\Tests\Fixtures\DummyCommand;
+use Symfony\Component\Messenger\Tests\Fixtures\DummyCommandHandler;
 use Symfony\Component\Messenger\Tests\Fixtures\DummyMessage;
+use Symfony\Component\Messenger\Tests\Fixtures\DummyQuery;
+use Symfony\Component\Messenger\Tests\Fixtures\DummyQueryHandler;
+use Symfony\Component\Messenger\Tests\Fixtures\MultipleBusesMessage;
+use Symfony\Component\Messenger\Tests\Fixtures\MultipleBusesMessageHandler;
 use Symfony\Component\Messenger\Tests\Fixtures\SecondMessage;
+use Symfony\Component\Messenger\Transport\AmqpExt\AmqpReceiver;
+use Symfony\Component\Messenger\Transport\AmqpExt\AmqpSender;
 use Symfony\Component\Messenger\Transport\ReceiverInterface;
 
 class MessengerPassTest extends TestCase
 {
     public function testProcess()
     {
-        $container = $this->getContainerBuilder();
+        $container = $this->getContainerBuilder($busId = 'message_bus');
         $container
             ->register(DummyHandler::class, DummyHandler::class)
             ->addTag('messenger.message_handler')
@@ -55,7 +63,7 @@ class MessengerPassTest extends TestCase
 
         $this->assertFalse($container->hasDefinition('messenger.middleware.debug.logging'));
 
-        $handlerLocatorDefinition = $container->getDefinition($container->getDefinition('messenger.handler_resolver')->getArgument(0));
+        $handlerLocatorDefinition = $container->getDefinition($container->getDefinition("$busId.messenger.handler_resolver")->getArgument(0));
         $this->assertSame(ServiceLocator::class, $handlerLocatorDefinition->getClass());
         $this->assertEquals(
             array(
@@ -71,9 +79,68 @@ class MessengerPassTest extends TestCase
         );
     }
 
+    public function testProcessHandlersByBus()
+    {
+        $container = $this->getContainerBuilder($commandBusId = 'command_bus');
+        $container->register($queryBusId = 'query_bus', MessageBusInterface::class)->setArgument(0, array())->addTag('messenger.bus');
+        $container->register('messenger.middleware.call_message_handler', HandleMessageMiddleware::class)
+            ->addArgument(null)
+            ->setAbstract(true)
+        ;
+
+        $middlewares = array(array('id' => 'call_message_handler'));
+
+        $container->setParameter($commandBusId.'.middleware', $middlewares);
+        $container->setParameter($queryBusId.'.middleware', $middlewares);
+
+        $container->register(DummyCommandHandler::class)->addTag('messenger.message_handler', array('bus' => $commandBusId));
+        $container->register(DummyQueryHandler::class)->addTag('messenger.message_handler', array('bus' => $queryBusId));
+        $container->register(MultipleBusesMessageHandler::class)
+            ->addTag('messenger.message_handler', array('bus' => $commandBusId))
+            ->addTag('messenger.message_handler', array('bus' => $queryBusId))
+        ;
+
+        (new ResolveClassPass())->process($container);
+        (new MessengerPass())->process($container);
+
+        $commandBusHandlerLocatorDefinition = $container->getDefinition($container->getDefinition("$commandBusId.messenger.handler_resolver")->getArgument(0));
+        $this->assertSame(ServiceLocator::class, $commandBusHandlerLocatorDefinition->getClass());
+        $this->assertEquals(
+            array(
+                'handler.'.DummyCommand::class => new ServiceClosureArgument(new Reference(DummyCommandHandler::class)),
+                'handler.'.MultipleBusesMessage::class => new ServiceClosureArgument(new Reference(MultipleBusesMessageHandler::class)),
+            ),
+            $commandBusHandlerLocatorDefinition->getArgument(0)
+        );
+
+        $queryBusHandlerLocatorDefinition = $container->getDefinition($container->getDefinition("$queryBusId.messenger.handler_resolver")->getArgument(0));
+        $this->assertSame(ServiceLocator::class, $queryBusHandlerLocatorDefinition->getClass());
+        $this->assertEquals(
+            array(
+                'handler.'.DummyQuery::class => new ServiceClosureArgument(new Reference(DummyQueryHandler::class)),
+                'handler.'.MultipleBusesMessage::class => new ServiceClosureArgument(new Reference(MultipleBusesMessageHandler::class)),
+            ),
+            $queryBusHandlerLocatorDefinition->getArgument(0)
+        );
+    }
+
+    /**
+     * @expectedException \Symfony\Component\DependencyInjection\Exception\RuntimeException
+     * @expectedExceptionMessage Invalid handler service "Symfony\Component\Messenger\Tests\Fixtures\DummyCommandHandler": bus "unknown_bus" specified on the tag "messenger.message_handler" does not exist (known ones are: command_bus).
+     */
+    public function testProcessTagWithUnknownBus()
+    {
+        $container = $this->getContainerBuilder($commandBusId = 'command_bus');
+
+        $container->register(DummyCommandHandler::class)->addTag('messenger.message_handler', array('bus' => 'unknown_bus'));
+
+        (new ResolveClassPass())->process($container);
+        (new MessengerPass())->process($container);
+    }
+
     public function testGetClassesFromTheHandlerSubscriberInterface()
     {
-        $container = $this->getContainerBuilder();
+        $container = $this->getContainerBuilder($busId = 'message_bus');
         $container
             ->register(HandlerWithMultipleMessages::class, HandlerWithMultipleMessages::class)
             ->addTag('messenger.message_handler')
@@ -85,7 +152,7 @@ class MessengerPassTest extends TestCase
 
         (new MessengerPass())->process($container);
 
-        $handlerLocatorDefinition = $container->getDefinition($container->getDefinition('messenger.handler_resolver')->getArgument(0));
+        $handlerLocatorDefinition = $container->getDefinition($container->getDefinition("$busId.messenger.handler_resolver")->getArgument(0));
         $handlerMapping = $handlerLocatorDefinition->getArgument(0);
 
         $this->assertArrayHasKey('handler.'.DummyMessage::class, $handlerMapping);
@@ -101,7 +168,7 @@ class MessengerPassTest extends TestCase
 
     public function testGetClassesAndMethodsAndPrioritiesFromTheSubscriber()
     {
-        $container = $this->getContainerBuilder();
+        $container = $this->getContainerBuilder($busId = 'message_bus');
         $container
             ->register(HandlerMappingMethods::class, HandlerMappingMethods::class)
             ->addTag('messenger.message_handler')
@@ -113,7 +180,7 @@ class MessengerPassTest extends TestCase
 
         (new MessengerPass())->process($container);
 
-        $handlerLocatorDefinition = $container->getDefinition($container->getDefinition('messenger.handler_resolver')->getArgument(0));
+        $handlerLocatorDefinition = $container->getDefinition($container->getDefinition("$busId.messenger.handler_resolver")->getArgument(0));
         $handlerMapping = $handlerLocatorDefinition->getArgument(0);
 
         $this->assertArrayHasKey('handler.'.DummyMessage::class, $handlerMapping);
@@ -138,6 +205,7 @@ class MessengerPassTest extends TestCase
     public function testThrowsExceptionIfTheHandlerMethodDoesNotExist()
     {
         $container = $this->getContainerBuilder();
+        $container->register('message_bus', MessageBusInterface::class)->addTag('messenger.bus');
         $container
             ->register(HandlerMappingWithNonExistentMethod::class, HandlerMappingWithNonExistentMethod::class)
             ->addTag('messenger.message_handler')
@@ -149,6 +217,7 @@ class MessengerPassTest extends TestCase
     public function testItRegistersReceivers()
     {
         $container = $this->getContainerBuilder();
+        $container->register('message_bus', MessageBusInterface::class)->addTag('messenger.bus');
         $container->register(AmqpReceiver::class, AmqpReceiver::class)->addTag('messenger.receiver', array('alias' => 'amqp'));
 
         (new MessengerPass())->process($container);
@@ -159,6 +228,7 @@ class MessengerPassTest extends TestCase
     public function testItRegistersReceiversWithoutTagName()
     {
         $container = $this->getContainerBuilder();
+        $container->register('message_bus', MessageBusInterface::class)->addTag('messenger.bus');
         $container->register(AmqpReceiver::class, AmqpReceiver::class)->addTag('messenger.receiver');
 
         (new MessengerPass())->process($container);
@@ -326,9 +396,8 @@ class MessengerPassTest extends TestCase
     {
         $dataCollector = $this->getMockBuilder(MessengerDataCollector::class)->getMock();
 
-        $container = $this->getContainerBuilder();
+        $container = $this->getContainerBuilder($fooBusId = 'messenger.bus.foo');
         $container->register('messenger.data_collector', $dataCollector);
-        $container->register($fooBusId = 'messenger.bus.foo', MessageBusInterface::class)->addTag('messenger.bus');
         $container->setParameter('kernel.debug', true);
 
         (new MessengerPass())->process($container);
@@ -340,8 +409,7 @@ class MessengerPassTest extends TestCase
 
     public function testRegistersMiddlewareFromServices()
     {
-        $container = $this->getContainerBuilder();
-        $container->register($fooBusId = 'messenger.bus.foo', MessageBusInterface::class)->setArgument(0, array())->addTag('messenger.bus');
+        $container = $this->getContainerBuilder($fooBusId = 'messenger.bus.foo');
         $container->register('messenger.middleware.allow_no_handler', AllowNoHandlerMiddleware::class)->setAbstract(true);
         $container->register('middleware_with_factory', UselessMiddleware::class)->addArgument('some_default')->setAbstract(true);
         $container->register('middleware_with_factory_using_default', UselessMiddleware::class)->addArgument('some_default')->setAbstract(true);
@@ -388,8 +456,7 @@ class MessengerPassTest extends TestCase
      */
     public function testCannotRegistersAnUndefinedMiddleware()
     {
-        $container = $this->getContainerBuilder();
-        $container->register($fooBusId = 'messenger.bus.foo', MessageBusInterface::class)->setArgument(0, array())->addTag('messenger.bus');
+        $container = $this->getContainerBuilder($fooBusId = 'messenger.bus.foo');
         $container->setParameter($middlewareParameter = $fooBusId.'.middleware', array(
             array('id' => 'not_defined_middleware', 'arguments' => array()),
         ));
@@ -403,9 +470,8 @@ class MessengerPassTest extends TestCase
      */
     public function testMiddlewareFactoryDefinitionMustBeAbstract()
     {
-        $container = $this->getContainerBuilder();
+        $container = $this->getContainerBuilder($fooBusId = 'messenger.bus.foo');
         $container->register('not_an_abstract_definition', UselessMiddleware::class);
-        $container->register($fooBusId = 'messenger.bus.foo', MessageBusInterface::class)->setArgument(0, array())->addTag('messenger.bus', array('name' => 'foo'));
         $container->setParameter($middlewareParameter = $fooBusId.'.middleware', array(
             array('id' => 'not_an_abstract_definition', 'arguments' => array('foo')),
         ));
@@ -413,18 +479,58 @@ class MessengerPassTest extends TestCase
         (new MessengerPass())->process($container);
     }
 
-    private function getContainerBuilder(): ContainerBuilder
+    public function testItRegistersTheDebugCommand()
+    {
+        $container = $this->getContainerBuilder($commandBusId = 'command_bus');
+        $container->register($queryBusId = 'query_bus', MessageBusInterface::class)->setArgument(0, array())->addTag('messenger.bus');
+        $container->register($emptyBus = 'empty_bus', MessageBusInterface::class)->setArgument(0, array())->addTag('messenger.bus');
+        $container->register('messenger.middleware.call_message_handler', HandleMessageMiddleware::class)
+            ->addArgument(null)
+            ->setAbstract(true)
+        ;
+
+        $container->register('console.command.messenger_debug', DebugCommand::class)->addArgument(array());
+
+        $middlewares = array(array('id' => 'call_message_handler'));
+
+        $container->setParameter($commandBusId.'.middleware', $middlewares);
+        $container->setParameter($queryBusId.'.middleware', $middlewares);
+
+        $container->register(DummyCommandHandler::class)->addTag('messenger.message_handler', array('bus' => $commandBusId));
+        $container->register(DummyQueryHandler::class)->addTag('messenger.message_handler', array('bus' => $queryBusId));
+        $container->register(MultipleBusesMessageHandler::class)
+            ->addTag('messenger.message_handler', array('bus' => $commandBusId))
+            ->addTag('messenger.message_handler', array('bus' => $queryBusId))
+        ;
+
+        (new ResolveClassPass())->process($container);
+        (new MessengerPass())->process($container);
+
+        $this->assertEquals(array(
+            $commandBusId => array(
+                DummyCommand::class => array(DummyCommandHandler::class),
+                MultipleBusesMessage::class => array(MultipleBusesMessageHandler::class),
+            ),
+            $queryBusId => array(
+                DummyQuery::class => array(DummyQueryHandler::class),
+                MultipleBusesMessage::class => array(MultipleBusesMessageHandler::class),
+            ),
+            $emptyBus => array(),
+        ), $container->getDefinition('console.command.messenger_debug')->getArgument(0));
+    }
+
+    private function getContainerBuilder(string $busId = 'message_bus'): ContainerBuilder
     {
         $container = new ContainerBuilder();
         $container->setParameter('kernel.debug', true);
 
-        $container
-            ->register('messenger.sender_locator', ServiceLocator::class)
-            ->addArgument(new Reference('service_container'))
-        ;
+        $container->register($busId, MessageBusInterface::class)->addTag('messenger.bus')->setArgument(0, array());
+        if ('message_bus' !== $busId) {
+            $container->setAlias('message_bus', $busId);
+        }
 
         $container
-            ->register('messenger.handler_resolver', ContainerHandlerLocator::class)
+            ->register('messenger.sender_locator', ServiceLocator::class)
             ->addArgument(new Reference('service_container'))
         ;
 

src/Symfony/Component/Messenger/Tests/Fixtures/DummyCommand.php

@@ -0,0 +1,16 @@
+<?php
+
+/*
+ * This file is part of the Symfony package.
+ *
+ * (c) Fabien Potencier <fabien@symfony.com>
+ *
+ * For the full copyright and license information, please view the LICENSE
+ * file that was distributed with this source code.
+ */
+
+namespace Symfony\Component\Messenger\Tests\Fixtures;
+
+class DummyCommand
+{
+}

src/Symfony/Component/Messenger/Tests/Fixtures/DummyCommandHandler.php

@@ -0,0 +1,19 @@
+<?php
+
+/*
+ * This file is part of the Symfony package.
+ *
+ * (c) Fabien Potencier <fabien@symfony.com>
+ *
+ * For the full copyright and license information, please view the LICENSE
+ * file that was distributed with this source code.
+ */
+
+namespace Symfony\Component\Messenger\Tests\Fixtures;
+
+class DummyCommandHandler
+{
+    public function __invoke(DummyCommand $command)
+    {
+    }
+}

src/Symfony/Component/Messenger/Tests/Fixtures/DummyQuery.php

@@ -0,0 +1,16 @@
+<?php
+
+/*
+ * This file is part of the Symfony package.
+ *
+ * (c) Fabien Potencier <fabien@symfony.com>
+ *
+ * For the full copyright and license information, please view the LICENSE
+ * file that was distributed with this source code.
+ */
+
+namespace Symfony\Component\Messenger\Tests\Fixtures;
+
+class DummyQuery
+{
+}

src/Symfony/Component/Messenger/Tests/Fixtures/DummyQueryHandler.php

@@ -0,0 +1,19 @@
+<?php
+
+/*
+ * This file is part of the Symfony package.
+ *
+ * (c) Fabien Potencier <fabien@symfony.com>
+ *
+ * For the full copyright and license information, please view the LICENSE
+ * file that was distributed with this source code.
+ */
+
+namespace Symfony\Component\Messenger\Tests\Fixtures;
+
+class DummyQueryHandler
+{
+    public function __invoke(DummyQuery $query)
+    {
+    }
+}

src/Symfony/Component/Messenger/Tests/Fixtures/MultipleBusesMessage.php

@@ -0,0 +1,16 @@
+<?php
+
+/*
+ * This file is part of the Symfony package.
+ *
+ * (c) Fabien Potencier <fabien@symfony.com>
+ *
+ * For the full copyright and license information, please view the LICENSE
+ * file that was distributed with this source code.
+ */
+
+namespace Symfony\Component\Messenger\Tests\Fixtures;
+
+class MultipleBusesMessage
+{
+}

src/Symfony/Component/Messenger/Tests/Fixtures/MultipleBusesMessageHandler.php

@@ -0,0 +1,19 @@
+<?php
+
+/*
+ * This file is part of the Symfony package.
+ *
+ * (c) Fabien Potencier <fabien@symfony.com>
+ *
+ * For the full copyright and license information, please view the LICENSE
+ * file that was distributed with this source code.
+ */
+
+namespace Symfony\Component\Messenger\Tests\Fixtures;
+
+class MultipleBusesMessageHandler
+{
+    public function __invoke(MultipleBusesMessage $message)
+    {
+    }
+}