Merge pull request #217 from php-enqueue/amqp-add-basic-consume-support

[BC break] Amqp add basic consume support

Author: makasim

composer.json

@@ -27,8 +27,8 @@
     "enqueue/test": "*@dev",
     "enqueue/async-event-dispatcher": "*@dev",
     "queue-interop/queue-interop": "^0.6@dev",
-    "queue-interop/amqp-interop": "^0.6@dev",
-    "queue-interop/queue-spec": "^0.5@dev",
+    "queue-interop/amqp-interop": "^0.7@dev",
+    "queue-interop/queue-spec": "^0.5.1@dev",
 
     "phpunit/phpunit": "^5",
     "doctrine/doctrine-bundle": "~1.2",

docker/Dockerfile

@@ -17,6 +17,7 @@ RUN echo "extension=rdkafka.so" > /etc/php/7.1/cli/conf.d/10-rdkafka.ini
 RUN echo "extension=rdkafka.so" > /etc/php/7.1/fpm/conf.d/10-rdkafka.ini
 
 COPY ./php/cli.ini /etc/php/7.1/cli/conf.d/1-dev_cli.ini
+COPY ./php/amqp.so /usr/lib/php/20160303/amqp.so
 COPY ./bin/dev_entrypoiny.sh /usr/local/bin/entrypoint.sh
 RUN chmod u+x /usr/local/bin/entrypoint.sh
 

docker/php/amqp.so

@@ -7,4 +7,5 @@ parameters:
 		- pkg/redis/PhpRedis.php
 		- pkg/redis/RedisConnectionFactory.php
 		- pkg/gearman
-		- pkg/amqp-ext/AmqpConsumer.php
+		- pkg/amqp-ext/AmqpConsumer.php
+		- pkg/amqp-ext/AmqpContext.php

phpstan.neon

@@ -3,9 +3,12 @@
 namespace Enqueue\AmqpBunny;
 
 use Bunny\Channel;
+use Bunny\Client;
+use Bunny\Message;
 use Enqueue\AmqpTools\DelayStrategyAware;
 use Enqueue\AmqpTools\DelayStrategyAwareTrait;
 use Interop\Amqp\AmqpBind as InteropAmqpBind;
+use Interop\Amqp\AmqpConsumer as InteropAmqpConsumer;
 use Interop\Amqp\AmqpContext as InteropAmqpContext;
 use Interop\Amqp\AmqpMessage as InteropAmqpMessage;
 use Interop\Amqp\AmqpQueue as InteropAmqpQueue;
@@ -43,6 +46,13 @@ class AmqpContext implements InteropAmqpContext, DelayStrategyAware
      */
     private $buffer;
 
+    /**
+     * an item contains an array: [AmqpConsumerInterop $consumer, callable $callback];.
+     *
+     * @var array
+     */
+    private $subscribers;
+
     /**
      * Callable must return instance of \Bunny\Channel once called.
      *
@@ -309,6 +319,77 @@ public function setQos($prefetchSize, $prefetchCount, $global)
         $this->getBunnyChannel()->qos($prefetchSize, $prefetchCount, $global);
     }
 
+    /**
+     * {@inheritdoc}
+     */
+    public function subscribe(InteropAmqpConsumer $consumer, callable $callback)
+    {
+        if ($consumer->getConsumerTag() && array_key_exists($consumer->getConsumerTag(), $this->subscribers)) {
+            return;
+        }
+
+        $bunnyCallback = function (Message $message, Channel $channel, Client $bunny) {
+            $receivedMessage = $this->convertMessage($message);
+            $receivedMessage->setConsumerTag($message->consumerTag);
+
+            /**
+             * @var AmqpConsumer
+             * @var callable     $callback
+             */
+            list($consumer, $callback) = $this->subscribers[$message->consumerTag];
+
+            if (false === call_user_func($callback, $receivedMessage, $consumer)) {
+                $bunny->stop();
+            }
+        };
+
+        $frame = $this->getBunnyChannel()->consume(
+            $bunnyCallback,
+            $consumer->getQueue()->getQueueName(),
+            $consumer->getConsumerTag(),
+            (bool) ($consumer->getFlags() & InteropAmqpConsumer::FLAG_NOLOCAL),
+            (bool) ($consumer->getFlags() & InteropAmqpConsumer::FLAG_NOACK),
+            (bool) ($consumer->getFlags() & InteropAmqpConsumer::FLAG_EXCLUSIVE),
+            (bool) ($consumer->getFlags() & InteropAmqpConsumer::FLAG_NOWAIT)
+        );
+
+        if (empty($frame->consumerTag)) {
+            throw new Exception('Got empty consumer tag');
+        }
+
+        $consumer->setConsumerTag($frame->consumerTag);
+
+        $this->subscribers[$frame->consumerTag] = [$consumer, $callback];
+    }
+
+    /**
+     * {@inheritdoc}
+     */
+    public function unsubscribe(InteropAmqpConsumer $consumer)
+    {
+        if (false == $consumer->getConsumerTag()) {
+            return;
+        }
+
+        $consumerTag = $consumer->getConsumerTag();
+
+        $this->getBunnyChannel()->cancel($consumerTag);
+        $consumer->setConsumerTag(null);
+        unset($this->subscribers[$consumerTag]);
+    }
+
+    /**
+     * {@inheritdoc}
+     */
+    public function consume($timeout = 0)
+    {
+        if (empty($this->subscribers)) {
+            throw new \LogicException('There is no subscribers. Consider calling basicConsumeSubscribe before consuming');
+        }
+
+        $this->getBunnyChannel()->getClient()->run($timeout / 1000);
+    }
+
     /**
      * @return Channel
      */
