Queue monitoring.

pkg/enqueue/Tests/Client/Extension/PrepareBodyExtensionTest.php

@@ -4,9 +4,10 @@
 
 use Enqueue\Client\DriverInterface;
 use Enqueue\Client\Extension\PrepareBodyExtension;
-use Enqueue\Client\ExtensionInterface;
 use Enqueue\Client\Message;
 use Enqueue\Client\PreSend;
+use Enqueue\Client\PreSendCommandExtensionInterface;
+use Enqueue\Client\PreSendEventExtensionInterface;
 use Enqueue\Client\ProducerInterface;
 use Enqueue\Tests\Mocks\JsonSerializableObject;
 use PHPUnit\Framework\TestCase;
@@ -17,7 +18,8 @@ public function testShouldImplementExtensionInterface()
     {
         $rc = new \ReflectionClass(PrepareBodyExtension::class);
 
-        $this->assertTrue($rc->implementsInterface(ExtensionInterface::class));
+        $this->assertTrue($rc->implementsInterface(PreSendEventExtensionInterface::class));
+        $this->assertTrue($rc->implementsInterface(PreSendCommandExtensionInterface::class));
     }
 
     public function testCouldConstructedWithoutAnyArguments()

pkg/enqueue/Tests/Client/ProducerSendCommandTest.php

@@ -2,6 +2,7 @@
 
 namespace Enqueue\Tests\Client;
 
+use Enqueue\Client\ChainExtension;
 use Enqueue\Client\Config;
 use Enqueue\Client\DriverInterface;
 use Enqueue\Client\DriverPreSend;
@@ -304,7 +305,7 @@ public function testShouldSerializeMessageByCustomExtension()
             })
         ;
 
-        $producer = new Producer($driver, $this->createRpcFactoryMock(), new CustomPrepareBodyClientExtension());
+        $producer = new Producer($driver, $this->createRpcFactoryMock(), new ChainExtension([new CustomPrepareBodyClientExtension()]));
         $producer->sendCommand('command', ['foo' => 'fooVal']);
     }
 

pkg/enqueue/Tests/Client/ProducerSendEventTest.php

@@ -2,6 +2,7 @@
 
 namespace Enqueue\Tests\Client;
 
+use Enqueue\Client\ChainExtension;
 use Enqueue\Client\Config;
 use Enqueue\Client\DriverInterface;
 use Enqueue\Client\DriverPreSend;
@@ -210,7 +211,7 @@ public function testShouldSerializeMessageByCustomExtension()
             })
         ;
 
-        $producer = new Producer($driver, $this->createRpcFactoryMock(), new CustomPrepareBodyClientExtension());
+        $producer = new Producer($driver, $this->createRpcFactoryMock(), new ChainExtension([new CustomPrepareBodyClientExtension()]));
         $producer->sendEvent('topic', ['foo' => 'fooVal']);
     }
 

pkg/monitoring/ClientMonitoringExtension.php

@@ -0,0 +1,63 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Enqueue\Monitoring;
+
+use Enqueue\Client\PostSend;
+use Enqueue\Client\PostSendExtensionInterface;
+use Interop\Queue\Topic;
+use Psr\Log\LoggerInterface;
+
+class ClientMonitoringExtension implements PostSendExtensionInterface
+{
+    /**
+     * @var StatsStorage
+     */
+    private $storage;
+
+    /**
+     * @var LoggerInterface
+     */
+    private $logger;
+
+    public function __construct(StatsStorage $storage, LoggerInterface $logger)
+    {
+        $this->storage = $storage;
+        $this->logger = $logger;
+    }
+
+    public function onPostSend(PostSend $context): void
+    {
+        $timestampMs = (int) (microtime(true) * 1000);
+
+        $destination = $context->getTransportDestination() instanceof Topic
+            ? $context->getTransportDestination()->getTopicName()
+            : $context->getTransportDestination()->getQueueName()
+        ;
+
+        $stats = new SentMessageStats(
+            $timestampMs,
+            $destination,
+            $context->getTransportMessage()->getMessageId(),
+            $context->getTransportMessage()->getCorrelationId(),
+            $context->getTransportMessage()->getHeaders(),
+            $context->getTransportMessage()->getProperties()
+        );
+
+        $this->safeCall(function () use ($stats) {
+            $this->storage->pushSentMessageStats($stats);
+        });
+    }
+
+    private function safeCall(callable $fun)
+    {
+        try {
+            return call_user_func($fun);
+        } catch (\Throwable $e) {
+            $this->logger->error(sprintf('[ClientMonitoringExtension] Push to storage failed: %s', $e->getMessage()));
+        }
+
+        return null;
+    }
+}

