[amqp] Add specs for AMQP's basic consume method.

src/Amqp/BasicConsumeBreakOnFalseSpec.php

@@ -0,0 +1,80 @@
+<?php
+
+namespace Interop\Queue\Spec\Amqp;
+
+use Interop\Amqp\AmqpConsumer;
+use Interop\Amqp\AmqpContext;
+use Interop\Amqp\AmqpMessage;
+use Interop\Amqp\AmqpQueue;
+use PHPUnit\Framework\TestCase;
+
+/**
+ * @group functional
+ */
+abstract class BasicConsumeBreakOnFalseSpec extends TestCase
+{
+    /**
+     * @var AmqpContext
+     */
+    private $context;
+
+    public function tearDown()
+    {
+        if ($this->context) {
+            $this->context->close();
+        }
+
+        parent::tearDown();
+    }
+
+    public function test()
+    {
+        $this->context = $context = $this->createContext();
+        $fooQueue = $this->createQueue($context, 'foo_basic_consume_break_on_false_spec');
+        $barQueue = $this->createQueue($context, 'bar_basic_consume_break_on_false_spec');
+
+        $expectedFooBody = __CLASS__.'foo'.time();
+        $expectedBarBody = __CLASS__.'bar'.time();
+
+        $context->createProducer()->send($fooQueue, $context->createMessage($expectedFooBody));
+        $context->createProducer()->send($barQueue, $context->createMessage($expectedBarBody));
+
+        $consumedMessages = 0;
+        $callback = function(AmqpMessage $message, AmqpConsumer $consumer) use (&$consumedMessages) {
+            $consumedMessages++;
+
+            $consumer->acknowledge($message);
+
+            return false;
+        };
+
+        $fooConsumer = $context->createConsumer($fooQueue);
+        $barConsumer = $context->createConsumer($barQueue);
+
+        $context->basicConsumeSubscribe($fooConsumer, $callback);
+        $context->basicConsumeSubscribe($barConsumer, $callback);
+        $context->basicConsume(1000);
+
+        $this->assertEquals(1, $consumedMessages);
+    }
+
+    /**
+     * @return AmqpContext
+     */
+    abstract protected function createContext();
+
+    /**
+     * @param AmqpContext $context
+     * @param string      $queueName
+     *
+     * @return AmqpQueue
+     */
+    protected function createQueue(AmqpContext $context, $queueName)
+    {
+        $queue = $context->createQueue($queueName);
+        $context->declareQueue($queue);
+        $context->purgeQueue($queue);
+
+        return $queue;
+    }
+}

src/Amqp/BasicConsumeFromAllSubscribedQueuesSpec.php

@@ -0,0 +1,89 @@
+<?php
+
+namespace Interop\Queue\Spec\Amqp;
+
+use Interop\Amqp\AmqpConsumer;
+use Interop\Amqp\AmqpContext;
+use Interop\Amqp\AmqpMessage;
+use Interop\Amqp\AmqpQueue;
+use PHPUnit\Framework\TestCase;
+
+/**
+ * @group functional
+ */
+abstract class BasicConsumeFromAllSubscribedQueuesSpec extends TestCase
+{
+    /**
+     * @var AmqpContext
+     */
+    private $context;
+
+    public function tearDown()
+    {
+        if ($this->context) {
+            $this->context->close();
+        }
+
+        parent::tearDown();
+    }
+
+    public function test()
+    {
+        $this->context = $context = $this->createContext();
+        $fooQueue = $this->createQueue($context, 'foo_basic_consume_from_all_subscribed_queues_spec');
+        $barQueue = $this->createQueue($context, 'bar_basic_consume_from_all_subscribed_queues_spec');
+
+        $expectedFooBody = 'fooBody';
+        $expectedBarBody = 'barBody';
+
+        $context->createProducer()->send($fooQueue, $context->createMessage($expectedFooBody));
+        $context->createProducer()->send($barQueue, $context->createMessage($expectedBarBody));
+
+        $fooConsumer = $context->createConsumer($fooQueue);
+        $barConsumer = $context->createConsumer($barQueue);
+
+        $actualBodies = [];
+        $actualQueues = [];
+        $callback = function(AmqpMessage $message, AmqpConsumer $consumer) use (&$actualBodies, &$actualQueues) {
+            $actualBodies[] = $message->getBody();
+            $actualQueues[] = $consumer->getQueue()->getQueueName();
+
+            $consumer->acknowledge($message);
+
+            return true;
+        };
+
+        $context->basicConsumeSubscribe($fooConsumer, $callback);
+        $context->basicConsumeSubscribe($barConsumer, $callback);
+        $context->basicConsume(1000);
+
+        $this->assertEquals([$expectedFooBody, $expectedBarBody], $actualBodies);
+        $this->assertEquals(
+            [
+                'foo_basic_consume_from_all_subscribed_queues_spec',
+                'bar_basic_consume_from_all_subscribed_queues_spec'
+            ],
+            $actualQueues
+        );
+    }
+
+    /**
+     * @return AmqpContext
+     */
+    abstract protected function createContext();
+
+    /**
+     * @param AmqpContext $context
+     * @param string      $queueName
+     *
+     * @return AmqpQueue
+     */
+    protected function createQueue(AmqpContext $context, $queueName)
+    {
+        $queue = $context->createQueue($queueName);
+        $context->declareQueue($queue);
+        $context->purgeQueue($queue);
+
+        return $queue;
+    }
+}

