Queue monitoring.

composer.json

@@ -32,7 +32,8 @@
     "richardfullmer/rabbitmq-management-api": "^2.0",
     "predis/predis": "^1.1",
     "thruway/pawl-transport": "^0.5.0",
-    "voryx/thruway": "^0.5.3"
+    "voryx/thruway": "^0.5.3",
+    "influxdb/influxdb-php": "^1.14"
   },
   "require-dev": {
     "phpunit/phpunit": "^5.5",
@@ -78,6 +79,7 @@
       "Enqueue\\Test\\": "pkg/test/",
       "Enqueue\\Dsn\\": "pkg/dsn/",
       "Enqueue\\Wamp\\": "pkg/wamp/",
+      "Enqueue\\Monitoring\\": "pkg/monitoring/",
       "Enqueue\\": "pkg/enqueue/"
     },
     "exclude-from-classmap": [

docker-compose.yml

@@ -125,3 +125,16 @@ services:
       HOSTNAME_EXTERNAL: 'localstack'
       SERVICES: 'sqs'
 
+  influxdb:
+    image: 'influxdb:latest'
+
+  chronograf:
+    image: 'chronograf:latest'
+    entrypoint: 'chronograf --influxdb-url=http://influxdb:8086'
+    ports:
+      - '8888:8888'
+
+  grafana:
+    image: 'grafana/grafana:latest'
+    ports:
+      - '3000:3000'

pkg/enqueue/Consumption/Context/MessageResult.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 MessageResult
      */
     private $context;
 
+    /**
+     * @var Consumer
+     */
+    private $consumer;
+
     /**
      * @var Message
      */
@@ -34,9 +40,10 @@ final class MessageResult
      */
     private $logger;
 
-    public function __construct(Context $context, Message $message, $result, int $receivedAt, LoggerInterface $logger)
+    public function __construct(Context $context, Consumer $consumer, Message $message, $result, int $receivedAt, LoggerInterface $logger)
     {
         $this->context = $context;
+        $this->consumer = $consumer;
         $this->message = $message;
         $this->logger = $logger;
         $this->result = $result;
@@ -48,6 +55,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/Context/ProcessorException.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 ProcessorException
      */
     private $context;
 
+    /**
+     * @var Consumer
+     */
+    private $consumer;
+
     /**
      * @var Message
      */
@@ -38,9 +44,10 @@ final class ProcessorException
      */
     private $logger;
 
-    public function __construct(Context $context, Message $message, \Exception $exception, int $receivedAt, LoggerInterface $logger)
+    public function __construct(Context $context, Consumer $consumer, Message $message, \Exception $exception, int $receivedAt, LoggerInterface $logger)
     {
         $this->context = $context;
+        $this->consumer = $consumer;
         $this->message = $message;
         $this->exception = $exception;
         $this->logger = $logger;
@@ -52,6 +59,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/QueueConsumer.php

@@ -196,11 +196,11 @@ public function consume(ExtensionInterface $runtimeExtension = null): void
                 try {
                     $result = $processor->process($message, $this->interopContext);
                 } catch (\Exception $e) {
-                    $result = $this->onProcessorException($extension, $message, $e, $receivedAt);
+                    $result = $this->onProcessorException($extension, $consumer, $message, $e, $receivedAt);
                 }
             }
 
-            $messageResult = new MessageResult($this->interopContext, $message, $result, $receivedAt, $this->logger);
+            $messageResult = new MessageResult($this->interopContext, $consumer, $message, $result, $receivedAt, $this->logger);
             $extension->onResult($messageResult);
             $result = $messageResult->getResult();
 
@@ -274,8 +274,6 @@ public function consume(ExtensionInterface $runtimeExtension = null): void
 
             ++$cycle;
         }
-
-        $this->enableSubscriptionConsumer = $enableSubscriptionConsumer;
     }
 
     /**
@@ -304,9 +302,9 @@ private function onEnd(ExtensionInterface $extension, int $startTime, Subscripti
      *
      * https://github.com/symfony/symfony/blob/cbe289517470eeea27162fd2d523eb29c95f775f/src/Symfony/Component/HttpKernel/EventListener/ExceptionListener.php#L77
      */
