Queue Interop Support

composer.json

@@ -1,8 +1,8 @@
 {
     "name": "yiisoft/yii2-queue",
-    "description": "Yii2 Queue Extension which supported DB, Redis, RabbitMQ, Beanstalk and Gearman",
+    "description": "Yii2 Queue Extension which supported DB, Redis, RabbitMQ, Beanstalk, Gearman, Kafka, AWS SQS, STOMP, Filesystem",
     "type": "yii2-extension",
-    "keywords": ["yii", "queue", "async", "gii", "db", "redis", "rabbitmq", "beanstalk", "gearman"],
+    "keywords": ["yii", "queue", "async", "gii", "db", "redis", "rabbitmq", "beanstalk", "gearman", "amqp-interop", "queue-interop", "php-enqueue"],
     "license": "BSD-3-Clause",
     "authors": [
         {
@@ -22,19 +22,32 @@
     },
     "require-dev": {
         "yiisoft/yii2-redis": "*",
-        "php-amqplib/php-amqplib": "*",
+        "php-amqplib/php-amqplib": "2.7.x-dev",
         "pda/pheanstalk": "*",
         "jeremeamia/superclosure": "*",
         "yiisoft/yii2-debug": "*",
         "yiisoft/yii2-gii": "*",
-        "phpunit/phpunit": "~4.4"
+        "phpunit/phpunit": "~4.4",
+        "enqueue/amqp-lib": "0.7.x-dev"
     },
     "suggest": {
         "ext-pcntl": "Need for process signals.",
         "yiisoft/yii2-redis": "Need for Redis queue.",
         "pda/pheanstalk": "Need for Beanstalk queue.",
         "php-amqplib/php-amqplib": "Need for AMQP queue.",
-        "ext-gearman": "Need for Gearman queue."
+        "ext-gearman": "Need for Gearman queue.",
+        "enqueue/amqp-ext": "Required for support of AMQP with queue interop",
+        "enqueue/amqp-lib": "Required for support of AMQP with queue interop",
+        "enqueue/amqp-bunny": "Required for support of AMQP with queue interop",
+        "enqueue/amqp-tools": "Required for support of message delay with AMQP",
+        "enqueue/rdkafka": "Required for support of Kafka with queue interop",
+        "enqueue/sqs": "Required for support of AWS SQS with queue interop",
+        "enqueue/dbal": "Required for support of Doctrine Dbal queue with queue interop",
+        "enqueue/redis": "Required for support of Redis queue with queue interop",
+        "enqueue/fs": "Required for support of filesystem queue with queue interop",
+        "enqueue/gearman": "Required for support of Gearman queue with queue interop",
+        "enqueue/pheanstalk": "Required for support of Beanstalk queue with queue interop",
+        "enqueue/stomp": "Required for support of STOMP with queue interop"
     },
     "autoload": {
         "psr-4": {
@@ -45,7 +58,8 @@
             "yii\\queue\\file\\": "src/drivers/file",
             "yii\\queue\\gearman\\": "src/drivers/gearman",
             "yii\\queue\\redis\\": "src/drivers/redis",
-            "yii\\queue\\sync\\": "src/drivers/sync"
+            "yii\\queue\\sync\\": "src/drivers/sync",
+            "yii\\queue\\queue_interop\\": "src/drivers/queue_interop"
         }
     },
     "extra": {

docs/guide/README.md

@@ -21,3 +21,4 @@ Queue Drivers
 * [RabbitMQ](driver-amqp.md)
 * [Beanstalk](driver-beanstalk.md)
 * [Gearman](driver-gearman.md)
+* [Queue Interop (php-enqueue)](driver-queue-interop.md)

docs/guide/driver-queue-interop.md

@@ -0,0 +1,51 @@
+Queue Interop Driver (php-enqueue)
+==================================
+
+The driver works with many queue brokers.
+
+You can find full list of supported brokers on the [Queue Interop](https://github.com/queue-interop/queue-interop) page
+
+In order for it to work you need to install and configure one the implementations supported.
+
+Configuration example for the RabbitMQ AMQP:
+
+```php
+return [
+    'bootstrap' => [
+        'queue', // The component registers own console commands
+    ],
+    'components' => [
+        'queue' => [
+            'class' => \yii\queue\queue_interop\Queue::class,
+            'queueName' => 'queue',
+            'factoryClass' => \Enqueue\AmqpLib\AmqpConnectionFactory::class,
+            'factoryConfig' => [
+                'host' => 'localhost',
+                'port' => 5672,
+                'user' => 'guest',
+                'pass' => 'guest',
+                'vhost' => '/',
+            ],
+        ],
+    ],
+];
+```
+
+Console
+-------
+
+Console is used to listen and process queued tasks.
+
+```sh
+yii queue/listen
+```
+
+`listen` command launches a daemon which infinitely queries the queue. If there are new tasks
+they're immediately obtained and executed. This method is most efficient when command is properly
+daemonized via [supervisor](worker.md#supervisor).
+
+`listen` command has options:
+
+- `--verbose`, `-v`: print execution status into console.
+- `--isolate`: verbose mode of a job execution. If enabled, execution result of each job will be printed.
+- `--color`: highlighting for verbose mode.

src/drivers/queue_interop/Command.php

@@ -0,0 +1,22 @@
+<?php
+
+namespace yii\queue\queue_interop;
+
+use yii\queue\cli\Command as CliCommand;
+
+class Command extends CliCommand
+{
+    /**
+     * @var Queue
+     */
+    public $queue;
+
+    /**
+     * Listens queue and runs new jobs.
+     * It can be used as demon process.
+     */
+    public function actionListen()
+    {
+        $this->queue->listen();
+    }
+}

src/drivers/queue_interop/Queue.php

@@ -0,0 +1,153 @@
+<?php
+
+namespace yii\queue\queue_interop;
+
+use Enqueue\AmqpTools\DelayStrategyAware;
+use Enqueue\AmqpTools\RabbitMqDelayPluginDelayStrategy;
+use Enqueue\AmqpTools\RabbitMqDlxDelayStrategy;
+use Interop\Amqp\AmqpContext;
+use Interop\Amqp\AmqpQueue;
+use Interop\Queue\PsrConnectionFactory;
+use Interop\Queue\PsrContext;
+use Interop\Queue\PsrProducer;
+use yii\base\NotSupportedException;
+use yii\queue\cli\Queue as CliQueue;
+
+class Queue extends CliQueue
+{
+    const RABBITMQ_DELAY_DLX = 'dlx';
+    const RABBITMQ_DELAY_DELAYED_MESSAGE_PLUGIN = 'delayed_message_plugin';
+
+    /**
+     * {@inheritdoc}
+     */
+    public $commandClass = Command::class;
+
+    /**
+     * @var string
+     */
+    public $factoryClass = null;
+
+    /**
+     * @var array
+     */
+    public $factoryConfig = [];
+
+    /**
+     * Supported strategies: "dlx", "delayed_message_plugin"
+     *
+     * @var string
+     */
+    public $rabbitmqDelayStrategy = self::RABBITMQ_DELAY_DLX;
+
+    /**
+     * @var string
+     */
+    public $queueName = 'queue';
+
+    /**
+     * @var PsrContext
+     */
+    private $context;
+
+    /**
+     * Listens queue and runs new jobs.
+     */
+    public function listen()
+    {
+        $consumer = $this->getContext()->createConsumer(
+            $this->getContext()->createQueue($this->queueName)
+        );
+
+        while (true) {
+            if ($message = $consumer->receive()) {
+                list($ttr, $body) = explode(';', $message->getBody(), 2);
+                if ($this->handleMessage(null, $body, $ttr, 1)) {
+                    $consumer->acknowledge($message);
+                } else {
+                    $consumer->reject($message, true);
+                }
+            }
+        }
+    }
+
+    /**
+     * {@inheritdoc}
+     */
+    protected function pushMessage($message, $ttr, $delay, $priority)
+    {
+        $producer = $this->getContext()->createProducer();
+
+        if ($delay !== null) {
+            $producer->setDeliveryDelay($delay * 1000);
+        }
+
+        if  ($priority !== null) {
+            $producer->setPriority($priority);
+        }
+
+        $producer->send(
+            $this->getContext()->createQueue($this->queueName),
+            $this->getContext()->createMessage("$ttr;$message")
+        );
+
+        return null;
+    }
+
+    /**
+     * {@inheritdoc}
+     */
+    public function status($id)
+    {
+        throw new NotSupportedException('Status is not supported in the driver.');
+    }
+
+    /**
+     * @return PsrContext
+     */
+    private function getContext()
+    {
+        if (null === $this->context) {
+            if (empty($this->factoryClass)) {
+                throw new \LogicException('The "factoryClass" option is required');
+            }
+
+            if (false == class_exists($this->factoryClass)) {
+                throw new \LogicException(sprintf('The "factoryClass" option "%s" is not a class', $this->factoryClass));
+            }
+
+            if (false == is_a($this->factoryClass, PsrConnectionFactory::class)) {
+                throw new \LogicException(sprintf('The "factoryClass" option must contain a class that implements "%s" but it is not', PsrConnectionFactory::class));
+            }
+
+            /** @var PsrConnectionFactory $factory */
+            $factory = new $this->factoryClass(isset($this->factoryConfig['dsn']) ? $this->factoryConfig['dsn'] : $this->factoryConfig);
+
+            if ($factory instanceof DelayStrategyAware) {
+                if (false != $this->rabbitmqDelayStrategy) {
+                    switch ($this->rabbitmqDelayStrategy) {
+                        case self::RABBITMQ_DELAY_DLX:
+                            $factory->setDelayStrategy(new RabbitMqDlxDelayStrategy());
+                            break;
+                        case self::RABBITMQ_DELAY_DELAYED_MESSAGE_PLUGIN:
+                            $factory->setDelayStrategy(new RabbitMqDelayPluginDelayStrategy());
+                            break;
+                        default:
+                            throw new \LogicException(sprintf('Unknown rabbitmq delay strategy: "%s"', $this->rabbitmqDelayStrategy));
+                    }
+                }
+            }
+
+            $this->context = $factory->createContext();
+
+            if ($this->context instanceof AmqpContext) {
+                $queue = $this->context->createQueue($this->queueName);
+                $queue->addFlag(AmqpQueue::FLAG_DURABLE);
+
+                $this->context->declareQueue($queue);
+            }
+        }
+
+        return $this->context;
+    }
+}

tests/app/config/main.php

@@ -11,6 +11,7 @@
         'redisQueue',
         'amqpQueue',
         'beanstalkQueue',
+        'interopQueue',
     ],
     'components' => [
         'syncQueue' => [
@@ -75,6 +76,18 @@
         'beanstalkQueue' => [
             'class' => \yii\queue\beanstalk\Queue::class,
         ],
+        'interopQueue' => [
+            'class' => \yii\queue\queue_interop\Queue::class,
+            'queueName' => 'interop_queue',
+            'factoryClass' => \Enqueue\AmqpLib\AmqpConnectionFactory::class,
+            'factoryConfig' => [
+                'host' => 'localhost',
+                'port' => 5672,
+                'user' => 'guest',
+                'pass' => 'guest',
+                'vhost' => '/'
+            ]
+        ],
     ],
 ];
 

tests/drivers/queue_interop/QueueTest.php

@@ -0,0 +1,37 @@
+<?php
+
+namespace tests\drivers\queue_interop;
+
+use tests\drivers\CliTestCase;
+use yii\queue\queue_interop\Queue;
+
+class QueueTest extends CliTestCase
+{
+    /**
+     * @return Queue
+     */
+    protected function getQueue()
+    {
+        return \Yii::$app->interopQueue;
+    }
+
+    public function testRun()
+    {
+        // Not supported
+    }
+
+    public function testStatus()
+    {
+        // Not supported
+    }
+
+    public function testLater()
+    {
+        // Not supported
+    }
+
+    public function testRetry()
+    {
+        // Limited support
+    }
+}