[Mesenger] Add support for resetting container services after each messenger message.

Without this patch, services are not resetted. For example Monolog
Finger Cross handler is never reset nor flushed. So if the first
message trigger and "error" level message, all others message will log
and overflow the buffer.

Usage with framework:

```yaml
framework:
    messenger:
        transports:
            async:
                dsn: '%env(MESSENGER_TRANSPORT_DSN)%'
                reset_on_message: true
            failed: 'doctrine://default?queue_name=failed'
            sync: 'sync://'
```

Author: lyrixx

CHANGELOG.md

@@ -10,6 +10,7 @@ CHANGELOG
  * Deprecate `get()`, `has()`, `getDoctrine()`, and `dispatchMessage()` in `AbstractController`, use method/constructor injection instead
  * Add `MicroKernelTrait::getBundlesPath` method to get bundles config path
  * Deprecate the `cache.adapter.doctrine` service
+ * Add support for resetting container services after each messenger message.
 
 5.3
 ---

DependencyInjection/Configuration.php

@@ -1333,6 +1333,10 @@ function ($a) {
                                 ->fixXmlConfig('option')
                                 ->children()
                                     ->scalarNode('dsn')->end()
+                                    ->booleanNode('reset_on_message')
+                                        ->defaultFalse()
+                                        ->info('Reset container services after each message. Turn it on when the transport is async and run in a worker.')
+                                    ->end()
                                     ->scalarNode('serializer')->defaultNull()->info('Service id of a custom serializer to use.')->end()
                                     ->arrayNode('options')
                                         ->normalizeKeys(false)

DependencyInjection/FrameworkExtension.php

@@ -2013,6 +2013,7 @@ private function registerMessengerConfiguration(array $config, ContainerBuilder
 
         $senderAliases = [];
         $transportRetryReferences = [];
+        $transportNamesForResetServices = [];
         foreach ($config['transports'] as $name => $transport) {
             $serializerId = $transport['serializer'] ?? 'messenger.default_serializer';
             $transportDefinition = (new Definition(TransportInterface::class))
@@ -2041,6 +2042,18 @@ private function registerMessengerConfiguration(array $config, ContainerBuilder
 
                 $transportRetryReferences[$name] = new Reference($retryServiceId);
             }
+            if ($transport['reset_on_message']) {
+                $transportNamesForResetServices[] = $name;
+            }
+        }
+
+        if ($transportNamesForResetServices) {
+            $container
+                ->getDefinition('messenger.listener.reset_services')
+                ->replaceArgument(1, $transportNamesForResetServices)
+            ;
+        } else {
+            $container->removeDefinition('messenger.listener.reset_services');
         }
 
         $senderReferences = [];

Resources/config/messenger.php

@@ -18,6 +18,7 @@
 use Symfony\Component\Messenger\Bridge\Redis\Transport\RedisTransportFactory;
 use Symfony\Component\Messenger\EventListener\AddErrorDetailsStampListener;
 use Symfony\Component\Messenger\EventListener\DispatchPcntlSignalListener;
+use Symfony\Component\Messenger\EventListener\ResetServicesListener;
 use Symfony\Component\Messenger\EventListener\SendFailedMessageForRetryListener;
 use Symfony\Component\Messenger\EventListener\SendFailedMessageToFailureTransportListener;
 use Symfony\Component\Messenger\EventListener\StopWorkerOnCustomStopExceptionListener;
@@ -195,6 +196,12 @@
             ->tag('kernel.event_subscriber')
 
         ->set('messenger.listener.stop_worker_on_stop_exception_listener', StopWorkerOnCustomStopExceptionListener::class)
+
+        ->set('messenger.listener.reset_services', ResetServicesListener::class)
+            ->args([
+                service('services_resetter'),
+                abstract_arg('receivers names'),
+            ])
             ->tag('kernel.event_subscriber')
 
         ->set('messenger.routable_message_bus', RoutableMessageBus::class)

Resources/config/schema/symfony-1.0.xsd

@@ -505,6 +505,7 @@
         <xsd:attribute name="serializer" type="xsd:string" />
         <xsd:attribute name="dsn" type="xsd:string" />
         <xsd:attribute name="failure-transport" type="xsd:string" />
+        <xsd:attribute name="reset-on-message" type="xsd:boolean" />
     </xsd:complexType>
 
     <xsd:complexType name="messenger_retry_strategy">

Tests/DependencyInjection/Fixtures/php/messenger_transports.php

@@ -11,6 +11,7 @@
             'default' => 'amqp://localhost/%2f/messages',
             'customised' => [
                 'dsn' => 'amqp://localhost/%2f/messages?exchange_name=exchange_name',
+                'reset_on_message' =>  true,
                 'options' => ['queue' => ['name' => 'Queue']],
                 'serializer' => 'messenger.transport.native_php_serializer',
                 'retry_strategy' => [

Tests/DependencyInjection/Fixtures/xml/messenger_transports.xml

@@ -10,7 +10,7 @@
         <framework:messenger failure-transport="failed">
             <framework:serializer default-serializer="messenger.transport.symfony_serializer" />
             <framework:transport name="default" dsn="amqp://localhost/%2f/messages" />
-            <framework:transport name="customised" dsn="amqp://localhost/%2f/messages?exchange_name=exchange_name" serializer="messenger.transport.native_php_serializer">
+            <framework:transport name="customised" dsn="amqp://localhost/%2f/messages?exchange_name=exchange_name" serializer="messenger.transport.native_php_serializer" reset-on-message="true">
                 <framework:options>
                     <framework:queue>
                         <framework:name>Queue</framework:name>

Tests/DependencyInjection/Fixtures/yml/messenger_transports.yml

@@ -8,6 +8,7 @@ framework:
             default: 'amqp://localhost/%2f/messages'
             customised:
                 dsn: 'amqp://localhost/%2f/messages?exchange_name=exchange_name'
+                reset_on_message: true
                 options:
                     queue:
                         name: Queue

Tests/DependencyInjection/FrameworkExtensionTest.php

@@ -723,6 +723,7 @@ public function testMessenger()
         $this->assertTrue($container->hasDefinition('messenger.transport.redis.factory'));
         $this->assertTrue($container->hasDefinition('messenger.transport_factory'));
         $this->assertSame(TransportFactory::class, $container->getDefinition('messenger.transport_factory')->getClass());
+        $this->assertFalse($container->hasDefinition('messenger.listener.reset_services'));
     }
 
     public function testMessengerMultipleFailureTransports()
@@ -867,6 +868,9 @@ public function testMessengerTransports()
             return array_shift($values);
         }, $failureTransports);
         $this->assertEquals($expectedTransportsByFailureTransports, $failureTransportsReferences);
+
+        $this->assertTrue($container->hasDefinition('messenger.listener.reset_services'));
+        $this->assertSame(['customised'], $container->getDefinition('messenger.listener.reset_services')->getArgument(1));
     }
 
     public function testMessengerRouting()