-    private function onProcessorException(ExtensionInterface $extension, Message $message, \Exception $exception, int $receivedAt)
+    private function onProcessorException(ExtensionInterface $extension, Consumer $consumer, Message $message, \Exception $exception, int $receivedAt)
     {
-        $processorException = new ProcessorException($this->interopContext, $message, $exception, $receivedAt, $this->logger);
+        $processorException = new ProcessorException($this->interopContext, $consumer, $message, $exception, $receivedAt, $this->logger);
 
         try {
             $extension->onProcessorException($processorException);
@@ -331,7 +329,5 @@ private function onProcessorException(ExtensionInterface $extension, Message $me
 
             throw $e;
         }
-
-        throw $exception;
     }
 }

pkg/enqueue/Tests/Consumption/ChainExtensionTest.php

@@ -169,6 +169,7 @@ public function testShouldProxyOnResultToAllInternalExtensions()
     {
         $context = new MessageResult(
             $this->createInteropContextMock(),
+            $this->createInteropConsumerMock(),
             $this->createMock(Message::class),
             'aResult',
             1,

pkg/monitoring/.gitattributes

@@ -0,0 +1,5 @@
+/Tests export-ignore
+.gitattributes export-ignore
+.gitignore export-ignore
+.travis.yml export-ignore
+phpunit.xml.dist export-ignore

pkg/monitoring/.gitignore

@@ -0,0 +1,6 @@
+*~
+/composer.lock
+/composer.phar
+/phpunit.xml
+/vendor/
+/.idea/

pkg/monitoring/.travis.yml

@@ -0,0 +1,21 @@
+sudo: false
+
+git:
+    depth: 10
+
+language: php
+
+php:
+    - '7.1'
+    - '7.2'
+
+cache:
+    directories:
+        - $HOME/.composer/cache
+
+install:
+    - composer self-update
+    - composer install
+
+script:
+    - vendor/bin/phpunit --exclude-group=functional

pkg/monitoring/ConsumedMessageStats.php

@@ -0,0 +1,101 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Enqueue\Monitoring;
+
+class ConsumedMessageStats implements Stats
+{
+    const STATUS_ACK = 'acknowledged';
+    const STATUS_REJECTED = 'rejected';
+    const STATUS_REQUEUED = 'requeued';
+    const STATUS_FAILED = 'failed';
+
+    /**
+     * @var string
+     */
+    protected $consumerId;
+
+    /**
+     * @var int
+     */
+    protected $timestampMs;
+
+    /**
+     * @var int
+     */
+    protected $receivedAtMs;
+
+    /**
+     * @var string
+     */
+    protected $queue;
+
+    /**
+     * @var array
+     */
+    protected $headers;
+
+    /**
+     * @var array
+     */
+    protected $properties;
+
+    /**
+     * @var string
+     */
+    protected $status;
+
+    public function __construct(
+        string $consumerId,
+        int $timestampMs,
+        int $receivedAtMs,
+        string $queue,
+        array $headers,
+        array $properties,
+        string $status
+    ) {
+        $this->consumerId = $consumerId;
+        $this->timestampMs = $timestampMs;
+        $this->receivedAtMs = $receivedAtMs;
+        $this->queue = $queue;
+        $this->headers = $headers;
+        $this->properties = $properties;
+        $this->status = $status;
+    }
+
+    public function getConsumerId(): string
+    {
+        return $this->consumerId;
+    }
+
+    public function getTimestampMs(): int
+    {
+        return $this->timestampMs;
+    }
+
+    public function getReceivedAtMs(): int
+    {
+        return $this->receivedAtMs;
+    }
+
+    public function getQueue(): string
+    {
+        return $this->queue;
+    }
+
+    public function getHeaders(): array
+    {
+        return $this->headers;
+    }
+
+    public function getProperties(): array
+    {
+        return $this->properties;
+    }
+
+    public function getStatus(): string
+    {
+        return $this->status;
+    }
+}