[Messenger] extract worker logic to listener and sent messages for retry and failure directly to transport instead of redispatching on the bus

Author: Tobion

DependencyInjection/FrameworkExtension.php

@@ -1810,20 +1810,29 @@ private function registerMessengerConfiguration(array $config, ContainerBuilder
             $messageToSendersMapping[$message] = $messageConfiguration['senders'];
         }
 
+        $sendersServiceLocator = ServiceLocatorTagPass::register($container, $senderReferences);
+
         $container->getDefinition('messenger.senders_locator')
             ->replaceArgument(0, $messageToSendersMapping)
-            ->replaceArgument(1, ServiceLocatorTagPass::register($container, $senderReferences))
+            ->replaceArgument(1, $sendersServiceLocator)
+        ;
+
+        $container->getDefinition('messenger.retry.send_failed_message_for_retry_listener')
+            ->replaceArgument(0, $sendersServiceLocator)
         ;
 
         $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(1, $config['failure_transport']);
+                ->replaceArgument(0, $senderReferences[$config['failure_transport']]);
             $container->getDefinition('console.command.messenger_failed_messages_retry')
-                ->replaceArgument(0, $config['failure_transport'])
-                ->replaceArgument(4, $transportRetryReferences[$config['failure_transport']] ?? null);
+                ->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')

Resources/config/console.xml

@@ -86,7 +86,6 @@
             <argument type="service" id="messenger.receiver_locator" />
             <argument type="service" id="logger" on-invalid="null" />
             <argument type="collection" /> <!-- Receiver names -->
-            <argument type="service" id="messenger.retry_strategy_locator" />
             <argument type="service" id="event_dispatcher" />
             <call method="setCachePoolForRestartSignal">
                 <argument type="service" id="cache.messenger.restart_workers_signal" />
@@ -116,25 +115,24 @@
 
         <service id="console.command.messenger_failed_messages_retry" class="Symfony\Component\Messenger\Command\FailedMessagesRetryCommand">
             <argument /> <!-- Receiver name -->
-            <argument /> <!-- Receiver locator -->
+            <argument /> <!-- Receiver -->
             <argument type="service" id="messenger.routable_message_bus" />
             <argument type="service" id="event_dispatcher" />
-            <argument /> <!-- Retry strategy -->
             <argument type="service" id="logger" />
 
             <tag name="console.command" command="messenger:failed:retry" />
         </service>
 
         <service id="console.command.messenger_failed_messages_show" class="Symfony\Component\Messenger\Command\FailedMessagesShowCommand">
             <argument /> <!-- Receiver name -->
-            <argument /> <!-- Receiver locator -->
+            <argument /> <!-- Receiver -->
 
             <tag name="console.command" command="messenger:failed:show" />
         </service>
 
         <service id="console.command.messenger_failed_messages_remove" class="Symfony\Component\Messenger\Command\FailedMessagesRemoveCommand">
             <argument /> <!-- Receiver name -->
-            <argument /> <!-- Receiver locator -->
+            <argument /> <!-- Receiver -->
 
             <tag name="console.command" command="messenger:failed:remove" />
         </service>

Resources/config/messenger.xml

@@ -10,7 +10,7 @@
         <!-- Asynchronous -->
         <service id="messenger.senders_locator" class="Symfony\Component\Messenger\Transport\Sender\SendersLocator">
             <argument type="collection" /> <!-- Per message senders map -->
-            <argument /> <!-- senders locator -->
+            <argument /> <!-- senders service locator -->
         </service>
         <service id="messenger.middleware.send_message" class="Symfony\Component\Messenger\Middleware\SendMessageMiddleware">
             <tag name="monolog.logger" channel="messenger" />
@@ -98,12 +98,19 @@
             <argument /> <!-- max delay ms -->
         </service>
 
+        <service id="messenger.retry.send_failed_message_for_retry_listener" class="Symfony\Component\Messenger\EventListener\SendFailedMessageForRetryListener">
+            <tag name="kernel.event_subscriber" />
+            <tag name="monolog.logger" channel="messenger" />
+            <argument /> <!-- senders service locator -->
+            <argument type="service" id="messenger.retry_strategy_locator" />
+            <argument type="service" id="logger" on-invalid="ignore" />
+        </service>
+
         <!-- failed handling -->
         <service id="messenger.failure.send_failed_message_to_failure_transport_listener" class="Symfony\Component\Messenger\EventListener\SendFailedMessageToFailureTransportListener">
             <tag name="kernel.event_subscriber" />
             <tag name="monolog.logger" channel="messenger" />
-            <argument type="service" id="messenger.routable_message_bus" />
-            <argument /> <!-- Failure transport name -->
+            <argument /> <!-- Failure transport -->
             <argument type="service" id="logger" on-invalid="ignore" />
         </service>
 

composer.json

@@ -45,7 +45,7 @@
         "symfony/http-client": "^4.4|^5.0",
         "symfony/lock": "^4.4|^5.0",
         "symfony/mailer": "^4.4|^5.0",
-        "symfony/messenger": "^4.3.6|^5.0",
+        "symfony/messenger": "^4.4|^5.0",
         "symfony/mime": "^4.4|^5.0",
         "symfony/process": "^3.4|^4.0|^5.0",
         "symfony/security-csrf": "^3.4|^4.0|^5.0",
@@ -77,7 +77,7 @@
         "symfony/form": "<4.3",
         "symfony/lock": "<4.4",
         "symfony/mailer": "<4.4",
-        "symfony/messenger": "<4.3.6",
+        "symfony/messenger": "<4.4",
         "symfony/mime": "<4.4",
         "symfony/property-info": "<3.4",
         "symfony/security-bundle": "<4.4",