Messenger multiple failed transports

monteiro · chalasr · commit 6ee878f98687 · 2021-03-30T16:28:46.000+02:00
diff --git a/DependencyInjection/Configuration.php b/DependencyInjection/Configuration.php
@@ -1313,6 +1313,10 @@ function ($a) {
                                         ->prototype('variable')
                                         ->end()
                                     ->end()
+                                    ->scalarNode('failure_transport')
+                                        ->defaultNull()
+                                        ->info('Transport name to send failed messages to (after all retries have failed).')
+                                    ->end()
                                     ->arrayNode('retry_strategy')
                                         ->addDefaultsIfNotSet()
                                         ->beforeNormalization()
diff --git a/DependencyInjection/FrameworkExtension.php b/DependencyInjection/FrameworkExtension.php
@@ -1915,15 +1915,38 @@ private function registerMessengerConfiguration(array $config, ContainerBuilder
             $container->setAlias('messenger.default_serializer', $config['serializer']['default_serializer']);
         }
 
+        $failureTransports = [];
+        if ($config['failure_transport']) {
+            if (!isset($config['transports'][$config['failure_transport']])) {
+                throw new LogicException(sprintf('Invalid Messenger configuration: the failure transport "%s" is not a valid transport or service id.', $config['failure_transport']));
+            }
+
+            $container->setAlias('messenger.failure_transports.default', 'messenger.transport.'.$config['failure_transport']);
+            $failureTransports[] = $config['failure_transport'];
+        }
+
+        $failureTransportsByName = [];
+        foreach ($config['transports'] as $name => $transport) {
+            if ($transport['failure_transport']) {
+                $failureTransports[] = $transport['failure_transport'];
+                $failureTransportsByName[$name] = $transport['failure_transport'];
+            } elseif ($config['failure_transport']) {
+                $failureTransportsByName[$name] = $config['failure_transport'];
+            }
+        }
+
         $senderAliases = [];
         $transportRetryReferences = [];
         foreach ($config['transports'] as $name => $transport) {
             $serializerId = $transport['serializer'] ?? 'messenger.default_serializer';
-
             $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])
+                ->addTag('messenger.receiver', [
+                        'alias' => $name,
+                        'is_failure_transport' => \in_array($name, $failureTransports),
+                    ]
+                )
             ;
             $container->setDefinition($transportId = 'messenger.transport.'.$name, $transportDefinition);
             $senderAliases[$name] = $transportId;
@@ -1954,6 +1977,18 @@ private function registerMessengerConfiguration(array $config, ContainerBuilder
             $senderReferences[$serviceId] = new Reference($serviceId);
         }
 