src/Amqp/BasicConsumeShouldAddConsumerTagOnSubscribeSpec.php

@@ -0,0 +1,61 @@
+<?php
+
+namespace Interop\Queue\Spec\Amqp;
+
+use Interop\Amqp\AmqpConsumer;
+use Interop\Amqp\AmqpContext;
+use Interop\Amqp\AmqpMessage;
+use Interop\Amqp\AmqpQueue;
+use PHPUnit\Framework\TestCase;
+
+/**
+ * @group functional
+ */
+abstract class BasicConsumeShouldAddConsumerTagOnSubscribeSpec extends TestCase
+{
+    /**
+     * @var AmqpContext
+     */
+    private $context;
+
+    public function tearDown()
+    {
+        if ($this->context) {
+            $this->context->close();
+        }
+
+        parent::tearDown();
+    }
+
+    public function test()
+    {
+        $this->context = $context = $this->createContext();
+        $queue = $this->createQueue($context, 'basic_consume_should_add_consumer_tag_on_subscribe_spec');
+
+        $consumer = $context->createConsumer($queue);
+
+        $context->basicConsumeSubscribe($consumer, function() {});
+
+        $this->assertNotEmpty($consumer->getConsumerTag());
+    }
+
+    /**
+     * @return AmqpContext
+     */
+    abstract protected function createContext();
+
+    /**
+     * @param AmqpContext $context
+     * @param string      $queueName
+     *
+     * @return AmqpQueue
+     */
+    protected function createQueue(AmqpContext $context, $queueName)
+    {
+        $queue = $context->createQueue($queueName);
+        $context->declareQueue($queue);
+        $context->purgeQueue($queue);
+
+        return $queue;
+    }
+}

src/Amqp/BasicConsumeShouldRemoveConsumerTagOnUnsubscribeSpec.php

