Merge pull request #215 from pvgnd/services-reset

Add ServicesResetterProcessor (closes #193)

CHANGELOG.md

@@ -5,6 +5,8 @@ This project adheres to [Semantic Versioning](http://semver.org/).
 
 ## [Unreleased]
 
+- Add ServicesResetterProcessor
+
 ## [3.6.1] - 2020-01-23
 
 - Fix NewRelicProcessor

README.md

@@ -160,6 +160,7 @@ if ($message = $messageProvider->get()) {
 * [RetryProcessor](src/Swarrot/Processor/Retry)
 * [RPC related processors](src/Swarrot/Processor/RPC) (thanks to [Baptiste Clavié](https://github.com/Taluu))
 * [SentryProcessor](src/Swarrot/Processor/Sentry) (thanks to [Floran Brutel](https://github.com/notFloran))
+* [ServicesResetterProcessor](src/Swarrot/Processor/ServicesResetter) (thanks to [Pierrick Vignand](https://github.com/pvgnd))
 * [SignalHandlerProcessor](src/Swarrot/Processor/SignalHandler)
 * [XDeathMaxCountProcessor](src/Swarrot/Processor/XDeath) (thanks to [Anthony Moutte](https://github.com/instabledesign))
 * [XDeathMaxLifetimeProcessor](src/Swarrot/Processor/XDeath) (thanks to [Anthony Moutte](https://github.com/instabledesign))

composer.json

@@ -27,12 +27,14 @@
         "stomp-php/stomp-php": "^4.3.1",
         "php-http/curl-client": "^2.0",
         "symfony/phpunit-bridge": "^5.0",
-        "symfony/error-handler": "^5.0"
+        "symfony/error-handler": "^5.0",
+        "symfony/contracts": "^1.0 || ^2.0"
     },
     "suggest": {
         "pecl-amqp": "*",
         "php-amqplib/php-amqplib": "^2.1",
         "symfony/console": "^3.4 || ^4.1",
+        "symfony/contracts": "^1.0 || ^2.0",
         "queue-interop/queue-interop": "^0.5",
         "queue-interop/amqp-interop": "^0.5",
         "aws/aws-sdk-php": "^2.8",

src/Swarrot/Processor/ServicesResetter/README.md

@@ -0,0 +1,5 @@
+# ServicesResetter Processor
+
+ServicesResetterProcessor is a [swarrot](https://github.com/swarrot/swarrot) processor.
+
+It resets services from Symfony Container after the processor ran to avoid memory leaks (clear Monolog logger for example).

src/Swarrot/Processor/ServicesResetter/ServicesResetterProcessor.php

@@ -0,0 +1,43 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Swarrot\Processor\ServicesResetter;
+
+use Swarrot\Broker\Message;
+use Swarrot\Processor\ProcessorInterface;
+use Symfony\Contracts\Service\ResetInterface;
+
+/**
+ * @author Pierrick Vignand <pierrick.vignand@gmail.com>
+ */
+class ServicesResetterProcessor implements ProcessorInterface
+{
+    /**
+     * @var ProcessorInterface
+     */
+    private $processor;
+
+    /**
+     * @var ResetInterface
+     */
+    private $servicesResetter;
+
+    public function __construct(ProcessorInterface $processor, ResetInterface $servicesResetter)
+    {
+        $this->processor = $processor;
+        $this->servicesResetter = $servicesResetter;
+    }
+
+    /**
+     * {@inheritdoc}
+     */
+    public function process(Message $message, array $options)
+    {
+        try {
+            return $this->processor->process($message, $options);
+        } finally {
+            $this->servicesResetter->reset();
+        }
+    }
+}

tests/Swarrot/Processor/ServicesResetter/ServicesResetterProcessorTest.php

@@ -0,0 +1,47 @@
+<?php
+
+namespace Swarrot\Tests\Processor\ServicesResetter;
+
+use PHPUnit\Framework\TestCase;
+use Swarrot\Broker\Message;
+use Swarrot\Processor\ProcessorInterface;
+use Swarrot\Processor\ServicesResetter\ServicesResetterProcessor;
+use Symfony\Contracts\Service\ResetInterface;
+
+class ServicesResetterProcessorTest extends TestCase
+{
+    public function test()
+    {
+        $message = new Message();
+        $options = [];
+
+        $innerProcessorProphecy = $this->prophesize(ProcessorInterface::class);
+        $innerProcessorProphecy->process($message, $options)->willReturn(true);
+
+        $serviceResetterProphecy = $this->prophesize(ResetInterface::class);
+        $serviceResetterProphecy->reset()->shouldBeCalled();
+
+        $processor = new ServicesResetterProcessor($innerProcessorProphecy->reveal(), $serviceResetterProphecy->reveal());
+
+        $this->assertEquals($processor->process($message, $options), true);
+    }
+
+    public function testWithException()
+    {
+        $message = new Message();
+        $options = [];
+
+        $innerProcessorProphecy = $this->prophesize(ProcessorInterface::class);
+        $innerProcessorProphecy->process($message, $options)->willThrow(new \Exception('my_fake_message'));
+
+        $serviceResetterProphecy = $this->prophesize(ResetInterface::class);
+        $serviceResetterProphecy->reset()->shouldBeCalled();
+
+        $processor = new ServicesResetterProcessor($innerProcessorProphecy->reveal(), $serviceResetterProphecy->reveal());
+
+        $this->expectException(\Exception::class);
+        $this->expectExceptionMessage('my_fake_message');
+
+        $processor->process($message, $options);
+    }
+}