[consumption] Add LogExtension.

Author: makasim

pkg/enqueue/Consumption/Context/PostMessageReceived.php

@@ -3,6 +3,7 @@
 namespace Enqueue\Consumption\Context;
 
 use Enqueue\Consumption\Result;
+use Interop\Queue\Consumer;
 use Interop\Queue\Context;
 use Interop\Queue\Message;
 use Psr\Log\LoggerInterface;
@@ -14,6 +15,11 @@ final class PostMessageReceived
      */
     private $context;
 
+    /**
+     * @var Consumer
+     */
+    private $consumer;
+
     /**
      * @var Message
      */
@@ -41,12 +47,14 @@ final class PostMessageReceived
 
     public function __construct(
         Context $context,
+        Consumer $consumer,
         Message $message,
         $result,
         int $receivedAt,
         LoggerInterface $logger
     ) {
         $this->context = $context;
+        $this->consumer = $consumer;
         $this->message = $message;
         $this->result = $result;
         $this->receivedAt = $receivedAt;
@@ -60,6 +68,11 @@ public function getContext(): Context
         return $this->context;
     }
 
+    public function getConsumer(): Consumer
+    {
+        return $this->consumer;
+    }
+
     public function getMessage(): Message
     {
         return $this->message;

pkg/enqueue/Consumption/Extension/LogExtension.php

@@ -0,0 +1,67 @@
+<?php
+
+namespace Enqueue\Consumption\Extension;
+
+use Enqueue\Consumption\Context\End;
+use Enqueue\Consumption\Context\MessageReceived;
+use Enqueue\Consumption\Context\PostMessageReceived;
+use Enqueue\Consumption\Context\Start;
+use Enqueue\Consumption\EndExtensionInterface;
+use Enqueue\Consumption\MessageReceivedExtensionInterface;
+use Enqueue\Consumption\PostMessageReceivedExtensionInterface;
+use Enqueue\Consumption\Result;
+use Enqueue\Consumption\StartExtensionInterface;
+use Enqueue\Util\Stringify;
+use Psr\Log\LogLevel;
+
+class LogExtension implements StartExtensionInterface, MessageReceivedExtensionInterface, PostMessageReceivedExtensionInterface, EndExtensionInterface
+{
+    public function onStart(Start $context): void
+    {
+        $context->getLogger()->debug('Consumption has started');
+    }
+
+    public function onEnd(End $context): void
+    {
+        $context->getLogger()->debug('Consumption has ended');
+    }
+
+    public function onMessageReceived(MessageReceived $context): void
+    {
+        $message = $context->getMessage();
+
+        $context->getLogger()->debug("Received from {queueName}\t{body}", [
+            'queueName' => $context->getConsumer()->getQueue()->getQueueName(),
+            'redelivered' => $message->isRedelivered(),
+            'body' => Stringify::that($message->getBody()),
+            'properties' => Stringify::that($message->getProperties()),
+            'headers' => Stringify::that($message->getHeaders()),
+        ]);
+    }
+
+    public function onPostMessageReceived(PostMessageReceived $context): void
+    {
+        $message = $context->getMessage();
+        $queue = $context->getConsumer()->getQueue();
+        $result = $context->getResult();
+
+        $reason = '';
+        $logMessage = "Processed from {queueName}\t{body}\t{result}";
+        if ($result instanceof Result && $result->getReason()) {
+            $reason = $result->getReason();
+            $logMessage .= ' {reason}';
+        }
+        $logContext = [
+            'result' => str_replace('enqueue.', '', $result),
+            'reason' => $reason,
+            'queueName' => $queue->getQueueName(),
+            'body' => Stringify::that($message->getBody()),
+            'properties' => Stringify::that($message->getProperties()),
+            'headers' => Stringify::that($message->getHeaders()),
+        ];
+
+        $logLevel = Result::REJECT == ((string) $result) ? LogLevel::ERROR : LogLevel::INFO;
+
+        $context->getLogger()->log($logLevel, $logMessage, $logContext);
+    }
+}

pkg/enqueue/Consumption/QueueConsumer.php

@@ -220,7 +220,7 @@ public function consume(ExtensionInterface $runtimeExtension = null): void
                     throw new \LogicException(sprintf('Status is not supported: %s', $result));
             }
 
