[Mesenger] Add support for reseting container services between 2 mess…

…ages 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://' ```
symfony · May 10, 2021 · 0a51dc6 · 0a51dc6
1 parent c151f76
commit 0a51dc6
Show file tree

Hide file tree

Showing 13 changed files with 130 additions and 3 deletions.
diff --git a/src/Symfony/Bundle/FrameworkBundle/CHANGELOG.md b/src/Symfony/Bundle/FrameworkBundle/CHANGELOG.md
@@ -1,6 +1,10 @@
 CHANGELOG
 =========
 
+5.4
+---
+ * Add support for resting container services after each messenger messages.
+
 5.3
 ---
 

diff --git a/src/Symfony/Bundle/FrameworkBundle/DependencyInjection/Configuration.php b/src/Symfony/Bundle/FrameworkBundle/DependencyInjection/Configuration.php
@@ -1325,6 +1325,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)

diff --git a/src/Symfony/Bundle/FrameworkBundle/DependencyInjection/FrameworkExtension.php b/src/Symfony/Bundle/FrameworkBundle/DependencyInjection/FrameworkExtension.php
@@ -1983,6 +1983,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))
@@ -2011,6 +2012,19 @@ 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 = [];

diff --git a/src/Symfony/Bundle/FrameworkBundle/Resources/config/messenger.php b/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\StopWorkerOnRestartSignalListener;
@@ -193,6 +194,13 @@
         ->set('messenger.listener.stop_worker_on_sigterm_signal_listener', StopWorkerOnSigtermSignalListener::class)
             ->tag('kernel.event_subscriber')
 
+        ->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)
             ->args([
                 abstract_arg('message bus locator'),

diff --git a/src/Symfony/Bundle/FrameworkBundle/Resources/config/schema/symfony-1.0.xsd b/src/Symfony/Bundle/FrameworkBundle/Resources/config/schema/symfony-1.0.xsd
@@ -503,6 +503,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">

diff --git a/...ny/Bundle/FrameworkBundle/Tests/DependencyInjection/Fixtures/php/messenger_transports.php b/...ny/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' => [

diff --git a/...ny/Bundle/FrameworkBundle/Tests/DependencyInjection/Fixtures/xml/messenger_transports.xml b/...ny/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>

diff --git a/...ny/Bundle/FrameworkBundle/Tests/DependencyInjection/Fixtures/yml/messenger_transports.yml b/...ny/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

diff --git a/src/Symfony/Bundle/FrameworkBundle/Tests/DependencyInjection/FrameworkExtensionTest.php b/src/Symfony/Bundle/FrameworkBundle/Tests/DependencyInjection/FrameworkExtensionTest.php
@@ -714,6 +714,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()
@@ -858,6 +859,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()

diff --git a/src/Symfony/Bundle/FrameworkBundle/composer.json b/src/Symfony/Bundle/FrameworkBundle/composer.json
@@ -48,7 +48,7 @@
         "symfony/http-client": "^4.4|^5.0",
         "symfony/lock": "^4.4|^5.0",
         "symfony/mailer": "^5.2",
-        "symfony/messenger": "^5.2",
+        "symfony/messenger": "^5.4",
         "symfony/mime": "^4.4|^5.0",
         "symfony/notifier": "^5.3",
         "symfony/allmysms-notifier": "^5.3",
@@ -115,7 +115,7 @@
         "symfony/form": "<5.2",
         "symfony/lock": "<4.4",
         "symfony/mailer": "<5.2",
-        "symfony/messenger": "<4.4",
+        "symfony/messenger": "<5.4",
         "symfony/mime": "<4.4",
         "symfony/property-info": "<4.4",
         "symfony/property-access": "<5.3",

diff --git a/src/Symfony/Component/Messenger/CHANGELOG.md b/src/Symfony/Component/Messenger/CHANGELOG.md
@@ -1,6 +1,10 @@
 CHANGELOG
 =========
 
+5.4
+---
+ * Add support for resting container services after each messages.
+
 5.3
 ---
 

diff --git a/src/Symfony/Component/Messenger/EventListener/ResetServicesListener.php b/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'],
+        ];
+    }
+}
diff --git a/src/Symfony/Component/Messenger/Tests/EventListener/ResetServicesListenerTest.php b/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);
+    }
+}