pkg/monitoring/MonitoringExtension.php → pkg/monitoring/ConsumerMonitoringExtension.php

@@ -1,5 +1,7 @@
 <?php
 
+declare(strict_types=1);
+
 namespace Enqueue\Monitoring;
 
 use Enqueue\Consumption\Context\End;
@@ -20,7 +22,7 @@
 use Psr\Log\LoggerInterface;
 use Ramsey\Uuid\Uuid;
 
-class MonitoringExtension implements StartExtensionInterface, PreSubscribeExtensionInterface, PreConsumeExtensionInterface, EndExtensionInterface, ProcessorExceptionExtensionInterface, MessageReceivedExtensionInterface, MessageResultExtensionInterface
+class ConsumerMonitoringExtension implements StartExtensionInterface, PreSubscribeExtensionInterface, PreConsumeExtensionInterface, EndExtensionInterface, ProcessorExceptionExtensionInterface, MessageReceivedExtensionInterface, MessageResultExtensionInterface
 {
     /**
      * @var StatsStorage
@@ -310,7 +312,7 @@ private function safeCall(callable $fun, LoggerInterface $logger)
         try {
             return call_user_func($fun);
         } catch (\Throwable $e) {
-            $logger->error(sprintf('[MonitoringExtension] Push to storage failed: %s', $e->getMessage()));
+            $logger->error(sprintf('[ConsumerMonitoringExtension] Push to storage failed: %s', $e->getMessage()));
         }
 
         return null;

pkg/monitoring/GenericStatsStorageFactory.php

@@ -0,0 +1,25 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Enqueue\Monitoring;
+
+use Enqueue\Dsn\Dsn;
+
+class GenericStatsStorageFactory implements StatsStorageFactory
+{
+    public function create(string $dsn): StatsStorage
+    {
+        $schema = (new Dsn($dsn))->getScheme();
+
+        switch ($schema) {
+            case 'influxdb':
+                return new InfluxDbStorage($dsn);
+            case 'wamp':
+            case 'ws':
+                return new WampStorage($dsn);
+            default:
+                throw new \LogicException(sprintf('Unsupported stats storage: "%s"', $dsn));
+        }
+    }
+}

pkg/monitoring/StatsStorageFactory.php

@@ -0,0 +1,10 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Enqueue\Monitoring;
+
+interface StatsStorageFactory
+{
+    public function create(string $dsn): StatsStorage;
+}

pkg/monitoring/Tests/GenericStatsStorageFactoryTest.php

@@ -0,0 +1,42 @@
+<?php
+
+namespace Enqueue\Monitoring\Tests;
+
+use Enqueue\Monitoring\GenericStatsStorageFactory;
+use Enqueue\Monitoring\InfluxDbStorage;
+use Enqueue\Monitoring\StatsStorageFactory;
+use Enqueue\Monitoring\WampStorage;
+use Enqueue\Test\ClassExtensionTrait;
+use PHPUnit\Framework\TestCase;
+
+class GenericStatsStorageFactoryTest extends TestCase
+{
+    use ClassExtensionTrait;
+
+    public function testShouldImplementStatsStorageFactoryInterface()
+    {
+        $this->assertClassImplements(StatsStorageFactory::class, GenericStatsStorageFactory::class);
+    }
+
+    public function testShouldCreateInfluxDbStorage()
+    {
+        $storage = (new GenericStatsStorageFactory())->create('influxdb:');
+
+        $this->assertInstanceOf(InfluxDbStorage::class, $storage);
+    }
+
+    public function testShouldCreateWampStorage()
+    {
+        $storage = (new GenericStatsStorageFactory())->create('wamp:');
+
+        $this->assertInstanceOf(WampStorage::class, $storage);
+    }
+
+    public function testShouldThrowIfStorageIsNotSupported()
+    {
+        $this->expectException(\LogicException::class);
+        $this->expectExceptionMessage('Unsupported stats storage: "unsupported:"');
+
+        (new GenericStatsStorageFactory())->create('unsupported:');
+    }
+}