[amqp-ext] fixes and impr for basic consume feature.

Author: makasim

pkg/amqp-ext/AmqpConsumer.php

@@ -6,7 +6,6 @@
 use Interop\Amqp\AmqpMessage as InteropAmqpMessage;
 use Interop\Amqp\AmqpQueue;
 use Interop\Amqp\Impl\AmqpMessage;
-use Interop\Queue\Exception;
 use Interop\Queue\InvalidMessageException;
 use Interop\Queue\PsrMessage;
 
@@ -32,11 +31,6 @@ class AmqpConsumer implements InteropAmqpConsumer
      */
     private $extQueue;
 
-    /**
-     * @var bool
-     */
-    private $isInit;
-
     /**
      * @var string
      */
@@ -65,19 +59,13 @@ public function __construct(AmqpContext $context, AmqpQueue $queue, Buffer $buff
         $this->buffer = $buffer;
         $this->receiveMethod = $receiveMethod;
         $this->flags = self::FLAG_NOPARAM;
-
-        $this->isInit = false;
     }
 
     /**
      * {@inheritdoc}
      */
     public function setConsumerTag($consumerTag)
     {
-        if ($this->isInit) {
-            throw new Exception('Consumer tag is not mutable after it has been subscribed to broker');
-        }
-
         $this->consumerTag = $consumerTag;
     }
 
@@ -157,7 +145,7 @@ public function receive($timeout = 0)
     public function receiveNoWait()
     {
         if ($extMessage = $this->getExtQueue()->get(Flags::convertConsumerFlags($this->flags))) {
-            return $this->convertMessage($extMessage);
+            return $this->context->convertMessage($extMessage);
         }
     }
 
