php-enqueue · makasim · Aug 7, 2019 · Aug 6, 2019
diff --git a/docs/bundle/config_reference.md b/docs/bundle/config_reference.md
@@ -69,6 +69,7 @@ enqueue:
         extensions:
             doctrine_ping_connection_extension: false
             doctrine_clear_identity_map_extension: false
+            reset_services_extension: false
             signal_extension:     true
             reply_extension:      true
 ```

diff --git a/docs/consumption/extensions.md b/docs/consumption/extensions.md
@@ -22,6 +22,12 @@ It clears Doctrine's identity map after a message is processed. It reduce memory
 
 It test a database connection and if it is lost it does reconnect. Fixes "MySQL has gone away" errors.
 
+## [ResetServicesExtension](https://github.com/php-enqueue/enqueue-dev/blob/master/pkg/enqueue-bundle/Consumption/Extension/ResetServicesExtension.php)
+
+It resets all services with tag "kernel.reset".
+For example, this includes all monolog loggers if installed and will flush/clean all buffers,
+reset internal state, and get them back to a state in which they can receive log records again.
+
 ## [ReplyExtension](https://github.com/php-enqueue/enqueue-dev/blob/master/pkg/enqueue/Consumption/Extension/ReplyExtension.php)
 
 It comes with RPC code and simplifies reply logic.

diff --git a/pkg/enqueue-bundle/Consumption/Extension/ResetServicesExtension.php b/pkg/enqueue-bundle/Consumption/Extension/ResetServicesExtension.php
@@ -0,0 +1,27 @@
+<?php
+
+namespace Enqueue\Bundle\Consumption\Extension;
+
+use Enqueue\Consumption\Context\MessageReceived;
+use Enqueue\Consumption\MessageReceivedExtensionInterface;
+use Symfony\Component\HttpKernel\DependencyInjection\ServicesResetter;
+
+class ResetServicesExtension implements MessageReceivedExtensionInterface
+{
+    /**
+     * @var ServicesResetter
+     */
+    private $resetter;
+
+    public function __construct(ServicesResetter $resetter)
+    {
+        $this->resetter = $resetter;
+    }
+
+    public function onMessageReceived(MessageReceived $context): void
+    {
+        $context->getLogger()->debug('[ResetServicesExtension] Resetting services.');
+
+        $this->resetter->reset();
+    }
+}
diff --git a/pkg/enqueue-bundle/DependencyInjection/Configuration.php b/pkg/enqueue-bundle/DependencyInjection/Configuration.php
@@ -47,6 +47,7 @@ public function getConfigTreeBuilder(): TreeBuilder
                     ->arrayNode('extensions')->addDefaultsIfNotSet()->children()
                         ->booleanNode('doctrine_ping_connection_extension')->defaultFalse()->end()
                         ->booleanNode('doctrine_clear_identity_map_extension')->defaultFalse()->end()
+                        ->booleanNode('reset_services_extension')->defaultFalse()->end()
                         ->booleanNode('signal_extension')->defaultValue(function_exists('pcntl_signal_dispatch'))->end()
                         ->booleanNode('reply_extension')->defaultTrue()->end()
                     ->end()->end()

diff --git a/pkg/enqueue-bundle/DependencyInjection/EnqueueExtension.php b/pkg/enqueue-bundle/DependencyInjection/EnqueueExtension.php
@@ -6,6 +6,7 @@
 use Enqueue\AsyncEventDispatcher\DependencyInjection\AsyncEventDispatcherExtension;
 use Enqueue\Bundle\Consumption\Extension\DoctrineClearIdentityMapExtension;
 use Enqueue\Bundle\Consumption\Extension\DoctrinePingConnectionExtension;
+use Enqueue\Bundle\Consumption\Extension\ResetServicesExtension;
 use Enqueue\Bundle\Profiler\MessageQueueCollector;
 use Enqueue\Client\CommandSubscriberInterface;
 use Enqueue\Client\TopicSubscriberInterface;
@@ -136,6 +137,7 @@ public function load(array $configs, ContainerBuilder $container): void
         // extensions
         $this->loadDoctrinePingConnectionExtension($config, $container);
         $this->loadDoctrineClearIdentityMapExtension($config, $container);
+        $this->loadResetServicesExtension($config, $container);
         $this->loadSignalExtension($config, $container);
         $this->loadReplyExtension($config, $container);
     }
@@ -210,7 +212,7 @@ private function loadDoctrinePingConnectionExtension(array $config, ContainerBui
             }
         }
 
-        if (false == $configNames) {
+        if ([] === $configNames) {
             return;
         }
 
@@ -233,7 +235,7 @@ private function loadDoctrineClearIdentityMapExtension(array $config, ContainerB
             }
         }
 
-        if (false == $configNames) {
+        if ([] === $configNames) {
             return;
         }
 
@@ -247,6 +249,28 @@ private function loadDoctrineClearIdentityMapExtension(array $config, ContainerB
         }
     }
 
+    private function loadResetServicesExtension(array $config, ContainerBuilder $container)
+    {
+        $configNames = [];
+        foreach ($config as $name => $modules) {
+            if ($modules['extensions']['reset_services_extension']) {
+                $configNames[] = $name;
+            }
+        }
+
+        if ([] === $configNames) {
+            return;
+        }
+
+        $extension = $container->register('enqueue.consumption.reset_services_extension', ResetServicesExtension::class)
+            ->addArgument(new Reference('services_resetter'));
+
+        foreach ($configNames as $name) {
+            $extension->addTag('enqueue.consumption_extension', ['client' => $name]);
+            $extension->addTag('enqueue.transport.consumption_extension', ['transport' => $name]);
+        }
+    }
+
     private function loadSignalExtension(array $config, ContainerBuilder $container): void
     {
         $configNames = [];
@@ -256,7 +280,7 @@ private function loadSignalExtension(array $config, ContainerBuilder $container)
             }
         }
 
-        if (false == $configNames) {
+        if ([] === $configNames) {
             return;
         }
 
@@ -277,7 +301,7 @@ private function loadReplyExtension(array $config, ContainerBuilder $container):
             }
         }
 
-        if (false == $configNames) {
+        if ([] === $configNames) {
             return;
         }
 

diff --git a/pkg/enqueue-bundle/Tests/Unit/Consumption/Extension/ResetServicesExtensionTest.php b/pkg/enqueue-bundle/Tests/Unit/Consumption/Extension/ResetServicesExtensionTest.php
@@ -0,0 +1,61 @@
+<?php
+
+namespace Enqueue\Bundle\Tests\Unit\Consumption\Extension;
+
+use Doctrine\Common\Persistence\ManagerRegistry;
+use Enqueue\Bundle\Consumption\Extension\ResetServicesExtension;
+use Enqueue\Consumption\Context\MessageReceived;
+use Interop\Queue\Consumer;
+use Interop\Queue\Context as InteropContext;
+use Interop\Queue\Message;
+use Interop\Queue\Processor;
+use PHPUnit\Framework\TestCase;
+use Psr\Log\LoggerInterface;
+use Symfony\Component\HttpKernel\DependencyInjection\ServicesResetter;
+
+class ResetServicesExtensionTest extends TestCase
+{
+    public function testCouldBeConstructedWithRequiredArguments()
+    {
+        new ResetServicesExtension($this->createResetterMock());
+    }
+
+    public function testItShouldResetServices()
+    {
+        $resetter = $this->createResetterMock();
+        $resetter
+            ->expects($this->once())
+            ->method('reset')
+        ;
+
+        $context = $this->createContext();
+        $context->getLogger()
+            ->expects($this->once())
+            ->method('debug')
+            ->with('[ResetServicesExtension] Resetting services.')
+        ;
+
+        $extension = new ResetServicesExtension($resetter);
+        $extension->onMessageReceived($context);
+    }
+
+    protected function createContext(): MessageReceived
+    {
+        return new MessageReceived(
+            $this->createMock(InteropContext::class),
+            $this->createMock(Consumer::class),
+            $this->createMock(Message::class),
+            $this->createMock(Processor::class),
+            1,
+            $this->createMock(LoggerInterface::class)
+        );
+    }
+
+    /**
+     * @return \PHPUnit_Framework_MockObject_MockObject|ManagerRegistry
+     */
+    protected function createResetterMock(): ServicesResetter
+    {
+        return $this->createMock(ServicesResetter::class);
+    }
+}
diff --git a/pkg/enqueue-bundle/Tests/Unit/DependencyInjection/ConfigurationTest.php b/pkg/enqueue-bundle/Tests/Unit/DependencyInjection/ConfigurationTest.php
@@ -336,6 +336,49 @@ public function testDoctrineClearIdentityMapExtensionCouldBeEnabled()
         ], $config);
     }
 
+    public function testResetServicesExtensionShouldBeDisabledByDefault()
+    {
+        $configuration = new Configuration(true);
+
+        $processor = new Processor();
+        $config = $processor->processConfiguration($configuration, [[
+            'default' => [
+                'transport' => null,
+            ],
+        ]]);
+
+        $this->assertArraySubset([
+            'default' => [
+                'extensions' => [
+                    'reset_services_extension' => false,
+                ],
+            ],
+        ], $config);
+    }
+
+    public function testResetServicesExtensionCouldBeEnabled()
+    {
+        $configuration = new Configuration(true);
+
+        $processor = new Processor();
+        $config = $processor->processConfiguration($configuration, [[
+            'default' => [
+                'transport' => [],
+                'extensions' => [
+                    'reset_services_extension' => true,
+                ],
+            ],
+        ]]);
+
+        $this->assertArraySubset([
+            'default' => [
+                'extensions' => [
+                    'reset_services_extension' => true,
+                ],
+            ],
+        ], $config);
+    }
+
     public function testSignalExtensionShouldBeEnabledIfPcntlExtensionIsLoaded()
     {
         $isLoaded = function_exists('pcntl_signal_dispatch');

diff --git a/pkg/enqueue-bundle/Tests/Unit/DependencyInjection/EnqueueExtensionTest.php b/pkg/enqueue-bundle/Tests/Unit/DependencyInjection/EnqueueExtensionTest.php
@@ -384,6 +384,42 @@ public function testShouldNotLoadDoctrineClearIdentityMapExtensionServiceIfDisab
         self::assertFalse($container->hasDefinition('enqueue.consumption.doctrine_clear_identity_map_extension'));
     }
 
+    public function testShouldLoadResetServicesExtensionServiceIfEnabled()
+    {
+        $container = $this->getContainerBuilder(true);
+
+        $extension = new EnqueueExtension();
+
+        $extension->load([[
+            'default' => [
+                'transport' => [],
+                'extensions' => [
+                    'reset_services_extension' => true,
+                ],
+            ],
+        ]], $container);
+
+        self::assertTrue($container->hasDefinition('enqueue.consumption.reset_services_extension'));
+    }
+
+    public function testShouldNotLoadResetServicesExtensionServiceIfDisabled()
+    {
+        $container = $this->getContainerBuilder(true);
+
+        $extension = new EnqueueExtension();
+
+        $extension->load([[
+            'default' => [
+                'transport' => [],
+                'extensions' => [
+                    'reset_services_extension' => false,
+                ],
+            ],
+        ]], $container);
+
+        self::assertFalse($container->hasDefinition('enqueue.consumption.reset_services_extension'));
+    }
+
     public function testShouldLoadSignalExtensionServiceIfEnabled()
     {
         $container = $this->getContainerBuilder(true);