Merge pull request #320 from formapro-forks/queue-interop

Add queue interop driver

README.md

@@ -14,6 +14,7 @@ Currently these are the supported backends, with more coming with each release:
  * Doctrine DBAL
  * Pheanstalk
  * PhpAmqp / RabbitMQ
+ * Queue interop
 
 You can learn more on our website about Bernard and its [related projects][website] or just dive directly into [the
 documentation][documentation].

composer.json

@@ -25,7 +25,9 @@
         "phpspec/phpspec": "^2.4",
         "phpunit/phpunit": "^5.5|^6.0",
         "iron-io/iron_mq": "~4.0",
-        "league/container": "~2.3"
+        "league/container": "~2.3",
+        "queue-interop/queue-interop": "^0.6",
+        "queue-interop/amqp-interop": "^0.6"
     },
     "suggest": {
         "php-amqplib/php-amqplib": "Allow sending messages to an AMQP server using php-amqplib",
@@ -34,7 +36,9 @@
         "pda/pheanstalk": "Allow sending messages to Beanstalk using pheanstalk",
         "predis/predis": "Allow sending messages to Redis using predis",
         "aws/aws-sdk-php": "Allow sending messages to AWS services like Simple Queue Service",
-        "mongodb/mongodb": "Allow sending messages to a MongoDB server via PHP Driver"
+        "mongodb/mongodb": "Allow sending messages to a MongoDB server via PHP Driver",
+        "queue-interop/queue-interop": "Allow sending messages using queue interop compatible transports",
+        "queue-interop/amqp-interop": "Allow sending messages using amqp interop compatible transports"
     },
 
     "autoload" : {

doc/drivers.rst

@@ -13,6 +13,7 @@ Several different types of drivers are supported. Currently these are available:
 * `Redis Extension`_
 * `Predis`_
 * `Amazon SQS`_
+* `Queue Interop`_
 
 Google AppEngine
 ----------------
@@ -475,3 +476,28 @@ require a HTTP request to amazon to be resolved.
     $driver = new SqsDriver($connection, array(
         'queue-name' => 'queue-url',
     ));
+
+Queue Interop
+-------------
+
+This driver adds ability to use any `queue interop <https://github.com/queue-interop/queue-interop#implementations>`_ compatible transport.
+For example we choose enqueue/fs one to demonstrate how it is working.
+
+.. code-block:: json
+
+    {
+        "require" : {
+            "enqueue/fs" : "^0.7"
+        }
+    }
+
+.. code-block:: php
+
+    <?php
+
+    use Bernard\Driver\InteropDriver;
+    use Enqueue\Fs\FsConnectionFactory;
+
+    $context = (new FsConnectionFactory('file://'.__DIR__.'/queues'))->createContext();
+
+    $driver = new InteropDriver($context);

src/Driver/InteropDriver.php

@@ -0,0 +1,150 @@
+<?php
+namespace Bernard\Driver;
+
+use Bernard\Driver;
+use Interop\Amqp\AmqpContext;
+use Interop\Amqp\AmqpQueue;
+use Interop\Queue\PsrConsumer;
+use Interop\Queue\PsrContext;
+
+final class InteropDriver implements Driver
+{
+    /**
+     * @var PsrContext
+     */
+    private $context;
+
+    /**
+     * @var PsrConsumer[]
+     */
+    private $consumers;
+
+    /**
+     * @param PsrContext $context
+     */
+    public function __construct(PsrContext $context)
+    {
+        $this->context = $context;
+
+        $this->consumers = [];
+    }
+
+    /**
+     * {@inheritdoc}
+     */
+    public function listQueues()
+    {
+        return [];
+    }
+
+    /**
+     * {@inheritdoc}
+     */
+    public function createQueue($queueName)
+    {
+        if ($this->context instanceof AmqpContext) {
+            $this->context->declareQueue($this->createAmqpQueue($queueName));
+        }
+    }
+
+    /**
+     * {@inheritdoc}
+     */
+    public function countMessages($queueName)
+    {
+        if ($this->context instanceof AmqpContext) {
+            return $this->context->declareQueue($this->createAmqpQueue($queueName));
+        }
+
+        return 0;
+    }
+
+    /**
+     * {@inheritdoc}
+     */
+    public function pushMessage($queueName, $message)
+    {
+        $queue = $this->context->createQueue($queueName);
+        $message = $this->context->createMessage($message);
+
+        $this->context->createProducer()->send($queue, $message);
+    }
+
+    /**
+     * {@inheritdoc}
+     */
+    public function popMessage($queueName, $duration = 5)
+    {
+        if ($message = $this->getQueueConsumer($queueName)->receive($duration * 1000)) {
+            return [$message->getBody(), $message];
+        }
+    }
+
+    /**
+     * {@inheritdoc}
+     */
+    public function acknowledgeMessage($queueName, $receipt)
+    {
+        $this->getQueueConsumer($queueName)->acknowledge($receipt);
+    }
+
+    /**
+     * {@inheritdoc}
+     */
+    public function peekQueue($queueName, $index = 0, $limit = 20)
+    {
+        return [];
+    }
+
+    /**
+     * {@inheritdoc}
+     */
+    public function removeQueue($queueName)
+    {
+        if ($this->context instanceof AmqpContext) {
+            $queue = $this->createAmqpQueue($queueName);
+
+            $this->context->deleteQueue($queue);
+        }
+    }
+
+    /**
+     * {@inheritdoc}
+     */
+    public function info()
+    {
+        return [];
+    }
+
+    /**
+     * @param string $queueName
+     *
+     * @return PsrConsumer
+     */
+    private function getQueueConsumer($queueName)
+    {
+        if (false == array_key_exists($queueName, $this->consumers)) {
+            $queue = $this->context->createQueue($queueName);
+
+            $this->consumers[$queueName] = $this->context->createConsumer($queue);
+        }
+
+        return $this->consumers[$queueName];
+    }
+
+    /**
+     * @param string $queueName
+     *
+     * @return AmqpQueue
+     */
+    private function createAmqpQueue($queueName)
+    {
+        /** @var AmqpContext $context */
+        $context = $this->context;
+
+        $queue = $context->createQueue($queueName);
+        $queue->addFlag(AmqpQueue::FLAG_DURABLE);
+
+        return $queue;
+    }
+}