@@ -213,85 +201,44 @@ private function receiveBasicGet($timeout)
      */
     private function receiveBasicConsume($timeout)
     {
-        if ($this->isInit && $message = $this->buffer->pop($this->getExtQueue()->getConsumerTag())) {
-            return $message;
+        if (false == $this->consumerTag) {
+            $this->context->subscribe($this, function (InteropAmqpMessage $message) {
+                $this->buffer->push($message->getConsumerTag(), $message);
+
+                return false;
+            });
         }
 
-        /** @var \AMQPQueue $extQueue */
-        $extConnection = $this->getExtQueue()->getChannel()->getConnection();
+        if ($message = $this->buffer->pop($this->consumerTag)) {
+            return $message;
+        }
 
-        $originalTimeout = $extConnection->getReadTimeout();
-        try {
-            $extConnection->setReadTimeout($timeout / 1000);
+        while (true) {
+            $start = microtime(true);
 
-            if (false == $this->isInit) {
-                $this->getExtQueue()->consume(null, Flags::convertConsumerFlags($this->flags), $this->consumerTag);
+            $this->context->consume($timeout);
 
-                $this->isInit = true;
+            if ($message = $this->buffer->pop($this->consumerTag)) {
+                return $message;
             }
 
-            /** @var AmqpMessage|null $message */
-            $message = null;
-
-            $this->getExtQueue()->consume(function (\AMQPEnvelope $extEnvelope, \AMQPQueue $q) use (&$message) {
-                $message = $this->convertMessage($extEnvelope);
-                $message->setConsumerTag($q->getConsumerTag());
-
-                if ($this->getExtQueue()->getConsumerTag() == $q->getConsumerTag()) {
-                    return false;
-                }
+            // is here when consumed message is not for this consumer
 
-                // not our message, put it to buffer and continue.
-                $this->buffer->push($q->getConsumerTag(), $message);
-
-                $message = null;
+            // as timeout is infinite have to continue consumption, but it can overflow message buffer
+            if ($timeout <= 0) {
+                continue;
+            }
 
-                return true;
-            }, AMQP_JUST_CONSUME);
+            // compute remaining timeout and continue until time is up
+            $stop = microtime(true);
+            $timeout -= ($stop - $start) * 1000;
 
-            return $message;
-        } catch (\AMQPQueueException $e) {
-            if ('Consumer timeout exceed' == $e->getMessage()) {
-                return null;
+            if ($timeout <= 0) {
+                break;
             }
-
-            throw $e;
-        } finally {
-            $extConnection->setReadTimeout($originalTimeout);
         }
     }
 
-    /**
-     * @param \AMQPEnvelope $extEnvelope
-     *
-     * @return AmqpMessage
-     */
-    private function convertMessage(\AMQPEnvelope $extEnvelope)
-    {
-        $message = new AmqpMessage(
-            $extEnvelope->getBody(),
-            $extEnvelope->getHeaders(),
-            [
-                'message_id' => $extEnvelope->getMessageId(),
-                'correlation_id' => $extEnvelope->getCorrelationId(),
-                'app_id' => $extEnvelope->getAppId(),
-                'type' => $extEnvelope->getType(),
-                'content_encoding' => $extEnvelope->getContentEncoding(),
-                'content_type' => $extEnvelope->getContentType(),
-                'expiration' => $extEnvelope->getExpiration(),
-                'priority' => $extEnvelope->getPriority(),
-                'reply_to' => $extEnvelope->getReplyTo(),
-                'timestamp' => $extEnvelope->getTimeStamp(),
-                'user_id' => $extEnvelope->getUserId(),
-            ]
-        );
-        $message->setRedelivered($extEnvelope->isRedelivery());
-        $message->setDeliveryTag($extEnvelope->getDeliveryTag());
-        $message->setRoutingKey($extEnvelope->getRoutingKey());
-
-        return $message;
-    }
-
     /**
      * @return \AMQPQueue
      */

pkg/amqp-ext/AmqpContext.php

@@ -384,11 +384,13 @@ public function consume($timeout = 0)
     }
 
     /**
+     * @internal It must be used here and in the consumer only
+     *
      * @param \AMQPEnvelope $extEnvelope
      *
      * @return AmqpMessage
      */
-    private function convertMessage(\AMQPEnvelope $extEnvelope)
+    public function convertMessage(\AMQPEnvelope $extEnvelope)
     {
         $message = new AmqpMessage(
             $extEnvelope->getBody(),

pkg/amqp-ext/Tests/Spec/AmqpSendAndReceiveDelayedMessageWithDelayPluginStrategyTest.php

@@ -3,6 +3,7 @@
 namespace Enqueue\AmqpExt\Tests\Spec;
 
 use Enqueue\AmqpExt\AmqpConnectionFactory;
+use Enqueue\AmqpExt\AmqpContext;
 use Enqueue\AmqpTools\RabbitMqDelayPluginDelayStrategy;
 use Interop\Queue\PsrContext;
 use Interop\Queue\Spec\SendAndReceiveDelayedMessageFromQueueSpec;
@@ -24,13 +25,16 @@ protected function createContext()
     }
 
     /**
+     * @param AmqpContext $context
+     *
      * {@inheritdoc}
      */
     protected function createQueue(PsrContext $context, $queueName)
     {
         $queue = parent::createQueue($context, $queueName);
 
         $context->declareQueue($queue);
+        $context->purgeQueue($queue);
 
         return $queue;
     }

pkg/amqp-ext/Tests/Spec/AmqpSendAndReceiveDelayedMessageWithDlxStrategyTest.php

@@ -3,6 +3,7 @@
 namespace Enqueue\AmqpExt\Tests\Spec;
 
 use Enqueue\AmqpExt\AmqpConnectionFactory;
+use Enqueue\AmqpExt\AmqpContext;
 use Enqueue\AmqpTools\RabbitMqDlxDelayStrategy;
 use Interop\Queue\PsrContext;
 use Interop\Queue\Spec\SendAndReceiveDelayedMessageFromQueueSpec;
@@ -24,13 +25,16 @@ protected function createContext()
     }
 
     /**
+     * @param AmqpContext $context
+     *
      * {@inheritdoc}
      */
     protected function createQueue(PsrContext $context, $queueName)
     {
         $queue = parent::createQueue($context, $queueName);
 
         $context->declareQueue($queue);
+        $context->purgeQueue($queue);
 
         return $queue;
     }