[Mesenger] Add support for reseting container services between 2 messages

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

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

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,52 @@
+<?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;
+use Symfony\Component\Messenger\Event\WorkerRunningEvent;
+
+/**
+ * @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'],
+            WorkerRunningEvent::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 {};
+
+        $resetListener = new ResetServicesListener($servicesResetter, ['foo']);
+        $resetListener->resetServices($event);
+    }
+}