[amqp] Delay Strategy

composer.json

@@ -9,6 +9,7 @@
     "enqueue/amqp-ext": "*@dev",
     "enqueue/amqp-lib": "*@dev",
     "enqueue/amqp-bunny": "*@dev",
+    "enqueue/amqp-tools": "*@dev",
     "php-amqplib/php-amqplib": "^2.7@dev",
     "enqueue/redis": "*@dev",
     "enqueue/fs": "*@dev",
@@ -84,6 +85,10 @@
       "type": "path",
       "url": "pkg/amqp-bunny"
     },
+    {
+      "type": "path",
+      "url": "pkg/amqp-tools"
+    },
     {
       "type": "path",
       "url": "pkg/redis"

pkg/amqp-bunny/AmqpConnectionFactory.php

@@ -3,10 +3,14 @@
 namespace Enqueue\AmqpBunny;
 
 use Bunny\Client;
+use Enqueue\AmqpTools\DelayStrategyAware;
+use Enqueue\AmqpTools\DelayStrategyAwareTrait;
 use Interop\Amqp\AmqpConnectionFactory as InteropAmqpConnectionFactory;
 
-class AmqpConnectionFactory implements InteropAmqpConnectionFactory
+class AmqpConnectionFactory implements InteropAmqpConnectionFactory, DelayStrategyAware
 {
+    use DelayStrategyAwareTrait;
+
     /**
      * @var array
      */
@@ -72,12 +76,18 @@ public function __construct($config = 'amqp://')
     public function createContext()
     {
         if ($this->config['lazy']) {
-            return new AmqpContext(function () {
+            $context = new AmqpContext(function () {
                 return $this->establishConnection()->channel();
             }, $this->config);
+            $context->setDelayStrategy($this->delayStrategy);
+
+            return $context;
         }
 
-        return new AmqpContext($this->establishConnection()->channel(), $this->config);
+        $context = new AmqpContext($this->establishConnection()->channel(), $this->config);
+        $context->setDelayStrategy($this->delayStrategy);
+
+        return $context;
     }
 
     /**

pkg/amqp-bunny/AmqpContext.php

@@ -3,6 +3,8 @@
 namespace Enqueue\AmqpBunny;
 
 use Bunny\Channel;
+use Enqueue\AmqpTools\DelayStrategyAware;
+use Enqueue\AmqpTools\DelayStrategyAwareTrait;
 use Interop\Amqp\AmqpBind as InteropAmqpBind;
 use Interop\Amqp\AmqpContext as InteropAmqpContext;
 use Interop\Amqp\AmqpMessage as InteropAmqpMessage;
@@ -17,8 +19,10 @@
 use Interop\Queue\PsrDestination;
 use Interop\Queue\PsrTopic;
 
-class AmqpContext implements InteropAmqpContext
+class AmqpContext implements InteropAmqpContext, DelayStrategyAware
 {
+    use DelayStrategyAwareTrait;
+
     /**
      * @var Channel
      */
@@ -124,7 +128,10 @@ public function createConsumer(PsrDestination $destination)
      */
     public function createProducer()
     {
-        return new AmqpProducer($this->getBunnyChannel());
+        $producer = new AmqpProducer($this->getBunnyChannel(), $this);
+        $producer->setDelayStrategy($this->delayStrategy);
+
+        return $producer;
     }
 
     /**

pkg/amqp-bunny/AmqpProducer.php

@@ -3,6 +3,8 @@
 namespace Enqueue\AmqpBunny;
 
 use Bunny\Channel;
+use Enqueue\AmqpTools\DelayStrategyAware;
+use Enqueue\AmqpTools\DelayStrategyAwareTrait;
 use Interop\Amqp\AmqpMessage as InteropAmqpMessage;
 use Interop\Amqp\AmqpProducer as InteropAmqpProducer;
 use Interop\Amqp\AmqpQueue as InteropAmqpQueue;
@@ -14,8 +16,10 @@
 use Interop\Queue\PsrMessage;
 use Interop\Queue\PsrTopic;
 
-class AmqpProducer implements InteropAmqpProducer
+class AmqpProducer implements InteropAmqpProducer, DelayStrategyAware
 {
+    use DelayStrategyAwareTrait;
+
     /**
      * @var int|null
      */
@@ -32,11 +36,23 @@ class AmqpProducer implements InteropAmqpProducer
     private $channel;
 
     /**
-     * @param Channel $channel
+     * @var int
+     */
+    private $deliveryDelay;
+
+    /**
+     * @var AmqpContext
+     */
+    private $context;
+
+    /**
+     * @param Channel     $channel
+     * @param AmqpContext $context
      */
-    public function __construct(Channel $channel)
+    public function __construct(Channel $channel, AmqpContext $context)
     {
         $this->channel = $channel;
+        $this->context = $context;
     }
 
     /**
@@ -47,8 +63,7 @@ public function send(PsrDestination $destination, PsrMessage $message)
     {
         $destination instanceof PsrTopic
             ? InvalidDestinationException::assertDestinationInstanceOf($destination, InteropAmqpTopic::class)
-            : InvalidDestinationException::assertDestinationInstanceOf($destination, InteropAmqpQueue::class)
-        ;
+            : InvalidDestinationException::assertDestinationInstanceOf($destination, InteropAmqpQueue::class);
 
         InvalidMessageException::assertMessageInstanceOf($message, InteropAmqpMessage::class);
 
@@ -66,7 +81,9 @@ public function send(PsrDestination $destination, PsrMessage $message)
             $amqpProperties['application_headers'] = $appProperties;
         }
 
-        if ($destination instanceof InteropAmqpTopic) {
+        if ($this->deliveryDelay) {
+            $this->delayStrategy->delayMessage($this->context, $destination, $message, $this->deliveryDelay);
+        } elseif ($destination instanceof InteropAmqpTopic) {
             $this->channel->publish(
                 $message->getBody(),
                 $amqpProperties,
@@ -92,19 +109,19 @@ public function send(PsrDestination $destination, PsrMessage $message)
      */
     public function setDeliveryDelay($deliveryDelay)
     {
-        if (null === $deliveryDelay) {
-            return;
+        if (null === $this->delayStrategy) {
+            throw DeliveryDelayNotSupportedException::providerDoestNotSupportIt();
         }
 
-        throw DeliveryDelayNotSupportedException::providerDoestNotSupportIt();
+        $this->deliveryDelay = $deliveryDelay;
     }
 
     /**
      * {@inheritdoc}
      */
     public function getDeliveryDelay()
     {
-        return null;
+        return $this->deliveryDelay;
     }
 
     /**

pkg/amqp-bunny/Tests/AmqpProducerTest.php

@@ -4,12 +4,15 @@
 
 use Bunny\Channel;
 use Bunny\Message;
+use Enqueue\AmqpBunny\AmqpContext;
 use Enqueue\AmqpBunny\AmqpProducer;
+use Enqueue\AmqpTools\DelayStrategy;
 use Enqueue\Test\ClassExtensionTrait;
 use Interop\Amqp\AmqpMessage as InteropAmqpMessage;
 use Interop\Amqp\Impl\AmqpMessage;
 use Interop\Amqp\Impl\AmqpQueue;
 use Interop\Amqp\Impl\AmqpTopic;
+use Interop\Queue\DeliveryDelayNotSupportedException;
 use Interop\Queue\InvalidDestinationException;
 use Interop\Queue\InvalidMessageException;
 use Interop\Queue\PsrDestination;
@@ -23,7 +26,7 @@ class AmqpProducerTest extends TestCase
 
     public function testCouldBeConstructedWithRequiredArguments()
     {
-        new AmqpProducer($this->createBunnyChannelMock());
+        new AmqpProducer($this->createBunnyChannelMock(), $this->createContextMock());
     }
 
     public function testShouldImplementPsrProducerInterface()
@@ -33,7 +36,7 @@ public function testShouldImplementPsrProducerInterface()
 
     public function testShouldThrowExceptionWhenDestinationTypeIsInvalid()
     {
-        $producer = new AmqpProducer($this->createBunnyChannelMock());
+        $producer = new AmqpProducer($this->createBunnyChannelMock(), $this->createContextMock());
 
         $this->expectException(InvalidDestinationException::class);
         $this->expectExceptionMessage('The destination must be an instance of Interop\Amqp\AmqpQueue but got');
@@ -43,7 +46,7 @@ public function testShouldThrowExceptionWhenDestinationTypeIsInvalid()
 
     public function testShouldThrowExceptionWhenMessageTypeIsInvalid()
     {
-        $producer = new AmqpProducer($this->createBunnyChannelMock());
+        $producer = new AmqpProducer($this->createBunnyChannelMock(), $this->createContextMock());
 
         $this->expectException(InvalidMessageException::class);
         $this->expectExceptionMessage('The message must be an instance of Interop\Amqp\AmqpMessage but it is');
@@ -65,7 +68,7 @@ public function testShouldPublishMessageToTopic()
         $message = new AmqpMessage('body');
         $message->setRoutingKey('routing-key');
 
-        $producer = new AmqpProducer($channel);
+        $producer = new AmqpProducer($channel, $this->createContextMock());
         $producer->send($topic, $message);
     }
 
@@ -80,10 +83,52 @@ public function testShouldPublishMessageToQueue()
 
         $queue = new AmqpQueue('queue');
 
-        $producer = new AmqpProducer($channel);
+        $producer = new AmqpProducer($channel, $this->createContextMock());
         $producer->send($queue, new AmqpMessage('body'));
     }
 
+    public function testShouldDelayMessage()
+    {
+        $channel = $this->createBunnyChannelMock();
+        $channel
+            ->expects($this->never())
+            ->method('publish')
+        ;
+
+        $message = new AmqpMessage('body');
+        $context = $this->createContextMock();
+        $queue = new AmqpQueue('queue');
+
+        $delayStrategy = $this->createDelayStrategyMock();
+        $delayStrategy
+            ->expects($this->once())
+            ->method('delayMessage')
+            ->with($this->identicalTo($context), $this->identicalTo($queue), $this->identicalTo($message), 10000)
+        ;
+
+        $producer = new AmqpProducer($channel, $context);
+        $producer->setDelayStrategy($delayStrategy);
+        $producer->setDeliveryDelay(10000);
+
+        $producer->send($queue, $message);
+    }
+
+    public function testShouldThrowExceptionOnSetDeliveryDelayWhenDeliveryStrategyIsNotSet()
+    {
+        $channel = $this->createBunnyChannelMock();
+        $channel
+            ->expects($this->never())
+            ->method('publish')
+        ;
+
+        $producer = new AmqpProducer($channel, $this->createContextMock());
+
+        $this->expectException(DeliveryDelayNotSupportedException::class);
+        $this->expectExceptionMessage('The provider does not support delivery delay feature');
+
+        $producer->setDeliveryDelay(10000);
+    }
+
     public function testShouldSetMessageHeaders()
     {
         $channel = $this->createBunnyChannelMock();
@@ -93,7 +138,7 @@ public function testShouldSetMessageHeaders()
             ->with($this->anything(), ['content_type' => 'text/plain'])
         ;
 
-        $producer = new AmqpProducer($channel);
+        $producer = new AmqpProducer($channel, $this->createContextMock());
         $producer->send(new AmqpTopic('name'), new AmqpMessage('body', [], ['content_type' => 'text/plain']));
     }
 
@@ -106,7 +151,7 @@ public function testShouldSetMessageProperties()
             ->with($this->anything(), ['application_headers' => ['key' => 'value']])
         ;
 
-        $producer = new AmqpProducer($channel);
+        $producer = new AmqpProducer($channel, $this->createContextMock());
         $producer->send(new AmqpTopic('name'), new AmqpMessage('body', ['key' => 'value']));
     }
 
@@ -123,7 +168,7 @@ public function testShouldPropagateFlags()
         $message->addFlag(InteropAmqpMessage::FLAG_IMMEDIATE);
         $message->addFlag(InteropAmqpMessage::FLAG_MANDATORY);
 
-        $producer = new AmqpProducer($channel);
+        $producer = new AmqpProducer($channel, $this->createContextMock());
         $producer->send(new AmqpTopic('name'), $message);
     }
 
@@ -150,4 +195,20 @@ private function createBunnyChannelMock()
     {
         return $this->createMock(Channel::class);
     }
+
+    /**
+     * @return \PHPUnit_Framework_MockObject_MockObject|AmqpContext
+     */
+    private function createContextMock()
+    {
+        return $this->createMock(AmqpContext::class);
+    }
+
+    /**
+     * @return \PHPUnit_Framework_MockObject_MockObject|DelayStrategy
+     */
+    private function createDelayStrategyMock()
+    {
+        return $this->createMock(DelayStrategy::class);
+    }
 }

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

@@ -0,0 +1,37 @@
+<?php
+
+namespace Enqueue\AmqpBunny\Tests\Spec;
+
+use Enqueue\AmqpLib\AmqpConnectionFactory;
+use Enqueue\AmqpTools\RabbitMqDelayPluginDelayStrategy;
+use Interop\Queue\PsrContext;
+use Interop\Queue\Spec\SendAndReceiveDelayedMessageFromQueueSpec;
+
+/**
+ * @group functional
+ */
+class AmqpSendAndReceiveDelayedMessageWithDelayPluginStrategyTest extends SendAndReceiveDelayedMessageFromQueueSpec
+{
+    /**
+     * {@inheritdoc}
+     */
+    protected function createContext()
+    {
+        $factory = new AmqpConnectionFactory(getenv('AMQP_DSN'));
+        $factory->setDelayStrategy(new RabbitMqDelayPluginDelayStrategy());
+
+        return $factory->createContext();
+    }
+
+    /**
+     * {@inheritdoc}
+     */
+    protected function createQueue(PsrContext $context, $queueName)
+    {
+        $queue = parent::createQueue($context, $queueName);
+
+        $context->declareQueue($queue);
+
+        return $queue;
+    }
+}