[Messenger] Add a command to setup transports

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

@@ -79,6 +79,13 @@
             <tag name="console.command" command="messenger:consume-messages" />
         </service>
 
+        <service id="console.command.messenger_setup_transports" class="Symfony\Component\Messenger\Command\SetupTransportsCommand">
+            <argument type="service" id="messenger.receiver_locator" />
+            <argument type="collection" /> <!-- Receiver names -->
+
+            <tag name="console.command" command="messenger:setup-transports" />
+        </service>
+
         <service id="console.command.messenger_debug" class="Symfony\Component\Messenger\Command\DebugCommand">
             <argument type="collection" /> <!-- Message to handlers mapping -->
             <tag name="console.command" command="debug:messenger" />

src/Symfony/Component/Messenger/CHANGELOG.md

@@ -1,6 +1,11 @@
 CHANGELOG
 =========
 
+4.3.0
+-----
+
+* Added a `SetupTransportsCommand` command to setup the transports
+
 4.2.0
 -----
 

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

@@ -0,0 +1,51 @@
+<?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\Command;
+
+use Symfony\Component\Console\Command\Command;
+use Symfony\Component\Console\Input\InputInterface;
+use Symfony\Component\Console\Output\OutputInterface;
+use Symfony\Component\Console\Style\SymfonyStyle;
+use Symfony\Component\DependencyInjection\ContainerInterface;
+use Symfony\Component\Messenger\Transport\SetupableTransportInterface;
+
+/**
+ * @author Vincent Touzet <vincent.touzet@gmail.com>
+ */
+class SetupTransportsCommand extends Command
+{
+    protected static $defaultName = 'messenger:setup-transports';
+
+    private $senderLocator;
+    private $receiverNames;
+
+    public function __construct(ContainerInterface $senderLocator, array $receiverNames = array())
+    {
+        $this->senderLocator = $senderLocator;
+        $this->receiverNames = $receiverNames;
+
+        parent::__construct();
+    }
+
+    protected function execute(InputInterface $input, OutputInterface $output)
+    {
+        $io = new SymfonyStyle($input, $output);
+
+        foreach ($this->receiverNames as $id => $receiverName) {
+            $receiver = $this->senderLocator->get($id);
+            if ($receiver instanceof SetupableTransportInterface) {
+                $receiver->setup();
+                $io->success("Setup $id");
+            }
+        }
+    }
+}

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

@@ -257,6 +257,15 @@ private function registerReceivers(ContainerBuilder $container, array $busIds)
                 ->replaceArgument(4, $busIds);
         }
 
+        if ($container->hasDefinition('console.command.messenger_setup_transports')) {
+            $receiverNames = array();
+            foreach ($receiverMapping as $name => $reference) {
+                $receiverNames[(string) $reference] = $name;
+            }
+            $container->getDefinition('console.command.messenger_setup_transports')
+                ->replaceArgument(1, array_values($receiverNames));
+        }
+
         $container->getDefinition('messenger.receiver_locator')->replaceArgument(0, $receiverMapping);
     }
 

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

@@ -20,6 +20,7 @@
 use Symfony\Component\DependencyInjection\ServiceLocator;
 use Symfony\Component\Messenger\Command\ConsumeMessagesCommand;
 use Symfony\Component\Messenger\Command\DebugCommand;
+use Symfony\Component\Messenger\Command\SetupTransportsCommand;
 use Symfony\Component\Messenger\DataCollector\MessengerDataCollector;
 use Symfony\Component\Messenger\DependencyInjection\MessengerPass;
 use Symfony\Component\Messenger\Envelope;
@@ -268,6 +269,22 @@ public function testItRegistersMultipleReceiversAndSetsTheReceiverNamesOnTheComm
         $this->assertSame(array('message_bus'), $container->getDefinition('console.command.messenger_consume_messages')->getArgument(4));
     }
 
+    public function testItSetsTheReceiverNamesOnTheSetupTransportsCommand()
+    {
+        $container = $this->getContainerBuilder();
+        $container->register('console.command.messenger_setup_transports', SetupTransportsCommand::class)->setArguments(array(
+            new Reference('messenger.receiver_locator'),
+            null,
+        ));
+
+        $container->register(AmqpReceiver::class, AmqpReceiver::class)->addTag('messenger.receiver', array('alias' => 'amqp'));
+        $container->register(DummyReceiver::class, DummyReceiver::class)->addTag('messenger.receiver', array('alias' => 'dummy'));
+
+        (new MessengerPass())->process($container);
+
+        $this->assertSame(array('amqp', 'dummy'), $container->getDefinition('console.command.messenger_setup_transports')->getArgument(1));
+    }
+
     public function testItShouldNotThrowIfGeneratorIsReturnedInsteadOfArray()
     {
         $container = $this->getContainerBuilder($busId = 'message_bus');

src/Symfony/Component/Messenger/Transport/AmqpExt/AmqpTransport.php

@@ -14,14 +14,15 @@
 use Symfony\Component\Messenger\Envelope;
 use Symfony\Component\Messenger\Transport\Serialization\Serializer;
 use Symfony\Component\Messenger\Transport\Serialization\SerializerInterface;
+use Symfony\Component\Messenger\Transport\SetupableTransportInterface;
 use Symfony\Component\Messenger\Transport\TransportInterface;
 
 /**
  * @author Nicolas Grekas <p@tchwork.com>
  *
  * @experimental in 4.2
  */
-class AmqpTransport implements TransportInterface
+class AmqpTransport implements TransportInterface, SetupableTransportInterface
 {
     private $serializer;
     private $connection;
@@ -58,6 +59,14 @@ public function send(Envelope $envelope): Envelope
         return ($this->sender ?? $this->getSender())->send($envelope);
     }
 
+    /**
+     * {@inheritdoc}
+     */
+    public function setup(): void
+    {
+        $this->connection->setup();
+    }
+
     private function getReceiver()
     {
         return $this->receiver = new AmqpReceiver($this->connection, $this->serializer);

src/Symfony/Component/Messenger/Transport/SetupableTransportInterface.php

@@ -0,0 +1,23 @@
+<?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\Transport;
+
+/**
+ * @author Vincent Touzet <vincent.touzet@gmail.com>
+ */
+interface SetupableTransportInterface
+{
+    /**
+     * Setup the transport
+     */
+    public function setup(): void;
+}