[amqp-bunny] implement basic consume related methods.

makasim · makasim · commit 8dfda8f98ffc · 2017-10-10T20:53:48.000+03:00
diff --git a/pkg/amqp-bunny/AmqpContext.php b/pkg/amqp-bunny/AmqpContext.php
@@ -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;
+    }
 }
diff --git a/pkg/amqp-bunny/Tests/Spec/AmqpBasicConsumeShouldAddConsumerTagOnSubscribeTest.php b/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\BasicConsumeShouldAddConsumerTagOnSubscribeSpec;
+
+/**
+ * @group functional
+ */
+class AmqpBasicConsumeShouldAddConsumerTagOnSubscribeTest extends BasicConsumeShouldAddConsumerTagOnSubscribeSpec
+{
+    /**
+     * {@inheritdoc}
+     */
+    protected function createContext()
+    {
+        $factory = new AmqpConnectionFactory(getenv('AMQP_DSN'));
+
+        return $factory->createContext();
+    }
+}
diff --git a/pkg/amqp-bunny/Tests/Spec/AmqpBasicConsumeShouldRemoveConsumerTagOnUnsubscribeTest.php b/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\BasicConsumeShouldRemoveConsumerTagOnUnsubscribeSpec;
+
+/**
+ * @group functional
+ */
+class AmqpBasicConsumeShouldRemoveConsumerTagOnUnsubscribeTest extends BasicConsumeShouldRemoveConsumerTagOnUnsubscribeSpec
+{
+    /**
+     * {@inheritdoc}
+     */
+    protected function createContext()
+    {
+        $factory = new AmqpConnectionFactory(getenv('AMQP_DSN'));
+
+        return $factory->createContext();
+    }
+}
diff --git a/pkg/amqp-bunny/Tests/Spec/AmqpBasicConsumeUntilUnsubscribedTest.php b/pkg/amqp-bunny/Tests/Spec/AmqpBasicConsumeUntilUnsubscribedTest.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();
+    }
+}
diff --git a/pkg/amqp-ext/AmqpContext.php b/pkg/amqp-ext/AmqpContext.php
@@ -43,9 +43,11 @@ class AmqpContext implements InteropAmqpContext, DelayStrategyAware
     private $receiveMethod;
 
     /**
+     * an item contains an array: [AmqpConsumerInterop $consumer, callable $callback];.
+     *
      * @var array
      */
-    private $basicConsumeSubscribers;
+    private $subscribers;
 
     /**
      * Callable must return instance of \AMQPChannel once called.
@@ -66,7 +68,7 @@ public function __construct($extChannel, $receiveMethod)
         }
 
         $this->buffer = new Buffer();
-        $this->basicConsumeSubscribers = [];
+        $this->subscribers = [];
     }
 
     /**
@@ -298,23 +300,11 @@ public function getExtChannel()
     }
 
     /**
-     * Notify broker that the channel is interested in consuming messages from this queue.
-     *
-     * @param InteropAmqpConsumer $consumer
-     * @param callable            $callback A callback function to which the
-     *                                      consumed message will be passed. The
-     *                                      function must accept at a minimum
-     *                                      one parameter, an \Interop\Amqp\AmqpMessage object,
-     *                                      and an optional second parameter
-     *                                      the \Interop\Amqp\AmqpConsumer from which the message was
-     *                                      consumed. The \Interop\Amqp\AmqpContext::basicConsume() will
-     *                                      not return the processing thread back to
-     *                                      the PHP script until the callback
-     *                                      function returns FALSE.
+     * {@inheritdoc}
      */
