[amqp] add spec for message timestamp, and pre fetch count.

Author: makasim

src/Amqp/BasicConsumeBreakOnFalseSpec.php

@@ -30,6 +30,8 @@ public function tearDown()
     public function test()
     {
         $this->context = $context = $this->createContext();
+        $context->setQos(0, 5, false);
+
         $fooQueue = $this->createQueue($context, 'foo_basic_consume_break_on_false_spec');
         $barQueue = $this->createQueue($context, 'bar_basic_consume_break_on_false_spec');
 

src/Amqp/BasicConsumeFromAllSubscribedQueuesSpec.php

@@ -30,6 +30,8 @@ public function tearDown()
     public function test()
     {
         $this->context = $context = $this->createContext();
+        $context->setQos(0, 5, false);
+
         $fooQueue = $this->createQueue($context, 'foo_basic_consume_from_all_subscribed_queues_spec');
         $barQueue = $this->createQueue($context, 'bar_basic_consume_from_all_subscribed_queues_spec');
 

src/Amqp/BasicConsumeShouldAddConsumerTagOnSubscribeSpec.php

@@ -30,6 +30,8 @@ public function tearDown()
     public function test()
     {
         $this->context = $context = $this->createContext();
+        $context->setQos(0, 5, false);
+
         $queue = $this->createQueue($context, 'basic_consume_should_add_consumer_tag_on_subscribe_spec');
 
         $consumer = $context->createConsumer($queue);

src/Amqp/BasicConsumeShouldRemoveConsumerTagOnUnsubscribeSpec.php

@@ -30,6 +30,8 @@ public function tearDown()
     public function test()
     {
         $this->context = $context = $this->createContext();
+        $context->setQos(0, 5, false);
+
         $queue = $this->createQueue($context, 'basic_consume_should_remove_consumer_tag_on_unsubscribe_spec');
 
         $consumer = $context->createConsumer($queue);

src/Amqp/BasicConsumeUntilUnsubscribedSpec.php

@@ -30,6 +30,8 @@ public function tearDown()
     public function test()
     {
         $this->context = $context = $this->createContext();
+        $context->setQos(0, 5, false);
+
         $fooQueue = $this->createQueue($context, 'foo_basic_consume_until_unsubscribed_spec');
         $barQueue = $this->createQueue($context, 'bar_basic_consume_until_unsubscribed_spec');
 

src/Amqp/PreFetchCountSpec.php

@@ -0,0 +1,73 @@
+<?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 PreFetchCountSpec 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, 'pre_fetch_count_spec');
+
+        $context->createProducer()->send($queue, $context->createMessage());
+        $context->createProducer()->send($queue, $context->createMessage());
+        $context->createProducer()->send($queue, $context->createMessage());
+        $context->createProducer()->send($queue, $context->createMessage());
+        $context->createProducer()->send($queue, $context->createMessage());
+
+        $this->context->setQos(0, 3, false);
+
+        $consumer = $context->createConsumer($queue);
+
+        $consumedMessages = 0;
+        $context->subscribe($consumer, function() use (&$consumedMessages) {
+            $consumedMessages++;
+        });
+        $context->consume(100);
+
+        $this->assertEquals(3, $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/SendAndReceiveTimestampAsIntegerSpec.php

@@ -0,0 +1,71 @@
+<?php
+
+namespace Interop\Queue\Spec\Amqp;
+
+use Interop\Amqp\AmqpContext;
+use Interop\Amqp\AmqpMessage;
+use Interop\Amqp\AmqpQueue;
+use PHPUnit\Framework\TestCase;
+
+/**
+ * @group functional
+ */
+abstract class SendAndReceiveTimestampAsIntegerSpec 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, 'send_and_receive_timestamp_as_integer_spec');
+
+        $expectedTime = time();
+        $expectedBody = __METHOD__.time();
+
+        $message = $context->createMessage($expectedBody);
+        $message->setTimestamp($expectedTime);
+
+        $context->createProducer()->send($queue, $message);
+
+        $consumer = $context->createConsumer($queue);
+
+        $receivedMessage = $consumer->receive(100);
+
+        $this->assertInstanceOf(AmqpMessage::class, $receivedMessage);
+        $this->assertSame($expectedBody, $receivedMessage->getBody());
+        $this->assertSame($expectedTime, $receivedMessage->getTimestamp());
+    }
+
+    /**
+     * @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;
+    }
+}