Merge pull request #2 from php-enqueue/amqp-interop-support

Amqp queue

Author: makasim

composer.json

@@ -13,7 +13,9 @@
     "require": {
         "php": ">=5.6",
         "illuminate/queue": "^5.4",
-        "queue-interop/queue-interop": "^0.6@dev"
+        "queue-interop/queue-interop": "^0.6@dev",
+        "queue-interop/amqp-interop": "^0.6@dev",
+        "enqueue/amqp-tools": "^0.7@dev"
     },
     "require-dev": {
         "phpunit/phpunit": "~5.5",
@@ -22,7 +24,7 @@
         "enqueue/test": "^0.7@dev"
     },
     "autoload": {
-        "psr-4": { "Enqueue\\LaravelQueue\\": "" },
+        "psr-4": { "Enqueue\\LaravelQueue\\": "src/" },
         "exclude-from-classmap": [
             "/Tests/"
         ]

src/AmqpConnector.php

@@ -0,0 +1,27 @@
+<?php
+
+namespace Enqueue\LaravelQueue;
+
+use Enqueue\AmqpTools\DelayStrategyAware;
+use Enqueue\AmqpTools\RabbitMqDlxDelayStrategy;
+use Interop\Amqp\AmqpContext;
+
+class AmqpConnector extends Connector
+{
+    public function connect(array $config)
+    {
+        $queue = parent::connect($config);
+
+        /** @var AmqpContext $amqpContext */
+        $amqpContext = $queue->getPsrContext();
+        if (false == $amqpContext instanceof AmqpContext) {
+            throw new \LogicException(sprintf('The context must be instance of "%s" but got "%s"', AmqpContext::class, get_class($queue->getPsrContext())));
+        }
+
+        if ($amqpContext instanceof DelayStrategyAware) {
+            $amqpContext->setDelayStrategy(new RabbitMqDlxDelayStrategy());
+        }
+
+        return $queue;
+    }
+}

src/AmqpQueue.php

@@ -0,0 +1,62 @@
+<?php
+
+namespace Enqueue\LaravelQueue;
+
+use Interop\Amqp\AmqpContext;
+
+/**
+ * @method AmqpContext getPsrContext()
+ */
+class AmqpQueue extends Queue
+{
+    /**
+     * {@inheritdoc}
+     *
+     * @param AmqpContext $psrContext
+     */
+    public function __construct(AmqpContext $psrContext, $queueName, $timeToRun)
+    {
+        parent::__construct($psrContext, $queueName, $timeToRun);
+    }
+
+    /**
+     * {@inheritdoc}
+     */
+    public function pushRaw($payload, $queue = null, array $options = [])
+    {
+        $this->declareQueue($queue);
+
+        parent::pushRaw($payload, $queue, $options);
+    }
+
+    /**
+     * {@inheritdoc}
+     */
+    public function later($delay, $job, $data = '', $queue = null)
+    {
+        $this->declareQueue($queue);
+
+        return parent::later($delay, $job, $data, $queue);
+    }
+
+    /**
+     * {@inheritdoc}
+     */
+    public function pop($queue = null)
+    {
+        $this->declareQueue($queue);
+
+        return parent::pop($queue);
+    }
+
+    /**
+     * @param string|null $queue
+     */
+    protected function declareQueue($queue = null)
+    {
+        $psrQueue = $this->getPsrContext()->createQueue($this->getQueue($queue));
+        $psrQueue->addFlag(\Interop\Amqp\AmqpQueue::FLAG_DURABLE);
+
+        $this->getPsrContext()->declareQueue($psrQueue);
+    }
+}

Command/ConsumeMessagesCommand.php renamed to src/Command/ConsumeMessagesCommand.php

@@ -65,5 +65,9 @@ private function bootInteropQueueDriver()
         $manager->addConnector('interop', function () {
             return new Connector();
         });
+
+        $manager->addConnector('amqp_interop', function () {
+            return new AmqpConnector();
+        });
     }
 }

Command/ProduceMessageCommand.php renamed to src/Command/ProduceMessageCommand.php

@@ -17,10 +17,11 @@ class Queue extends BaseQueue implements QueueContract
      * @var int
      */
     protected $timeToRun;
+
     /**
      * @var PsrContext
      */
-    private $psrContext;
+    protected $psrContext;
 
     /**
      * @param PsrContext $psrContext