Dbal Subscription Consumer feature

docs/transport/amqp.md

@@ -235,7 +235,7 @@ use Interop\Queue\PsrConsumer;
 $fooConsumer = $psrContext->createConsumer($fooQueue);
 $barConsumer = $psrContext->createConsumer($barQueue);
 
-$subscriptionConsumer =$psrContext->createSubscriptionConsumer();
+$subscriptionConsumer = $psrContext->createSubscriptionConsumer();
 $subscriptionConsumer->subscribe($fooConsumer, function(PsrMessage $message, PsrConsumer $consumer) {
     // process message
 

docs/transport/amqp_bunny.md

@@ -227,7 +227,7 @@ use Interop\Queue\PsrConsumer;
 $fooConsumer = $psrContext->createConsumer($fooQueue);
 $barConsumer = $psrContext->createConsumer($barQueue);
 
-$subscriptionConsumer =$psrContext->createSubscriptionConsumer();
+$subscriptionConsumer = $psrContext->createSubscriptionConsumer();
 $subscriptionConsumer->subscribe($fooConsumer, function(PsrMessage $message, PsrConsumer $consumer) {
     // process message
 

docs/transport/amqp_lib.md

@@ -235,7 +235,7 @@ use Interop\Queue\PsrConsumer;
 $fooConsumer = $psrContext->createConsumer($fooQueue);
 $barConsumer = $psrContext->createConsumer($barQueue);
 
-$subscriptionConsumer =$psrContext->createSubscriptionConsumer();
+$subscriptionConsumer = $psrContext->createSubscriptionConsumer();
 $subscriptionConsumer->subscribe($fooConsumer, function(PsrMessage $message, PsrConsumer $consumer) {
     // process message
 

docs/transport/dbal.md

@@ -11,6 +11,7 @@ It creates a table there. Pushes and pops messages to\from that table.
 * [Send message to topic](#send-message-to-topic)
 * [Send message to queue](#send-message-to-queue)
 * [Consume message](#consume-message)
+* [Subscription consumer](#subscription-consumer)
 
 ## Installation
 
@@ -103,4 +104,37 @@ $message = $consumer->receive();
 // process a message
 ```
 
+## Subscription consumer
+
+```php
+<?php
+use Interop\Queue\PsrMessage;
+use Interop\Queue\PsrConsumer;
+
+/** @var \Enqueue\Dbal\DbalContext $psrContext */
+/** @var \Enqueue\Dbal\DbalDestination $fooQueue */
+/** @var \Enqueue\Dbal\DbalDestination $barQueue */
+
+$fooConsumer = $psrContext->createConsumer($fooQueue);
+$barConsumer = $psrContext->createConsumer($barQueue);
+
+$subscriptionConsumer = $psrContext->createSubscriptionConsumer();
+$subscriptionConsumer->subscribe($fooConsumer, function(PsrMessage $message, PsrConsumer $consumer) {
+    // process message
+
+    $consumer->acknowledge($message);
+
+    return true;
+});
+$subscriptionConsumer->subscribe($barConsumer, function(PsrMessage $message, PsrConsumer $consumer) {
+    // process message
+
+    $consumer->acknowledge($message);
+
+    return true;
+});
+
+$subscriptionConsumer->consume(2000); // 2 sec
+```
+
 [back to index](../index.md)

pkg/dbal/DbalConsumer.php

@@ -142,7 +142,7 @@ protected function receiveMessage(): ?DbalMessage
             $this->dbal->commit();
 
             if (empty($dbalMessage['time_to_live']) || ($dbalMessage['time_to_live'] / 1000) > microtime(true)) {
-                return $this->convertMessage($dbalMessage);
+                return $this->context->convertMessage($dbalMessage);
             }
 
             return null;
@@ -153,27 +153,6 @@ protected function receiveMessage(): ?DbalMessage
         }
     }
 
-    protected function convertMessage(array $dbalMessage): DbalMessage
-    {
-        /** @var DbalMessage $message */
-        $message = $this->context->createMessage();
-
-        $message->setBody($dbalMessage['body']);
-        $message->setPriority((int) $dbalMessage['priority']);
-        $message->setRedelivered((bool) $dbalMessage['redelivered']);
-        $message->setPublishedAt((int) $dbalMessage['published_at']);
-
-        if ($dbalMessage['headers']) {
-            $message->setHeaders(JSON::decode($dbalMessage['headers']));
-        }
-
-        if ($dbalMessage['properties']) {
-            $message->setProperties(JSON::decode($dbalMessage['properties']));
-        }
-
-        return $message;
-    }
-
     private function fetchPrioritizedMessage(int $now): ?array
     {
         $query = $this->dbal->createQueryBuilder();

pkg/dbal/DbalContext.php

@@ -11,7 +11,6 @@
 use Interop\Queue\Context;
 use Interop\Queue\Destination;
 use Interop\Queue\Exception\InvalidDestinationException;
-use Interop\Queue\Exception\SubscriptionConsumerNotSupportedException;
 use Interop\Queue\Exception\TemporaryQueueNotSupportedException;
 use Interop\Queue\Message;
 use Interop\Queue\Producer;
@@ -126,7 +125,31 @@ public function close(): void
 
     public function createSubscriptionConsumer(): SubscriptionConsumer
     {
-        throw SubscriptionConsumerNotSupportedException::providerDoestNotSupportIt();
+        return new DbalSubscriptionConsumer($this);
+    }
+
+    /**
+     * @internal It must be used here and in the consumer only
+     */
+    public function convertMessage(array $dbalMessage): DbalMessage
+    {
+        $dbalMessageObj = $this->createMessage(
+            $dbalMessage['body'],
+            $dbalMessage['properties'] ? JSON::decode($dbalMessage['properties']) : [],
+            $dbalMessage['headers'] ? JSON::decode($dbalMessage['headers']) : []
+        );
+
+        if (isset($dbalMessage['redelivered'])) {
+            $dbalMessageObj->setRedelivered((bool) $dbalMessage['redelivered']);
+        }
+        if (isset($dbalMessage['priority'])) {
+            $dbalMessageObj->setPriority((int) $dbalMessage['priority']);
+        }
+        if (isset($dbalMessage['published_at'])) {
+            $dbalMessageObj->setPublishedAt((int) $dbalMessage['published_at']);
+        }
+
+        return $dbalMessageObj;
     }
 
     /**

pkg/dbal/DbalSubscriptionConsumer.php

@@ -0,0 +1,198 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Enqueue\Dbal;
+
+use Doctrine\DBAL\Types\Type;
+use Interop\Queue\Consumer;
+use Interop\Queue\SubscriptionConsumer;
+
+class DbalSubscriptionConsumer implements SubscriptionConsumer
+{
+    /**
+     * @var DbalContext
+     */
+    private $context;
+
+    /**
+     * an item contains an array: [DbalConsumer $consumer, callable $callback];.
+     *
+     * @var array
+     */
+    private $subscribers;
+
+    /**
+     * @var \Doctrine\DBAL\Connection
+     */
+    private $dbal;
+
+    /**
+     * @param DbalContext $context
+     */
+    public function __construct(DbalContext $context)
+    {
+        $this->context = $context;
+        $this->dbal = $this->context->getDbalConnection();
+        $this->subscribers = [];
+    }
+
+    public function consume(int $timeout = 0): void
+    {
+        if (empty($this->subscribers)) {
+            throw new \LogicException('No subscribers');
+        }
+
+        $timeout = (int) ceil($timeout / 1000);
+        $endAt = time() + $timeout;
+
+        $queueNames = [];
+        foreach (array_keys($this->subscribers) as $queueName) {
+            $queueNames[$queueName] = $queueName;
+        }
+
+        $currentQueueNames = [];
+        while (true) {
+            if (empty($currentQueueNames)) {
+                $currentQueueNames = $queueNames;
+            }
+
+            $message = $this->fetchPrioritizedMessage($currentQueueNames) ?: $this->fetchMessage($currentQueueNames);
+
+            if ($message) {
+                $this->dbal->delete($this->context->getTableName(), ['id' => $message['id']], ['id' => Type::GUID]);
+
+                $dbalMessage = $this->context->convertMessage($message);
+
+                /**
+                 * @var DbalConsumer
+                 * @var callable     $callback
+                 */
+                list($consumer, $callback) = $this->subscribers[$message['queue']];
+
+                if (false === call_user_func($callback, $dbalMessage, $consumer)) {
+                    return;
+                }
+
+                unset($currentQueueNames[$message['queue']]);
+            } else {
+                $currentQueueNames = [];
+
+                usleep(200000); // 200ms
+            }
+
+            if ($timeout && microtime(true) >= $endAt) {
+                return;
+            }
+        }
+    }
+
+    /**
+     * @param DbalConsumer $consumer
+     */
+    public function subscribe(Consumer $consumer, callable $callback): void
+    {
+        if (false == $consumer instanceof DbalConsumer) {
+            throw new \InvalidArgumentException(sprintf('The consumer must be instance of "%s" got "%s"', DbalConsumer::class, get_class($consumer)));
+        }
+
+        $queueName = $consumer->getQueue()->getQueueName();
+        if (array_key_exists($queueName, $this->subscribers)) {
+            if ($this->subscribers[$queueName][0] === $consumer && $this->subscribers[$queueName][1] === $callback) {
+                return;
+            }
+
+            throw new \InvalidArgumentException(sprintf('There is a consumer subscribed to queue: "%s"', $queueName));
+        }
+
+        $this->subscribers[$queueName] = [$consumer, $callback];
+    }
+
+    /**
+     * @param DbalConsumer $consumer
+     */
+    public function unsubscribe(Consumer $consumer): void
+    {
+        if (false == $consumer instanceof DbalConsumer) {
+            throw new \InvalidArgumentException(sprintf('The consumer must be instance of "%s" got "%s"', DbalConsumer::class, get_class($consumer)));
+        }
+
+        $queueName = $consumer->getQueue()->getQueueName();
+
+        if (false == array_key_exists($queueName, $this->subscribers)) {
+            return;
+        }
+
+        if ($this->subscribers[$queueName][0] !== $consumer) {
+            return;
+        }
+
+        unset($this->subscribers[$queueName]);
+    }
+
+    public function unsubscribeAll(): void
+    {
+        $this->subscribers = [];
+    }
+
+    private function fetchMessage(array $queues): ?array
+    {
+        $query = $this->dbal->createQueryBuilder();
+        $query
+            ->select('*')
+            ->from($this->context->getTableName())
+            ->andWhere('queue IN (:queues)')
+            ->andWhere('priority IS NULL')
+            ->andWhere('(delayed_until IS NULL OR delayed_until <= :delayedUntil)')
+            ->addOrderBy('published_at', 'asc')
+            ->setMaxResults(1)
+        ;
+
+        $sql = $query->getSQL().' '.$this->dbal->getDatabasePlatform()->getWriteLockSQL();
+
+        $result = $this->dbal->executeQuery(
+            $sql,
+            [
+                'queues' => array_keys($queues),
+                'delayedUntil' => time(),
+            ],
+            [
+                'queues' => \Doctrine\DBAL\Connection::PARAM_STR_ARRAY,
+                'delayedUntil' => \Doctrine\DBAL\ParameterType::INTEGER,
+            ]
+        )->fetch();
+
+        return $result ?: null;
+    }
+
+    private function fetchPrioritizedMessage(array $queues): ?array
+    {
+        $query = $this->dbal->createQueryBuilder();
+        $query
+            ->select('*')
+            ->from($this->context->getTableName())
+            ->andWhere('queue IN (:queues)')
+            ->andWhere('priority IS NOT NULL')
+            ->andWhere('(delayed_until IS NULL OR delayed_until <= :delayedUntil)')
+            ->addOrderBy('published_at', 'asc')
+            ->addOrderBy('priority', 'desc')
+            ->setMaxResults(1)
+        ;
+
+        $sql = $query->getSQL().' '.$this->dbal->getDatabasePlatform()->getWriteLockSQL();
+
+        $result = $this->dbal->executeQuery(
+            $sql,
+            [
+                'queues' => array_keys($queues),
+                'delayedUntil' => time(),
+            ],
+            [
+                'queues' => \Doctrine\DBAL\Connection::PARAM_STR_ARRAY,
+                'delayedUntil' => \Doctrine\DBAL\ParameterType::INTEGER,
+            ]
+        )->fetch();
+
+        return $result ?: null;
+    }
+}

pkg/dbal/Tests/DbalContextTest.php

@@ -64,6 +64,21 @@ public function testShouldCreateMessage()
         $this->assertFalse($message->isRedelivered());
     }
 
+    public function testShouldConvertArrayToDbalMessage()
+    {
+        $arrayData = [
+            'body' => 'theBody',
+            'properties' => json_encode(['barProp' => 'barPropVal']),
+            'headers' => json_encode(['fooHeader' => 'fooHeaderVal']),
+        ];
+        $context = new DbalContext($this->createConnectionMock());
+        $message = $context->convertMessage($arrayData);
+
+        $this->assertSame('theBody', $message->getBody());
+        $this->assertSame(['barProp' => 'barPropVal'], $message->getProperties());
+        $this->assertSame(['fooHeader' => 'fooHeaderVal'], $message->getHeaders());
+    }
+
     public function testShouldCreateTopic()
     {
         $context = new DbalContext($this->createConnectionMock());

pkg/dbal/Tests/DbalMessageTest.php

@@ -1,5 +1,7 @@
 <?php
 
+declare(strict_types=1);
+
 namespace Enqueue\Dbal\Tests;
 
 use Enqueue\Dbal\DbalMessage;