[amqp-lib] Add basic consume support.

Author: makasim

pkg/amqp-lib/AmqpContext.php

@@ -20,6 +20,8 @@
 use Interop\Queue\PsrTopic;
 use PhpAmqpLib\Channel\AMQPChannel;
 use PhpAmqpLib\Connection\AbstractConnection;
+use PhpAmqpLib\Exception\AMQPTimeoutException;
+use PhpAmqpLib\Message\AMQPMessage as LibAMQPMessage;
 use PhpAmqpLib\Wire\AMQPTable;
 
 class AmqpContext implements InteropAmqpContext, DelayStrategyAware
@@ -319,15 +321,55 @@ public function subscribe(InteropAmqpConsumer $consumer, callable $callback)
             return;
         }
 
-        throw new \LogicException('Not implemented');
+        $libCallback = function (LibAMQPMessage $message) {
+            $receivedMessage = $this->convertMessage($message);
+            $receivedMessage->setConsumerTag($message->delivery_info['consumer_tag']);
+
+            /**
+             * @var AmqpConsumer
+             * @var callable     $callback
+             */
+            list($consumer, $callback) = $this->subscribers[$message->delivery_info['consumer_tag']];
+
+            if (false === call_user_func($callback, $receivedMessage, $consumer)) {
+                throw new StopBasicConsumptionException();
+            }
+        };
+
+        $consumerTag = $this->getChannel()->basic_consume(
+            $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),
+            $libCallback
+        );
+
+        if (empty($consumerTag)) {
+            throw new Exception('Got empty consumer tag');
+        }
+
+        $consumer->setConsumerTag($consumerTag);
+
+        $this->subscribers[$consumerTag] = [$consumer, $callback];
     }
 
     /**
      * {@inheritdoc}
      */
     public function unsubscribe(InteropAmqpConsumer $consumer)
     {
-        throw new \LogicException('Not implemented');
+        if (false == $consumer->getConsumerTag()) {
+            return;
+        }
+
+        $consumerTag = $consumer->getConsumerTag();
+
+        $this->getChannel()->basic_cancel($consumerTag);
+
+        $consumer->setConsumerTag(null);
+        unset($this->subscribers[$consumerTag], $this->getChannel()->callbacks[$consumerTag]);
     }
 
     /**
@@ -339,7 +381,27 @@ public function consume($timeout = 0)
             throw new \LogicException('There is no subscribers. Consider calling basicConsumeSubscribe before consuming');
         }
 
-        throw new \LogicException('Not implemented');
+        try {
+            while (true) {
+                $start = microtime(true);
+
+                $this->channel->wait(null, false, $timeout / 1000);
+
+                if ($timeout <= 0) {
+                    continue;
+                }
+
+                // compute remaining timeout and continue until time is up
+                $stop = microtime(true);
+                $timeout -= ($stop - $start) * 1000;
+
+                if ($timeout <= 0) {
+                    break;
+                }
+            }
+        } catch (AMQPTimeoutException $e) {
+        } catch (StopBasicConsumptionException $e) {
+        }
     }
 
     /**
@@ -358,4 +420,28 @@ private function getChannel()
 
         return $this->channel;
     }
+
+    /**
+     * @param LibAMQPMessage $amqpMessage
+     *
+     * @return InteropAmqpMessage
+     */
+    private function convertMessage(LibAMQPMessage $amqpMessage)
+    {
+        $headers = new AMQPTable($amqpMessage->get_properties());
+        $headers = $headers->getNativeData();
+
+        $properties = [];
+        if (isset($headers['application_headers'])) {
+            $properties = $headers['application_headers'];
+        }
+        unset($headers['application_headers']);
+
+        $message = new AmqpMessage($amqpMessage->getBody(), $properties, $headers);
+        $message->setDeliveryTag($amqpMessage->delivery_info['delivery_tag']);
+        $message->setRedelivered($amqpMessage->delivery_info['redelivered']);
+        $message->setRoutingKey($amqpMessage->delivery_info['routing_key']);
+
+        return $message;
+    }
 }

pkg/amqp-lib/StopBasicConsumptionException.php

@@ -0,0 +1,7 @@
+<?php
+
+namespace Enqueue\AmqpLib;
+
+class StopBasicConsumptionException extends \LogicException
+{
+}

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

@@ -0,0 +1,22 @@
+<?php
+
+namespace Enqueue\AmqpLib\Tests\Spec;
+
+use Enqueue\AmqpLib\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-lib/Tests/Spec/AmqpBasicConsumeFromAllSubscribedQueuesTest.php

@@ -0,0 +1,22 @@
+<?php
+
+namespace Enqueue\AmqpLib\Tests\Spec;
+
+use Enqueue\AmqpLib\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-lib/Tests/Spec/AmqpBasicConsumeShouldAddConsumerTagOnSubscribeTest.php

@@ -0,0 +1,22 @@
+<?php
+
+namespace Enqueue\AmqpLib\Tests\Spec;
+
+use Enqueue\AmqpLib\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-lib/Tests/Spec/AmqpBasicConsumeShouldRemoveConsumerTagOnUnsubscribeTest.php

@@ -0,0 +1,22 @@
+<?php
+
+namespace Enqueue\AmqpLib\Tests\Spec;
+
+use Enqueue\AmqpLib\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-lib/Tests/Spec/AmqpBasicConsumeUntilUnsubscribedTest.php

@@ -0,0 +1,22 @@
+<?php
+
+namespace Enqueue\AmqpLib\Tests\Spec;
+
+use Enqueue\AmqpLib\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();
+    }
+}

pkg/enqueue/Consumption/QueueConsumer.php

@@ -221,6 +221,14 @@ public function consume(ExtensionInterface $runtimeExtension = null)
             } catch (ConsumptionInterruptedException $e) {
                 $logger->info(sprintf('Consuming interrupted'));
 
+                if ($this->psrContext instanceof AmqpContext) {
+                    foreach ($consumers as $consumer) {
+                        /* @var AmqpConsumer $consumer */
+
+                        $this->psrContext->unsubscribe($consumer);
+                    }
+                }
+
                 $context->setExecutionInterrupted(true);
 
                 $extension->onInterrupted($context);