-    public function basicConsumeSubscribe(InteropAmqpConsumer $consumer, callable $callback)
+    public function subscribe(InteropAmqpConsumer $consumer, callable $callback)
     {
-        if ($consumer->getConsumerTag() && array_key_exists($consumer->getConsumerTag(), $this->basicConsumeSubscribers)) {
+        if ($consumer->getConsumerTag() && array_key_exists($consumer->getConsumerTag(), $this->subscribers)) {
             return;
         }
 
@@ -325,10 +315,13 @@ public function basicConsumeSubscribe(InteropAmqpConsumer $consumer, callable $c
 
         $consumerTag = $extQueue->getConsumerTag();
         $consumer->setConsumerTag($consumerTag);
-        $this->basicConsumeSubscribers[$consumerTag] = [$consumer, $callback];
+        $this->subscribers[$consumerTag] = [$consumer, $callback];
     }
 
-    public function basicConsumeUnsubscribe(InteropAmqpConsumer $consumer)
+    /**
+     * {@inheritdoc}
+     */
+    public function unsubscribe(InteropAmqpConsumer $consumer)
     {
         if (false == $consumer->getConsumerTag()) {
             return;
@@ -341,15 +334,15 @@ public function basicConsumeUnsubscribe(InteropAmqpConsumer $consumer)
         $extQueue->setName($consumer->getQueue()->getQueueName());
 
         $extQueue->cancel($consumerTag);
-        unset($this->basicConsumeSubscribers[$consumerTag]);
+        unset($this->subscribers[$consumerTag]);
     }
 
     /**
-     * @param float|int $timeout milliseconds, consumes endlessly if zero set
+     * {@inheritdoc}
      */
-    public function basicConsume($timeout = 0)
+    public function consume($timeout = 0)
     {
-        if (empty($this->basicConsumeSubscribers)) {
+        if (empty($this->subscribers)) {
             throw new \LogicException('There is no subscribers. Consider calling basicConsumeSubscribe before consuming');
         }
 
@@ -360,9 +353,9 @@ public function basicConsume($timeout = 0)
         try {
             $extConnection->setReadTimeout($timeout / 1000);
 
-            reset($this->basicConsumeSubscribers);
+            reset($this->subscribers);
             /** @var $consumer AmqpConsumer */
-            list($consumer) = current($this->basicConsumeSubscribers);
+            list($consumer) = current($this->subscribers);
 
             $extQueue = new \AMQPQueue($this->getExtChannel());
             $extQueue->setName($consumer->getQueue()->getQueueName());
@@ -374,7 +367,7 @@ public function basicConsume($timeout = 0)
                  * @var AmqpConsumer
                  * @var callable     $callback
                  */
-                list($consumer, $callback) = $this->basicConsumeSubscribers[$q->getConsumerTag()];
+                list($consumer, $callback) = $this->subscribers[$q->getConsumerTag()];
 
                 return call_user_func($callback, $message, $consumer);
             }, AMQP_JUST_CONSUME);
diff --git a/pkg/amqp-lib/AmqpContext.php b/pkg/amqp-lib/AmqpContext.php
@@ -5,6 +5,7 @@
 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;
@@ -45,6 +46,13 @@ class AmqpContext implements InteropAmqpContext, DelayStrategyAware
      */
     private $buffer;
 
+    /**
+     * an item contains an array: [AmqpConsumerInterop $consumer, callable $callback];.
+     *
+     * @var array
+     */
+    private $subscribers;
+
     /**
      * @param AbstractConnection $connection
      * @param array              $config
@@ -302,6 +310,38 @@ public function setQos($prefetchSize, $prefetchCount, $global)
         $this->getChannel()->basic_qos($prefetchSize, $prefetchCount, $global);
     }
 
+    /**
+     * {@inheritdoc}
+     */
+    public function subscribe(InteropAmqpConsumer $consumer, callable $callback)
+    {
+        if ($consumer->getConsumerTag() && array_key_exists($consumer->getConsumerTag(), $this->subscribers)) {
+            return;
+        }
+
+        throw new \LogicException('Not implemented');
+    }
+
+    /**
+     * {@inheritdoc}
+     */
+    public function unsubscribe(InteropAmqpConsumer $consumer)
+    {
+        throw new \LogicException('Not implemented');
+    }
+
+    /**
+     * {@inheritdoc}
+     */
+    public function consume($timeout = 0)
+    {
+        if (empty($this->subscribers)) {
+            throw new \LogicException('There is no subscribers. Consider calling basicConsumeSubscribe before consuming');
+        }
+
+        throw new \LogicException('Not implemented');
+    }
+
     /**
      * @return AMQPChannel
      */
diff --git a/pkg/enqueue/Consumption/QueueConsumer.php b/pkg/enqueue/Consumption/QueueConsumer.php