+        foreach ($config['transports'] as $name => $transport) {
+            if ($transport['failure_transport']) {
+                if (!isset($senderReferences[$transport['failure_transport']])) {
+                    throw new LogicException(sprintf('Invalid Messenger configuration: the failure transport "%s" is not a valid transport or service id.', $transport['failure_transport']));
+                }
+            }
+        }
+
+        $failureTransportReferencesByTransportName = array_map(function ($failureTransportName) use ($senderReferences) {
+            return $senderReferences[$failureTransportName];
+        }, $failureTransportsByName);
+
         $messageToSendersMapping = [];
         foreach ($config['routing'] as $message => $messageConfiguration) {
             if ('*' !== $message && !class_exists($message) && !interface_exists($message, false)) {
@@ -1984,19 +2019,17 @@ private function registerMessengerConfiguration(array $config, ContainerBuilder
         $container->getDefinition('messenger.retry_strategy_locator')
             ->replaceArgument(0, $transportRetryReferences);
 
-        if ($config['failure_transport']) {
-            if (!isset($senderReferences[$config['failure_transport']])) {
-                throw new LogicException(sprintf('Invalid Messenger configuration: the failure transport "%s" is not a valid transport or service id.', $config['failure_transport']));
-            }
-
-            $container->getDefinition('messenger.failure.send_failed_message_to_failure_transport_listener')
-                ->replaceArgument(0, $senderReferences[$config['failure_transport']]);
+        if (\count($failureTransports) > 0) {
             $container->getDefinition('console.command.messenger_failed_messages_retry')
                 ->replaceArgument(0, $config['failure_transport']);
             $container->getDefinition('console.command.messenger_failed_messages_show')
                 ->replaceArgument(0, $config['failure_transport']);
             $container->getDefinition('console.command.messenger_failed_messages_remove')
                 ->replaceArgument(0, $config['failure_transport']);
+
+            $failureTransportsByTransportNameServiceLocator = ServiceLocatorTagPass::register($container, $failureTransportReferencesByTransportName);
+            $container->getDefinition('messenger.failure.send_failed_message_to_failure_transport_listener')
+                ->replaceArgument(0, $failureTransportsByTransportNameServiceLocator);
         } else {
             $container->removeDefinition('messenger.failure.send_failed_message_to_failure_transport_listener');
             $container->removeDefinition('console.command.messenger_failed_messages_retry');
diff --git a/Resources/config/console.php b/Resources/config/console.php
@@ -165,8 +165,8 @@
 
         ->set('console.command.messenger_failed_messages_retry', FailedMessagesRetryCommand::class)
             ->args([
-                abstract_arg('Receiver name'),
-                abstract_arg('Receiver'),
+                abstract_arg('Default failure receiver name'),
+                abstract_arg('Receivers'),
                 service('messenger.routable_message_bus'),
                 service('event_dispatcher'),
                 service('logger'),
@@ -175,15 +175,15 @@
 
         ->set('console.command.messenger_failed_messages_show', FailedMessagesShowCommand::class)
             ->args([
-                abstract_arg('Receiver name'),
-                abstract_arg('Receiver'),
+                abstract_arg('Default failure receiver name'),
+                abstract_arg('Receivers'),
             ])
             ->tag('console.command')
 
         ->set('console.command.messenger_failed_messages_remove', FailedMessagesRemoveCommand::class)
             ->args([
-                abstract_arg('Receiver name'),
-                abstract_arg('Receiver'),
+                abstract_arg('Default failure receiver name'),
+                abstract_arg('Receivers'),
             ])
             ->tag('console.command')
 
diff --git a/Resources/config/messenger.php b/Resources/config/messenger.php
@@ -172,7 +172,7 @@
 
         ->set('messenger.failure.send_failed_message_to_failure_transport_listener', SendFailedMessageToFailureTransportListener::class)
             ->args([
-                abstract_arg('failure transport'),
+                abstract_arg('failure transports'),
                 service('logger')->ignoreOnInvalid(),
             ])
             ->tag('kernel.event_subscriber')
diff --git a/Resources/config/schema/symfony-1.0.xsd b/Resources/config/schema/symfony-1.0.xsd
@@ -491,6 +491,7 @@
         <xsd:attribute name="name" type="xsd:string" />
         <xsd:attribute name="serializer" type="xsd:string" />
         <xsd:attribute name="dsn" type="xsd:string" />
+        <xsd:attribute name="failure-transport" type="xsd:string" />
     </xsd:complexType>
 
     <xsd:complexType name="messenger_retry_strategy">
diff --git a/Tests/DependencyInjection/Fixtures/php/messenger_multiple_failure_transports.php b/Tests/DependencyInjection/Fixtures/php/messenger_multiple_failure_transports.php
@@ -0,0 +1,19 @@
+<?php
+
+$container->loadFromExtension('framework', [
+    'messenger' => [
+        'transports' => [
+            'transport_1' => [
+                'dsn' => 'null://',
+                'failure_transport' => 'failure_transport_1'
+            ],
+            'transport_2' => 'null://',
+            'transport_3' => [
+                'dsn' => 'null://',
+                'failure_transport' => 'failure_transport_3'
+            ],
+            'failure_transport_1' => 'null://',
+            'failure_transport_3' => 'null://'
+        ],
+    ],
+]);
diff --git a/Tests/DependencyInjection/Fixtures/php/messenger_multiple_failure_transports_global.php b/Tests/DependencyInjection/Fixtures/php/messenger_multiple_failure_transports_global.php
@@ -0,0 +1,21 @@
+<?php
+
+$container->loadFromExtension('framework', [
+    'messenger' => [
+        'failure_transport' => 'failure_transport_global',
+        'transports' => [
+            'transport_1' => [
+                'dsn' => 'null://',
+                'failure_transport' => 'failure_transport_1'
+            ],
+            'transport_2' => 'null://',
+            'transport_3' => [
+                'dsn' => 'null://',
+                'failure_transport' => 'failure_transport_3'
+            ],
+            'failure_transport_global' => 'null://',
+            'failure_transport_1' => 'null://',
+            'failure_transport_3' => 'null://',
+        ],
+    ],
+]);
diff --git a/Tests/DependencyInjection/Fixtures/xml/messenger_multiple_failure_transports.xml b/Tests/DependencyInjection/Fixtures/xml/messenger_multiple_failure_transports.xml
@@ -0,0 +1,17 @@
+<?xml version="1.0" encoding="utf-8" ?>
+<container xmlns="http://symfony.com/schema/dic/services"
+    xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+    xmlns:framework="http://symfony.com/schema/dic/symfony"
+    xsi:schemaLocation="http://symfony.com/schema/dic/services https://symfony.com/schema/dic/services/services-1.0.xsd
+        http://symfony.com/schema/dic/symfony https://symfony.com/schema/dic/symfony/symfony-1.0.xsd">
+
+    <framework:config>
+        <framework:messenger>
+            <framework:transport name="transport_1" dsn="null://" failure-transport="failure_transport_1" />
+            <framework:transport name="transport_2" dsn="null://" />
+            <framework:transport name="transport_3" dsn="null://" failure-transport="failure_transport_3" />
+            <framework:transport name="failure_transport_1" dsn="null://" />
+            <framework:transport name="failure_transport_3" dsn="null://" />
+        </framework:messenger>
+    </framework:config>
+</container>
diff --git a/Tests/DependencyInjection/Fixtures/xml/messenger_multiple_failure_transports_global.xml b/Tests/DependencyInjection/Fixtures/xml/messenger_multiple_failure_transports_global.xml
@@ -0,0 +1,18 @@
+<?xml version="1.0" encoding="utf-8" ?>
+<container xmlns="http://symfony.com/schema/dic/services"
+    xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+    xmlns:framework="http://symfony.com/schema/dic/symfony"
+    xsi:schemaLocation="http://symfony.com/schema/dic/services https://symfony.com/schema/dic/services/services-1.0.xsd
+        http://symfony.com/schema/dic/symfony https://symfony.com/schema/dic/symfony/symfony-1.0.xsd">
+
+    <framework:config>
+        <framework:messenger failure-transport="failure_transport_global">
+            <framework:transport name="transport_1" dsn="null://" failure-transport="failure_transport_1" />
+            <framework:transport name="transport_2" dsn="null://" />
+            <framework:transport name="transport_3" dsn="null://" failure-transport="failure_transport_3" />
+            <framework:transport name="failure_transport_global" dsn="null://" />
+            <framework:transport name="failure_transport_1" dsn="null://" />
+            <framework:transport name="failure_transport_3" dsn="null://" />
+        </framework:messenger>
+    </framework:config>
+</container>
diff --git a/Tests/DependencyInjection/Fixtures/yml/messenger_multiple_failure_transports.yml b/Tests/DependencyInjection/Fixtures/yml/messenger_multiple_failure_transports.yml
@@ -0,0 +1,12 @@
+framework:
+    messenger:
+        transports:
+            transport_1:
+                dsn: 'null://'
+                failure_transport: failure_transport_1
+            transport_2: 'null://'
+            transport_3:
+                dsn: 'null://'
+                failure_transport: failure_transport_3
+            failure_transport_1: 'null://'
+            failure_transport_3: 'null://'
diff --git a/Tests/DependencyInjection/Fixtures/yml/messenger_multiple_failure_transports_global.yml b/Tests/DependencyInjection/Fixtures/yml/messenger_multiple_failure_transports_global.yml
@@ -0,0 +1,14 @@
+framework:
+    messenger:
+        failure_transport: failure_transport_global
+        transports:
+            transport_1:
+                dsn: 'null://'
+                failure_transport: failure_transport_1
+            transport_2: 'null://'
+            transport_3:
+                dsn: 'null://'
+                failure_transport: failure_transport_3
+            failure_transport_global: 'null://'
+            failure_transport_1: 'null://'
+            failure_transport_3: 'null://'
diff --git a/Tests/DependencyInjection/FrameworkExtensionTest.php b/Tests/DependencyInjection/FrameworkExtensionTest.php
@@ -32,6 +32,7 @@
 use Symfony\Component\Cache\Adapter\RedisTagAwareAdapter;
 use Symfony\Component\Cache\DependencyInjection\CachePoolPass;
 use Symfony\Component\Config\Definition\Exception\InvalidConfigurationException;
+use Symfony\Component\DependencyInjection\Argument\ServiceClosureArgument;
 use Symfony\Component\DependencyInjection\ChildDefinition;
 use Symfony\Component\DependencyInjection\Compiler\CompilerPassInterface;
 use Symfony\Component\DependencyInjection\Compiler\ResolveInstanceofConditionalsPass;
@@ -710,12 +711,92 @@ public function testMessenger()
         $this->assertSame(TransportFactory::class, $container->getDefinition('messenger.transport_factory')->getClass());
     }
 
+    public function testMessengerMultipleFailureTransports()
+    {
+        $container = $this->createContainerFromFile('messenger_multiple_failure_transports');
+
+        $failureTransport1Definition = $container->getDefinition('messenger.transport.failure_transport_1');
+        $failureTransport1Tags = $failureTransport1Definition->getTag('messenger.receiver')[0];
+
+        $this->assertEquals([
+            'alias' => 'failure_transport_1',
+            'is_failure_transport' => true,
+        ], $failureTransport1Tags);
+
+        $failureTransport3Definition = $container->getDefinition('messenger.transport.failure_transport_3');
+        $failureTransport3Tags = $failureTransport3Definition->getTag('messenger.receiver')[0];
+
+        $this->assertEquals([
+            'alias' => 'failure_transport_3',
+            'is_failure_transport' => true,
+        ], $failureTransport3Tags);
+
+        // transport 2 exists but does not appear in the mapping
+        $this->assertFalse($container->hasDefinition('messenger.transport.failure_transport_2'));
+
+        $failureTransportsByTransportNameServiceLocator = $container->getDefinition('messenger.failure.send_failed_message_to_failure_transport_listener')->getArgument(0);
+        $failureTransports = $container->getDefinition((string) $failureTransportsByTransportNameServiceLocator)->getArgument(0);
+        $expectedTransportsByFailureTransports = [
+            'transport_1' => new Reference('messenger.transport.failure_transport_1'),
+            'transport_3' => new Reference('messenger.transport.failure_transport_3'),
+        ];
+
+        $failureTransportsReferences = array_map(function (ServiceClosureArgument $serviceClosureArgument) {
+            $values = $serviceClosureArgument->getValues();
+
+            return array_shift($values);
+        }, $failureTransports);
+        $this->assertEquals($expectedTransportsByFailureTransports, $failureTransportsReferences);
+    }
+
+    public function testMessengerMultipleFailureTransportsWithGlobalFailureTransport()
+    {
+        $container = $this->createContainerFromFile('messenger_multiple_failure_transports_global');
+
+        $this->assertEquals('messenger.transport.failure_transport_global', (string) $container->getAlias('messenger.failure_transports.default'));
+
+        $failureTransport1Definition = $container->getDefinition('messenger.transport.failure_transport_1');
+        $failureTransport1Tags = $failureTransport1Definition->getTag('messenger.receiver')[0];
+
+        $this->assertEquals([
+            'alias' => 'failure_transport_1',
+            'is_failure_transport' => true,
+        ], $failureTransport1Tags);
+
+        $failureTransport3Definition = $container->getDefinition('messenger.transport.failure_transport_3');
+        $failureTransport3Tags = $failureTransport3Definition->getTag('messenger.receiver')[0];
+
+        $this->assertEquals([
+            'alias' => 'failure_transport_3',
+            'is_failure_transport' => true,
+        ], $failureTransport3Tags);
+
+        $failureTransportsByTransportNameServiceLocator = $container->getDefinition('messenger.failure.send_failed_message_to_failure_transport_listener')->getArgument(0);
+        $failureTransports = $container->getDefinition((string) $failureTransportsByTransportNameServiceLocator)->getArgument(0);
+        $expectedTransportsByFailureTransports = [
+            'failure_transport_1' => new Reference('messenger.transport.failure_transport_global'),
+            'failure_transport_3' => new Reference('messenger.transport.failure_transport_global'),
+            'failure_transport_global' => new Reference('messenger.transport.failure_transport_global'),
+            'transport_1' => new Reference('messenger.transport.failure_transport_1'),
+            'transport_2' => new Reference('messenger.transport.failure_transport_global'),
+            'transport_3' => new Reference('messenger.transport.failure_transport_3'),
+        ];
+
+        $failureTransportsReferences = array_map(function (ServiceClosureArgument $serviceClosureArgument) {
+            $values = $serviceClosureArgument->getValues();
+
+            return array_shift($values);
+        }, $failureTransports);
+        $this->assertEquals($expectedTransportsByFailureTransports, $failureTransportsReferences);
+    }
+
     public function testMessengerTransports()
     {
         $container = $this->createContainerFromFile('messenger_transports');
         $this->assertTrue($container->hasDefinition('messenger.transport.default'));
         $this->assertTrue($container->getDefinition('messenger.transport.default')->hasTag('messenger.receiver'));
-        $this->assertEquals([['alias' => 'default']], $container->getDefinition('messenger.transport.default')->getTag('messenger.receiver'));
+        $this->assertEquals([
+            ['alias' => 'default', 'is_failure_transport' => false], ], $container->getDefinition('messenger.transport.default')->getTag('messenger.receiver'));
         $transportArguments = $container->getDefinition('messenger.transport.default')->getArguments();
         $this->assertEquals(new Reference('messenger.default_serializer'), $transportArguments[2]);
 
@@ -756,7 +837,22 @@ public function testMessengerTransports()
         $this->assertSame(3, $container->getDefinition('messenger.retry.multiplier_retry_strategy.customised')->getArgument(2));
         $this->assertSame(100, $container->getDefinition('messenger.retry.multiplier_retry_strategy.customised')->getArgument(3));
 
-        $this->assertEquals(new Reference('messenger.transport.failed'), $container->getDefinition('messenger.failure.send_failed_message_to_failure_transport_listener')->getArgument(0));
+        $failureTransportsByTransportNameServiceLocator = $container->getDefinition('messenger.failure.send_failed_message_to_failure_transport_listener')->getArgument(0);
+        $failureTransports = $container->getDefinition((string) $failureTransportsByTransportNameServiceLocator)->getArgument(0);
+        $expectedTransportsByFailureTransports = [
+            'beanstalkd' => new Reference('messenger.transport.failed'),
+            'customised' => new Reference('messenger.transport.failed'),
+            'default' => new Reference('messenger.transport.failed'),
+            'failed' => new Reference('messenger.transport.failed'),
+            'redis' => new Reference('messenger.transport.failed'),
+        ];
+
+        $failureTransportsReferences = array_map(function (ServiceClosureArgument $serviceClosureArgument) {
+            $values = $serviceClosureArgument->getValues();
+
+            return array_shift($values);
+        }, $failureTransports);
+        $this->assertEquals($expectedTransportsByFailureTransports, $failureTransportsReferences);
     }
 
     public function testMessengerRouting()