@@ -328,4 +409,27 @@ public function getBunnyChannel()
 
         return $this->bunnyChannel;
     }
+
+    /**
+     * @param Message $bunnyMessage
+     *
+     * @return InteropAmqpMessage
+     */
+    private function convertMessage(Message $bunnyMessage)
+    {
+        $headers = $bunnyMessage->headers;
+
+        $properties = [];
+        if (isset($headers['application_headers'])) {
+            $properties = $headers['application_headers'];
+        }
+        unset($headers['application_headers']);
+
+        $message = new AmqpMessage($bunnyMessage->content, $properties, $headers);
+        $message->setDeliveryTag($bunnyMessage->deliveryTag);
+        $message->setRedelivered($bunnyMessage->redelivered);
+        $message->setRoutingKey($bunnyMessage->routingKey);
+
+        return $message;
+    }
 }

pkg/amqp-bunny/AmqpContext.php

@@ -0,0 +1,22 @@
+<?php
+
+namespace Enqueue\AmqpBunny\Tests\Spec;
+
+use Enqueue\AmqpBunny\AmqpConnectionFactory;
+use Interop\Queue\Spec\Amqp\BasicConsumeBreakOnFalseSpec;
+
+/**
+ * @group functional
+ */
+class AmqpBasicConsumeBreakOnFalseTest extends BasicConsumeBreakOnFalseSpec
+{
+    /**
+     * {@inheritdoc}
+     */
+    protected function createContext()
+    {
+        $factory = new AmqpConnectionFactory(getenv('AMQP_DSN'));
+
+        return $factory->createContext();
+    }
+}

pkg/amqp-bunny/Tests/Spec/AmqpBasicConsumeBreakOnFalseTest.php

@@ -0,0 +1,22 @@
+<?php
+
+namespace Enqueue\AmqpBunny\Tests\Spec;
+
+use Enqueue\AmqpBunny\AmqpConnectionFactory;
+use Interop\Queue\Spec\Amqp\BasicConsumeFromAllSubscribedQueuesSpec;
+
+/**
+ * @group functional
+ */
+class AmqpBasicConsumeFromAllSubscribedQueuesTest extends BasicConsumeFromAllSubscribedQueuesSpec
+{
+    /**
+     * {@inheritdoc}
+     */
+    protected function createContext()
+    {
+        $factory = new AmqpConnectionFactory(getenv('AMQP_DSN'));
+
+        return $factory->createContext();
+    }
+}

pkg/amqp-bunny/Tests/Spec/AmqpBasicConsumeFromAllSubscribedQueuesTest.php

@@ -0,0 +1,22 @@
+<?php
+
+namespace Enqueue\AmqpBunny\Tests\Spec;
+
+use Enqueue\AmqpBunny\AmqpConnectionFactory;
+use Interop\Queue\Spec\Amqp\BasicConsumeShouldAddConsumerTagOnSubscribeSpec;
+
+/**
+ * @group functional
+ */
+class AmqpBasicConsumeShouldAddConsumerTagOnSubscribeTest extends BasicConsumeShouldAddConsumerTagOnSubscribeSpec
+{
+    /**
+     * {@inheritdoc}
+     */
+    protected function createContext()
+    {
+        $factory = new AmqpConnectionFactory(getenv('AMQP_DSN'));
+
+        return $factory->createContext();
+    }
+}

pkg/amqp-bunny/Tests/Spec/AmqpBasicConsumeShouldAddConsumerTagOnSubscribeTest.php

@@ -0,0 +1,22 @@
+<?php
+
+namespace Enqueue\AmqpBunny\Tests\Spec;
+
+use Enqueue\AmqpBunny\AmqpConnectionFactory;
+use Interop\Queue\Spec\Amqp\BasicConsumeShouldRemoveConsumerTagOnUnsubscribeSpec;
+
+/**
+ * @group functional
+ */
+class AmqpBasicConsumeShouldRemoveConsumerTagOnUnsubscribeTest extends BasicConsumeShouldRemoveConsumerTagOnUnsubscribeSpec
+{
+    /**
+     * {@inheritdoc}
+     */
+    protected function createContext()
+    {
+        $factory = new AmqpConnectionFactory(getenv('AMQP_DSN'));
+
+        return $factory->createContext();
+    }
+}

pkg/amqp-bunny/Tests/Spec/AmqpBasicConsumeShouldRemoveConsumerTagOnUnsubscribeTest.php

@@ -0,0 +1,22 @@
+<?php
+
+namespace Enqueue\AmqpBunny\Tests\Spec;
+
+use Enqueue\AmqpBunny\AmqpConnectionFactory;
+use Interop\Queue\Spec\Amqp\BasicConsumeUntilUnsubscribedSpec;
+
+/**
+ * @group functional
+ */
+class AmqpBasicConsumeUntilUnsubscribedTest extends BasicConsumeUntilUnsubscribedSpec
+{
+    /**
+     * {@inheritdoc}
+     */
+    protected function createContext()
+    {
+        $factory = new AmqpConnectionFactory(getenv('AMQP_DSN'));
+
+        return $factory->createContext();
+    }
+}