[Mesenger] Add support for resetting container services after each me…

…ssenger 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://'
```

src/Symfony/Bundle/FrameworkBundle/CHANGELOG.md

@@ -8,6 +8,7 @@ CHANGELOG
  * Deprecate the `AdapterInterface` autowiring alias, use `CacheItemPoolInterface` instead
  * Deprecate the public `profiler` service to private
  * Deprecate `get()`, `has()`, `getDoctrine()`, and `dispatchMessage()` in `AbstractController`, use method/constructor injection instead
+ * Add support for resetting container services after each messenger message.
 
 5.3
 ---

src/Symfony/Bundle/FrameworkBundle/DependencyInjection/Configuration.php

@@ -1333,6 +1333,7 @@ function ($a) {
                                 ->fixXmlConfig('option')
                                 ->children()
                                     ->scalarNode('dsn')->end()
+                                    ->booleanNode('reset_on_message')->defaultFalse()->info('Reset the container services after each message.')->end()
                                     ->scalarNode('serializer')->defaultNull()->info('Service id of a custom serializer to use.')->end()
                                     ->arrayNode('options')
                                         ->normalizeKeys(false)

src/Symfony/Bundle/FrameworkBundle/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 = [];

src/Symfony/Bundle/FrameworkBundle/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)

src/Symfony/Bundle/FrameworkBundle/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">

src/Symfony/Bundle/FrameworkBundle/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' => [

src/Symfony/Bundle/FrameworkBundle/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>

src/Symfony/Bundle/FrameworkBundle/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

src/Symfony/Bundle/FrameworkBundle/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()

src/Symfony/Component/Messenger/CHANGELOG.md

@@ -5,6 +5,7 @@ CHANGELOG
 ---
 
  * Add `StopWorkerExceptionInterface` and its implementation `StopWorkerException` to stop the worker.
+ * Add support for resetting container services after each messenger message.
 
 5.3
 ---

src/Symfony/Component/Messenger/EventListener/ResetServicesListener.php

@@ -0,0 +1,50 @@
+<?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\EventListener;
+
+use Symfony\Component\EventDispatcher\EventSubscriberInterface;
+use Symfony\Component\HttpKernel\DependencyInjection\ServicesResetter;
+use Symfony\Component\Messenger\Event\AbstractWorkerMessageEvent;
+use Symfony\Component\Messenger\Event\WorkerMessageFailedEvent;
+use Symfony\Component\Messenger\Event\WorkerMessageHandledEvent;
+
+/**
+ * @author Grégoire Pineau <lyrixx@lyrixx.info>
+ */
+class ResetServicesListener implements EventSubscriberInterface
+{
+    private $servicesResetter;
+    private $receiversName;
+
+    public function __construct(ServicesResetter $servicesResetter, array $receiversName)
+    {
+        $this->servicesResetter = $servicesResetter;
+        $this->receiversName = $receiversName;
+    }
+
+    public function resetServices(AbstractWorkerMessageEvent $event)
+    {
+        if (!\in_array($event->getReceiverName(), $this->receiversName, true)) {
+            return;
+        }
+
+        $this->servicesResetter->reset();
+    }
+
+    public static function getSubscribedEvents()
+    {
+        return [
+            WorkerMessageHandledEvent::class => ['resetServices'],
+            WorkerMessageFailedEvent::class => ['resetServices'],
+        ];
+    }
+}

src/Symfony/Component/Messenger/Tests/EventListener/ResetServicesListenerTest.php

@@ -0,0 +1,39 @@
+<?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\Tests\EventListener;
+
+use PHPUnit\Framework\TestCase;
+use Symfony\Component\HttpKernel\DependencyInjection\ServicesResetter;
+use Symfony\Component\Messenger\Envelope;
+use Symfony\Component\Messenger\Event\AbstractWorkerMessageEvent;
+use Symfony\Component\Messenger\EventListener\ResetServicesListener;
+
+class ResetServicesListenerTest extends TestCase
+{
+    public function provideTests(): iterable
+    {
+        yield ['foo', true];
+        yield ['bar', false];
+    }
+
+    /** @dataProvider provideTests */
+    public function test(string $receiverName, bool $shouldReset)
+    {
+        $servicesResetter = $this->createMock(ServicesResetter::class);
+        $servicesResetter->expects($shouldReset ? $this->once() : $this->never())->method('reset');
+
+        $event = new class(new Envelope(new \stdClass()), $receiverName) extends AbstractWorkerMessageEvent {};
+
+        $memoryLimitListener = new ResetServicesListener($servicesResetter, ['foo']);
+        $memoryLimitListener->resetServices($event);
+    }
+}