monitoring

Author: ASKozienko

ConsumedMessageStats.php

@@ -31,6 +31,16 @@ class ConsumedMessageStats implements Stats
      */
     protected $queue;
 
+    /**
+     * @var string
+     */
+    protected $messageId;
+
+    /**
+     * @var string
+     */
+    protected $correlationId;
+
     /**
      * @var array
      */
@@ -41,27 +51,81 @@ class ConsumedMessageStats implements Stats
      */
     protected $properties;
 
+    /**
+     * @var bool;
+     */
+    protected $redelivered;
+
     /**
      * @var string
      */
     protected $status;
 
+    /**
+     * @var string
+     */
+    protected $errorClass;
+
+    /**
+     * @var string
+     */
+    protected $errorMessage;
+
+    /**
+     * @var int
+     */
+    protected $errorCode;
+
+    /**
+     * @var string
+     */
+    protected $errorFile;
+
+    /**
+     * @var int
+     */
+    protected $errorLine;
+
+    /**
+     * @var string
+     */
+    protected $trance;
+
     public function __construct(
         string $consumerId,
         int $timestampMs,
         int $receivedAtMs,
         string $queue,
+        ?string $messageId,
+        ?string $correlationId,
         array $headers,
         array $properties,
-        string $status
+        bool $redelivered,
+        string $status,
+        string $errorClass = null,
+        string $errorMessage = null,
+        int $errorCode = null,
+        string $errorFile = null,
+        int $errorLine = null,
+        string $trace = null
     ) {
         $this->consumerId = $consumerId;
         $this->timestampMs = $timestampMs;
         $this->receivedAtMs = $receivedAtMs;
         $this->queue = $queue;
+        $this->messageId = $messageId;
+        $this->correlationId = $correlationId;
         $this->headers = $headers;
         $this->properties = $properties;
+        $this->redelivered = $redelivered;
         $this->status = $status;
+
+        $this->errorClass = $errorClass;
+        $this->errorMessage = $errorMessage;
+        $this->errorCode = $errorCode;
+        $this->errorFile = $errorFile;
+        $this->errorLine = $errorLine;
+        $this->trance = $trace;
     }
 
     public function getConsumerId(): string
@@ -84,6 +148,16 @@ public function getQueue(): string
         return $this->queue;
     }
 
+    public function getMessageId(): ?string
+    {
+        return $this->messageId;
+    }
+
+    public function getCorrelationId(): ?string
+    {
+        return $this->correlationId;
+    }
+
     public function getHeaders(): array
     {
         return $this->headers;
@@ -94,8 +168,43 @@ public function getProperties(): array
         return $this->properties;
     }
 
+    public function isRedelivered(): bool
+    {
+        return $this->redelivered;
+    }
+
     public function getStatus(): string
     {
         return $this->status;
     }
+
+    public function getErrorClass(): ?string
+    {
+        return $this->errorClass;
+    }
+
+    public function getErrorMessage(): ?string
+    {
+        return $this->errorMessage;
+    }
+
+    public function getErrorCode(): ?int
+    {
+        return $this->errorCode;
+    }
+
+    public function getErrorFile(): ?string
+    {
+        return $this->errorFile;
+    }
+
+    public function getErrorLine(): ?int
+    {
+        return $this->errorLine;
+    }
+
+    public function getTrance(): ?string
+    {
+        return $this->trance;
+    }
 }

InfluxDbStorage.php