@@ -0,0 +1,67 @@
+<?php
+
+namespace Interop\Queue\Spec\Amqp;
+
+use Interop\Amqp\AmqpConsumer;
+use Interop\Amqp\AmqpContext;
+use Interop\Amqp\AmqpMessage;
+use Interop\Amqp\AmqpQueue;
+use PHPUnit\Framework\TestCase;
+
+/**
+ * @group functional
+ */
+abstract class BasicConsumeShouldRemoveConsumerTagOnUnsubscribeSpec extends TestCase
+{
+    /**
+     * @var AmqpContext
+     */
+    private $context;
+
+    public function tearDown()
+    {
+        if ($this->context) {
+            $this->context->close();
+        }
+
+        parent::tearDown();
+    }
+
+    public function test()
+    {
+        $this->context = $context = $this->createContext();
+        $queue = $this->createQueue($context, 'basic_consume_should_remove_consumer_tag_on_unsubscribe_spec');
+
+        $consumer = $context->createConsumer($queue);
+
+        $context->basicConsumeSubscribe($consumer, function() {});
+        $context->basicConsume(100);
+
+        // guard
+        $this->assertNotEmpty($consumer->getConsumerTag());
+
+        $context->basicConsumeUnsubscribe($consumer);
+
+        $this->assertEmpty($consumer->getConsumerTag());
+    }
+
+    /**
+     * @return AmqpContext
+     */
+    abstract protected function createContext();
+
+    /**
+     * @param AmqpContext $context
+     * @param string      $queueName
+     *
+     * @return AmqpQueue
+     */
+    protected function createQueue(AmqpContext $context, $queueName)
+    {
+        $queue = $context->createQueue($queueName);
+        $context->declareQueue($queue);
+        $context->purgeQueue($queue);
+
+        return $queue;
+    }
+}

src/Amqp/BasicConsumeUntilUnsubscribedSpec.php

@@ -0,0 +1,86 @@
+<?php
+
+namespace Interop\Queue\Spec\Amqp;
+
+use Interop\Amqp\AmqpConsumer;
+use Interop\Amqp\AmqpContext;
+use Interop\Amqp\AmqpMessage;
+use Interop\Amqp\AmqpQueue;
+use PHPUnit\Framework\TestCase;
+
+/**
+ * @group functional
+ */
+abstract class BasicConsumeUntilUnsubscribedSpec extends TestCase
+{
+    /**
+     * @var AmqpContext
+     */
+    private $context;
+
+    public function tearDown()
+    {
+        if ($this->context) {
+            $this->context->close();
+        }
+
+        parent::tearDown();
+    }
+
+    public function test()
+    {
+        $this->context = $context = $this->createContext();
+        $fooQueue = $this->createQueue($context, 'foo_basic_consume_until_unsubscribed_spec');
+        $barQueue = $this->createQueue($context, 'bar_basic_consume_until_unsubscribed_spec');
+
+        $context->createProducer()->send($fooQueue, $context->createMessage());
+        $context->createProducer()->send($barQueue, $context->createMessage());
+
+        $fooConsumer = $context->createConsumer($fooQueue);
+        $barConsumer = $context->createConsumer($barQueue);
+
+        $consumedMessages = 0;
+        $callback = function(AmqpMessage $message, AmqpConsumer $consumer) use (&$consumedMessages) {
+            $consumedMessages++;
+
+            $consumer->acknowledge($message);
+
+            return true;
+        };
+
+        $context->basicConsumeSubscribe($fooConsumer, $callback);
+        $context->basicConsumeSubscribe($barConsumer, $callback);
+        $context->basicConsume(1000);
+
+        $this->assertEquals(2, $consumedMessages);
+
+        $context->createProducer()->send($fooQueue, $context->createMessage());
+        $context->createProducer()->send($barQueue, $context->createMessage());
+
+        $consumedMessages = 0;
+        $context->basicConsumeUnsubscribe($fooConsumer);
+        $context->basicConsume(1000);
+
+        $this->assertEquals(1, $consumedMessages);
+    }
+
+    /**
+     * @return AmqpContext
+     */
+    abstract protected function createContext();
+
+    /**
+     * @param AmqpContext $context
+     * @param string      $queueName
+     *
+     * @return AmqpQueue
+     */
+    protected function createQueue(AmqpContext $context, $queueName)
+    {
+        $queue = $context->createQueue($queueName);
+        $context->declareQueue($queue);
+        $context->purgeQueue($queue);
+
+        return $queue;
+    }
+}

src/SendAndReceiveDelayedMessageFromQueueSpec.php

@@ -36,7 +36,7 @@ public function test()
         $consumer->acknowledge($message);
         $this->assertSame($expectedBody, $message->getBody());
 
-        $this->assertGreaterThanOrEqual(5, microtime(true) - $sendAt);
+        $this->assertGreaterThanOrEqual(4, microtime(true) - $sendAt);
     }
 
     /**