Call reset on the logger in order to clean buffers

services.yml

@@ -37,8 +37,12 @@ services:
         arguments:
           - 'text/html'
 
-  Psr\Log\LoggerInterface:
-    class: \Monolog\Logger
+  Psr\Log\LoggerInterface: '@webgriffe_esb_logger'
+
+  Webgriffe\Esb\LoggerResettableInterface: '@webgriffe_esb_logger'
+
+  webgriffe_esb_logger:
+    class: \Webgriffe\Esb\Logger\LoggerResettable
     arguments: [ 'esb', ['@Monolog\Handler\StreamHandler', '@Monolog\Handler\ErrorLogHandler', '@Monolog\Handler\NativeMailerHandler'] ]
     public: true
 

src/Logger/LoggerResettable.php

@@ -0,0 +1,11 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Webgriffe\Esb\Logger;
+
+use Monolog\Logger;
+
+final class LoggerResettable extends Logger implements LoggerResettableInterface
+{
+}

src/Logger/LoggerResettableInterface.php

@@ -0,0 +1,12 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Webgriffe\Esb\Logger;
+
+use Monolog\ResettableInterface;
+use Psr\Log\LoggerInterface;
+
+interface LoggerResettableInterface extends LoggerInterface, ResettableInterface
+{
+}

src/ProducerInstance.php

@@ -8,7 +8,7 @@
 use function Amp\call;
 use Amp\Loop;
 use Amp\Promise;
-use Psr\Log\LoggerInterface;
+use Webgriffe\Esb\Logger\LoggerResettableInterface;
 use Webgriffe\Esb\Model\FlowConfig;
 use Webgriffe\Esb\Model\Job;
 use Webgriffe\Esb\Model\ProducedJobEvent;
@@ -31,7 +31,7 @@ final class ProducerInstance implements ProducerInstanceInterface
     private $producer;
 
     /**
-     * @var LoggerInterface
+     * @var LoggerResettableInterface
      */
     private $logger;
 
@@ -54,7 +54,7 @@ public function __construct(
         FlowConfig $flowConfig,
         ProducerInterface $producer,
         ?BeanstalkClient $beanstalkClient,
-        LoggerInterface $logger,
+        LoggerResettableInterface $logger,
         HttpProducersServer $httpProducersServer,
         CronProducersServer $cronProducersServer,
         ?ElasticSearch $elasticSearch,
@@ -195,6 +195,8 @@ public function produceAndQueueJobs($data = null): Promise
                         'test' => $test
                     ]
                 );
+            } finally {
+                $this->logger->reset();
             }
             return $jobsCount;
         });

src/WorkerInstance.php

@@ -8,8 +8,8 @@
 use function Amp\call;
 use function Amp\delay;
 use Amp\Promise;
-use Psr\Log\LoggerInterface;
 use Webgriffe\Esb\Exception\FatalQueueException;
+use Webgriffe\Esb\Logger\LoggerResettableInterface;
 use Webgriffe\Esb\Model\ErroredJobEvent;
 use Webgriffe\Esb\Model\FlowConfig;
 use Webgriffe\Esb\Model\JobInterface;
@@ -37,7 +37,7 @@ final class WorkerInstance implements WorkerInstanceInterface
     private $worker;
 
     /**
-     * @var LoggerInterface
+     * @var LoggerResettableInterface
      */
     private $logger;
 
@@ -56,7 +56,7 @@ public function __construct(
         int $instanceId,
         WorkerInterface $worker,
         ?BeanstalkClient $beanstalkClient,
-        LoggerInterface $logger,
+        LoggerResettableInterface $logger,
         ?ElasticSearch $elasticSearch,
         WorkerQueueManagerInterface $queueManager = null
     ) {
@@ -146,13 +146,16 @@ public function boot(): Promise
                 try {
                     /** @var JobInterface $job */
                     if (!($job = yield $this->queueManager->getNextJob())) {
+                        $this->logger->reset(); // todo: necessary?
                         break;
                     }
                 } catch (FatalQueueException $ex) {
                     //Let this pass to stop the loop
+                    $this->logger->reset(); // todo: necessary?
                     throw $ex;
                 } catch (\Exception $ex) {
                     $this->logger->critical($ex->getMessage(), $globalLogContext);
+                    $this->logger->reset();
                     continue;
                 }
 
@@ -207,6 +210,7 @@ public function boot(): Promise
                             )
                         );
                         unset(self::$workCounts[$jobUuid]);
+                        $this->logger->reset();
                         continue;
                     }
 
@@ -215,6 +219,8 @@ public function boot(): Promise
                         'Worker released a Job',
                         array_merge($logContext, ['release_delay' => $this->flowConfig->getWorkerReleaseDelay()])
                     );
+                } finally {
+                    $this->logger->reset();
                 }
             }
         });