Merge pull request php-enqueue#7 from php-enqueue/amqp-fix-receive-wait-time

makasim · web-flow · commit 937f9cd4f247 · 2017-01-09T18:40:21.000+02:00
[bug][amqp-ext] Receive timeout parameter is miliseconds
diff --git a/pkg/amqp-ext/AmqpConsumer.php b/pkg/amqp-ext/AmqpConsumer.php
@@ -68,7 +68,7 @@ public function receive($timeout = 0)
 
         $originalTimeout = $extConnection->getReadTimeout();
         try {
-            $extConnection->setReadTimeout($timeout);
+            $extConnection->setReadTimeout($timeout / 1000);
 
             if (false == $this->isInit) {
                 $this->getExtQueue()->consume(null, AMQP_NOPARAM, $this->consumerId);
diff --git a/pkg/amqp-ext/Tests/Functional/AmqpCommonUseCasesTest.php b/pkg/amqp-ext/Tests/Functional/AmqpCommonUseCasesTest.php
@@ -41,7 +41,7 @@ public function testWaitsForTwoSecondsAndReturnNullOnReceive()
         $startAt = microtime(true);
 
         $consumer = $this->amqpContext->createConsumer($queue);
-        $message = $consumer->receive(2);
+        $message = $consumer->receive(2000);
 
         $endAt = microtime(true);
 
@@ -83,7 +83,7 @@ public function testProduceAndReceiveOneMessageSentDirectlyToQueue()
         $producer->send($queue, $message);
 
         $consumer = $this->amqpContext->createConsumer($queue);
-        $message = $consumer->receive(1);
+        $message = $consumer->receive(1000);
 
         $this->assertInstanceOf(AmqpMessage::class, $message);
         $consumer->acknowledge($message);
@@ -115,7 +115,7 @@ public function testProduceAndReceiveOneMessageSentDirectlyToTemporaryQueue()
         $producer->send($queue, $message);
 
         $consumer = $this->amqpContext->createConsumer($queue);
-        $message = $consumer->receive(1);
+        $message = $consumer->receive(1000);
 
         $this->assertInstanceOf(AmqpMessage::class, $message);
         $consumer->acknowledge($message);
@@ -140,7 +140,7 @@ public function testProduceAndReceiveOneMessageSentDirectlyToTopic()
         $producer->send($topic, $message);
 
         $consumer = $this->amqpContext->createConsumer($queue);
-        $message = $consumer->receive(1);
+        $message = $consumer->receive(1000);
 
         $this->assertInstanceOf(AmqpMessage::class, $message);
         $consumer->acknowledge($message);
@@ -157,13 +157,37 @@ public function testConsumerReceiveMessageFromTopicDirectly()
 
         $consumer = $this->amqpContext->createConsumer($topic);
         //guard
-        $this->assertNull($consumer->receive(1));
+        $this->assertNull($consumer->receive(1000));
 
         $message = $this->amqpContext->createMessage(__METHOD__);
 
         $producer = $this->amqpContext->createProducer();
         $producer->send($topic, $message);
-        $actualMessage = $consumer->receive(1);
+        $actualMessage = $consumer->receive(1000);
+
+        $this->assertInstanceOf(AmqpMessage::class, $actualMessage);
+        $consumer->acknowledge($message);
+
+        $this->assertEquals(__METHOD__, $message->getBody());
+    }
+
+    public function testConsumerReceiveMessageWithZeroTimeout()
+    {
+        $topic = $this->amqpContext->createTopic('amqp_ext.test_exchange');
+        $topic->setType(AMQP_EX_TYPE_FANOUT);
+
+        $this->amqpContext->declareTopic($topic);
+
+        $consumer = $this->amqpContext->createConsumer($topic);
+        //guard
+        $this->assertNull($consumer->receive(1000));
+
+        $message = $this->amqpContext->createMessage(__METHOD__);
+
+        $producer = $this->amqpContext->createProducer();
+        $producer->send($topic, $message);
+        usleep(100);
+        $actualMessage = $consumer->receive(0);
 
         $this->assertInstanceOf(AmqpMessage::class, $actualMessage);
         $consumer->acknowledge($message);