@@ -122,6 +122,7 @@ public function pushConsumedMessageStats(ConsumedMessageStats $stats): void
         $values = [
             'receivedAt' => $stats->getReceivedAtMs(),
             'processedAt' => $stats->getTimestampMs(),
+            'redelivered' => $stats->isRedelivered(),
         ];
 
         if (ConsumedMessageStats::STATUS_FAILED === $stats->getStatus()) {

MonitoringExtension.php

@@ -3,21 +3,24 @@
 namespace Enqueue\Monitoring;
 
 use Enqueue\Consumption\Context\End;
-use Enqueue\Consumption\Context\InitLogger;
 use Enqueue\Consumption\Context\MessageReceived;
 use Enqueue\Consumption\Context\MessageResult;
-use Enqueue\Consumption\Context\PostConsume;
-use Enqueue\Consumption\Context\PostMessageReceived;
 use Enqueue\Consumption\Context\PreConsume;
 use Enqueue\Consumption\Context\PreSubscribe;
 use Enqueue\Consumption\Context\ProcessorException;
 use Enqueue\Consumption\Context\Start;
-use Enqueue\Consumption\ExtensionInterface;
+use Enqueue\Consumption\EndExtensionInterface;
+use Enqueue\Consumption\MessageReceivedExtensionInterface;
+use Enqueue\Consumption\MessageResultExtensionInterface;
+use Enqueue\Consumption\PreConsumeExtensionInterface;
+use Enqueue\Consumption\PreSubscribeExtensionInterface;
+use Enqueue\Consumption\ProcessorExceptionExtensionInterface;
 use Enqueue\Consumption\Result;
+use Enqueue\Consumption\StartExtensionInterface;
 use Psr\Log\LoggerInterface;
 use Ramsey\Uuid\Uuid;
 
-class MonitoringExtension implements ExtensionInterface
+class MonitoringExtension implements StartExtensionInterface, PreSubscribeExtensionInterface, PreConsumeExtensionInterface, EndExtensionInterface, ProcessorExceptionExtensionInterface, MessageReceivedExtensionInterface, MessageResultExtensionInterface
 {
     /**
      * @var StatsStorage
@@ -165,9 +168,18 @@ public function onProcessorException(ProcessorException $context): void
             $timeMs,
             $context->getReceivedAt(),
             $context->getConsumer()->getQueue()->getQueueName(),
+            $context->getMessage()->getMessageId(),
+            $context->getMessage()->getCorrelationId(),
             $context->getMessage()->getHeaders(),
             $context->getMessage()->getProperties(),
-            ConsumedMessageStats::STATUS_FAILED
+            $context->getMessage()->isRedelivered(),
+            ConsumedMessageStats::STATUS_FAILED,
+            get_class($context->getException()),
+            $context->getException()->getMessage(),
+            $context->getException()->getCode(),
+            $context->getException()->getFile(),
+            $context->getException()->getLine(),
+            $context->getException()->getTraceAsString()
         );
 
         $this->safeCall(function () use ($event) {
@@ -238,8 +250,11 @@ public function onResult(MessageResult $context): void
             $timeMs,
             $context->getReceivedAt(),
             $context->getConsumer()->getQueue()->getQueueName(),
+            $context->getMessage()->getMessageId(),
+            $context->getMessage()->getCorrelationId(),
             $context->getMessage()->getHeaders(),
             $context->getMessage()->getProperties(),
+            $context->getMessage()->isRedelivered(),
             $status
         );
 
@@ -275,18 +290,6 @@ public function onResult(MessageResult $context): void
         }
     }
 
-    public function onPostConsume(PostConsume $context): void
-    {
-    }
-
-    public function onPostMessageReceived(PostMessageReceived $context): void
-    {
-    }
-
-    public function onInitLogger(InitLogger $context): void
-    {
-    }
-
     private function getNowMs(): int
     {
         return (int) (microtime(true) * 1000);

SentMessageStats.php

@@ -11,6 +11,16 @@ class SentMessageStats implements Stats
      */
     protected $timestampMs;
 
+    /**
+     * @var string
+     */
+    protected $messageId;
+
+    /**
+     * @var string
+     */
+    protected $correlationId;
+
     /**
      * @var string
      */
@@ -29,11 +39,15 @@ class SentMessageStats implements Stats
     public function __construct(
         int $timestampMs,
         string $queue,
+        ?string $messageId,
+        ?string $correlationId,
         array $headers,
         array $properties
     ) {
         $this->timestampMs = $timestampMs;
         $this->queue = $queue;
+        $this->messageId = $messageId;
+        $this->correlationId = $correlationId;
         $this->headers = $headers;
         $this->properties = $properties;
     }
@@ -48,6 +62,16 @@ public function getQueue(): string
         return $this->queue;
     }
 
+    public function getMessageId(): ?string
+    {
+        return $this->messageId;
+    }
+
+    public function getCorrelationId(): ?string
+    {
+        return $this->correlationId;
+    }
+
     public function getHeaders(): array
     {
         return $this->headers;