Adding global retry support, events & more to messenger transport

Co-authored-by: Samuel ROZE <samuel.roze@gmail.com>

Author: weaverryan

DependencyInjection/Configuration.php

@@ -1111,6 +1111,20 @@ function ($a) {
                                         ->prototype('variable')
                                         ->end()
                                     ->end()
+                                    ->arrayNode('retry_strategy')
+                                        ->addDefaultsIfNotSet()
+                                        ->validate()
+                                            ->ifTrue(function ($v) { return null !== $v['service'] && (isset($v['max_retries']) || isset($v['delay']) || isset($v['multiplier']) || isset($v['max_delay'])); })
+                                            ->thenInvalid('"service" cannot be used along with the other retry_strategy options.')
+                                        ->end()
+                                        ->children()
+                                            ->scalarNode('service')->defaultNull()->info('Service id to override the retry strategy entirely')->end()
+                                            ->integerNode('max_retries')->defaultValue(3)->min(0)->end()
+                                            ->integerNode('delay')->defaultValue(1000)->min(0)->info('Time in ms to delay (or the initial value when multiplier is used)')->end()
+                                            ->floatNode('multiplier')->defaultValue(2)->min(1)->info('If greater than 1, delay will grow exponentially for each retry: this delay = (delay * (multiple ^ retries))')->end()
+                                            ->integerNode('max_delay')->defaultValue(0)->min(0)->info('Max time in ms that a retry should ever be delayed (0 = infinite)')->end()
+                                        ->end()
+                                    ->end()
                                 ->end()
                             ->end()
                         ->end()

DependencyInjection/FrameworkExtension.php

@@ -1653,6 +1653,7 @@ private function registerMessengerConfiguration(array $config, ContainerBuilder
         }
 
         $senderAliases = [];
+        $transportRetryReferences = [];
         foreach ($config['transports'] as $name => $transport) {
             if (0 === strpos($transport['dsn'], 'amqp://') && !$container->hasDefinition('messenger.transport.amqp.factory')) {
                 throw new LogicException('The default AMQP transport is not available. Make sure you have installed and enabled the Serializer component. Try enabling it or running "composer require symfony/serializer-pack".');
@@ -1665,6 +1666,21 @@ private function registerMessengerConfiguration(array $config, ContainerBuilder
             ;
             $container->setDefinition($transportId = 'messenger.transport.'.$name, $transportDefinition);
             $senderAliases[$name] = $transportId;
+
+            if (null !== $transport['retry_strategy']['service']) {
+                $transportRetryReferences[$name] = new Reference($transport['retry_strategy']['service']);
+            } else {
+                $retryServiceId = sprintf('messenger.retry.multiplier_retry_strategy.%s', $name);
+                $retryDefinition = new ChildDefinition('messenger.retry.abstract_multiplier_retry_strategy');
+                $retryDefinition
+                    ->replaceArgument(0, $transport['retry_strategy']['max_retries'])
+                    ->replaceArgument(1, $transport['retry_strategy']['delay'])
+                    ->replaceArgument(2, $transport['retry_strategy']['multiplier'])
+                    ->replaceArgument(3, $transport['retry_strategy']['max_delay']);
+                $container->setDefinition($retryServiceId, $retryDefinition);
+
+                $transportRetryReferences[$name] = new Reference($retryServiceId);
+            }
         }
 
         $messageToSendersMapping = [];
@@ -1686,6 +1702,9 @@ private function registerMessengerConfiguration(array $config, ContainerBuilder
             ->replaceArgument(0, $messageToSendersMapping)
             ->replaceArgument(1, $messagesToSendAndHandle)
         ;
+
+        $container->getDefinition('messenger.retry_strategy_locator')
+            ->replaceArgument(0, $transportRetryReferences);
     }
 
     private function registerCacheConfiguration(array $config, ContainerBuilder $container)

Resources/config/console.xml

@@ -82,8 +82,12 @@
             <argument type="service" id="logger" on-invalid="null" />
             <argument type="collection" /> <!-- Receiver names -->
             <argument type="collection" /> <!-- Message bus names -->
+            <argument type="service" id="messenger.retry_strategy_locator" />
+            <argument type="service" id="event_dispatcher" />
 
+            <tag name="console.command" command="messenger:consume" />
             <tag name="console.command" command="messenger:consume-messages" />
+            <tag name="monolog.logger" channel="messenger" />
         </service>
 
         <service id="console.command.messenger_debug" class="Symfony\Component\Messenger\Command\DebugCommand">

Resources/config/messenger.xml

@@ -64,5 +64,18 @@
             <tag name="messenger.transport_factory" />
             <argument type="service" id="messenger.transport.serializer" />
         </service>
+
+        <!-- retry -->
+        <service id="messenger.retry_strategy_locator">
+            <tag name="container.service_locator" />
+            <argument type="collection" />
+        </service>
+
+        <service id="messenger.retry.abstract_multiplier_retry_strategy" class="Symfony\Component\Messenger\Retry\MultiplierRetryStrategy" abstract="true">
+            <argument /> <!-- max retries -->
+            <argument /> <!-- delay ms -->
+            <argument /> <!-- multiplier -->
+            <argument /> <!-- max delay ms -->
+        </service>
     </services>
 </container>