-            $postMessageReceived = new PostMessageReceived($this->interopContext, $message, $result, $receivedAt, $this->logger);
+            $postMessageReceived = new PostMessageReceived($this->interopContext, $consumer, $message, $result, $receivedAt, $this->logger);
             $extension->onPostMessageReceived($postMessageReceived);
 
             if ($postMessageReceived->isExecutionInterrupted()) {

pkg/enqueue/Symfony/DependencyInjection/TransportFactory.php

@@ -5,6 +5,7 @@
 use Enqueue\ConnectionFactoryFactory;
 use Enqueue\ConnectionFactoryFactoryInterface;
 use Enqueue\Consumption\ChainExtension;
+use Enqueue\Consumption\Extension\LogExtension;
 use Enqueue\Consumption\QueueConsumer;
 use Enqueue\Consumption\QueueConsumerInterface;
 use Enqueue\Resources;
@@ -166,8 +167,13 @@ public function buildQueueConsumer(ContainerBuilder $container, array $config):
 
         $container->setParameter($this->format('receive_timeout'), $config['receive_timeout'] ?? 10000);
 
+        $logExtensionId = $this->format('log_extension');
+        $container->register($logExtensionId, LogExtension::class);
+
         $container->register($this->format('consumption_extensions'), ChainExtension::class)
-            ->addArgument([])
+            ->addArgument([
+                new Reference($logExtensionId),
+            ])
         ;
 
         $container->register($this->format('queue_consumer'), QueueConsumer::class)

pkg/enqueue/Tests/Client/ConsumptionExtension/FlushSpoolProducerExtensionTest.php

@@ -9,6 +9,7 @@
 use Enqueue\Consumption\EndExtensionInterface;
 use Enqueue\Consumption\PostMessageReceivedExtensionInterface;
 use Enqueue\Test\ClassExtensionTrait;
+use Interop\Queue\Consumer;
 use Interop\Queue\Context;
 use Interop\Queue\Message;
 use PHPUnit\Framework\TestCase;
@@ -57,6 +58,7 @@ public function testShouldFlushSpoolProducerOnPostReceived()
 
         $context = new PostMessageReceived(
             $this->createInteropContextMock(),
+            $this->createMock(Consumer::class),
             $this->createMock(Message::class),
             'aResult',
             1,

pkg/enqueue/Tests/Consumption/ChainExtensionTest.php

@@ -197,6 +197,7 @@ public function testShouldProxyOnPostReceiveToAllInternalExtensions()
     {
         $context = new PostMessageReceived(
             $this->createInteropContextMock(),
+            $this->createMock(Consumer::class),
             $this->createMock(Message::class),
             'aResult',
             1,

pkg/enqueue/Tests/Consumption/Extension/LimitConsumedMessagesExtensionTest.php

@@ -5,6 +5,7 @@
 use Enqueue\Consumption\Context\PostMessageReceived;
 use Enqueue\Consumption\Context\PreConsume;
 use Enqueue\Consumption\Extension\LimitConsumedMessagesExtension;
+use Interop\Queue\Consumer;
 use Interop\Queue\Context;
 use Interop\Queue\Message;
 use Interop\Queue\SubscriptionConsumer;
@@ -49,6 +50,7 @@ public function testOnPreConsumeShouldInterruptWhenLimitIsReached()
 
         $postReceivedMessage = new PostMessageReceived(
             $this->createInteropContextMock(),
+            $this->createMock(Consumer::class),
             $this->createMock(Message::class),
             'aResult',
             1,
@@ -135,6 +137,7 @@ public function testOnPostReceivedShouldInterruptExecutionIfMessageLimitExceeded
 
         $postReceivedMessage = new PostMessageReceived(
             $this->createInteropContextMock(),
+            $this->createMock(Consumer::class),
             $this->createMock(Message::class),
             'aResult',
             1,

pkg/enqueue/Tests/Consumption/Extension/LimitConsumerMemoryExtensionTest.php

@@ -6,6 +6,7 @@
 use Enqueue\Consumption\Context\PostMessageReceived;
 use Enqueue\Consumption\Context\PreConsume;
 use Enqueue\Consumption\Extension\LimitConsumerMemoryExtension;
+use Interop\Queue\Consumer;
 use Interop\Queue\Context;
 use Interop\Queue\Message;
 use Interop\Queue\SubscriptionConsumer;
@@ -66,6 +67,7 @@ public function testOnPostReceivedShouldInterruptExecutionIfMemoryLimitReached()
 
         $postReceivedMessage = new PostMessageReceived(
             $this->createInteropContextMock(),
+            $this->createMock(Consumer::class),
             $this->createMock(Message::class),
             'aResult',
             1,
@@ -156,6 +158,7 @@ public function testOnPostMessageReceivedShouldNotInterruptExecutionIfMemoryLimi
     {
         $postReceivedMessage = new PostMessageReceived(
             $this->createInteropContextMock(),
+            $this->createMock(Consumer::class),
             $this->createMock(Message::class),
             'aResult',
             1,

pkg/enqueue/Tests/Consumption/Extension/LimitConsumptionTimeExtensionTest.php

@@ -6,6 +6,7 @@
 use Enqueue\Consumption\Context\PostMessageReceived;
 use Enqueue\Consumption\Context\PreConsume;
 use Enqueue\Consumption\Extension\LimitConsumptionTimeExtension;
+use Interop\Queue\Consumer;
 use Interop\Queue\Context;
 use Interop\Queue\Message;
 use Interop\Queue\SubscriptionConsumer;
@@ -68,6 +69,7 @@ public function testOnPostReceivedShouldInterruptExecutionIfConsumptionTimeExcee
     {
         $postReceivedMessage = new PostMessageReceived(
             $this->createInteropContextMock(),
+            $this->createMock(Consumer::class),
             $this->createMock(Message::class),
             'aResult',
             1,
@@ -133,6 +135,7 @@ public function testOnPostReceivedShouldNotInterruptExecutionIfConsumptionTimeIs
     {
         $postReceivedMessage = new PostMessageReceived(
             $this->createInteropContextMock(),
+            $this->createMock(Consumer::class),
             $this->createMock(Message::class),
             'aResult',
             1,