Delay Strategy Configuration

phpunit.xml.dist

@@ -29,6 +29,14 @@
             <directory>pkg/amqp-bunny/Tests</directory>
         </testsuite>
 
+        <testsuite name="amqp-lib transport">
+            <directory>pkg/amqp-lib/Tests</directory>
+        </testsuite>
+
+        <testsuite name="amqp-tools">
+            <directory>pkg/amqp-tools/Tests</directory>
+        </testsuite>
+
         <testsuite name="fs transport">
             <directory>pkg/fs/Tests</directory>
         </testsuite>

pkg/amqp-bunny/Symfony/RabbitMqAmqpBunnyTransportFactory.php

@@ -2,6 +2,7 @@
 
 namespace Enqueue\AmqpBunny\Symfony;
 
+use Enqueue\AmqpTools\DelayStrategyTransportFactoryTrait;
 use Enqueue\Client\Amqp\RabbitMqDriver;
 use Symfony\Component\Config\Definition\Builder\ArrayNodeDefinition;
 use Symfony\Component\DependencyInjection\ContainerBuilder;
@@ -10,6 +11,8 @@
 
 class RabbitMqAmqpBunnyTransportFactory extends AmqpBunnyTransportFactory
 {
+    use DelayStrategyTransportFactoryTrait;
+
     /**
      * @param string $name
      */
@@ -27,13 +30,25 @@ public function addConfiguration(ArrayNodeDefinition $builder)
 
         $builder
             ->children()
-                ->booleanNode('delay_plugin_installed')
-                    ->defaultFalse()
-                    ->info('The option tells whether RabbitMQ broker has delay plugin installed or not')
+                ->scalarNode('delay_strategy')
+                    ->defaultValue('dlx')
+                    ->info('The delay strategy to be used. Possible values are "dlx", "delayed_message_plugin" or service id')
                 ->end()
         ;
     }
 
+    /**
+     * {@inheritdoc}
+     */
+    public function createConnectionFactory(ContainerBuilder $container, array $config)
+    {
+        $factoryId = parent::createConnectionFactory($container, $config);
+
+        $this->registerDelayStrategy($container, $config, $factoryId, $this->getName());
+
+        return $factoryId;
+    }
+
     /**
      * {@inheritdoc}
      */

pkg/amqp-bunny/Tests/Symfony/RabbitMqAmqpBunnyTransportFactoryTest.php

@@ -58,7 +58,7 @@ public function testShouldAllowAddConfiguration()
             'user' => 'guest',
             'pass' => 'guest',
             'vhost' => '/',
-            'delay_plugin_installed' => false,
+            'delay_strategy' => 'dlx',
             'lazy' => true,
             'receive_method' => 'basic_get',
             'heartbeat' => 0,
@@ -78,7 +78,7 @@ public function testShouldCreateConnectionFactory()
             'pass' => 'guest',
             'vhost' => '/',
             'persisted' => false,
-            'delay_plugin_installed' => false,
+            'delay_strategy' => null,
         ]);
 
         $this->assertTrue($container->hasDefinition($serviceId));
@@ -91,7 +91,7 @@ public function testShouldCreateConnectionFactory()
             'pass' => 'guest',
             'vhost' => '/',
             'persisted' => false,
-            'delay_plugin_installed' => false,
+            'delay_strategy' => null,
         ]], $factory->getArguments());
     }
 
@@ -108,7 +108,7 @@ public function testShouldCreateContext()
             'pass' => 'guest',
             'vhost' => '/',
             'persisted' => false,
-            'delay_plugin_installed' => false,
+            'delay_strategy' => null,
         ]);
 
         $this->assertEquals('enqueue.transport.rabbitmq_amqp_bunny.context', $serviceId);

pkg/amqp-ext/Symfony/RabbitMqAmqpTransportFactory.php

@@ -2,6 +2,7 @@
 
 namespace Enqueue\AmqpExt\Symfony;
 
+use Enqueue\AmqpTools\DelayStrategyTransportFactoryTrait;
 use Enqueue\Client\Amqp\RabbitMqDriver;
 use Symfony\Component\Config\Definition\Builder\ArrayNodeDefinition;
 use Symfony\Component\DependencyInjection\ContainerBuilder;
@@ -10,6 +11,8 @@
 
 class RabbitMqAmqpTransportFactory extends AmqpTransportFactory
 {
+    use DelayStrategyTransportFactoryTrait;
+
     /**
      * @param string $name
      */
@@ -27,13 +30,25 @@ public function addConfiguration(ArrayNodeDefinition $builder)
 
         $builder
             ->children()
-                ->booleanNode('delay_plugin_installed')
-                    ->defaultFalse()
-                    ->info('The option tells whether RabbitMQ broker has delay plugin installed or not')
+                ->scalarNode('delay_strategy')
+                    ->defaultValue('dlx')
+                    ->info('The delay strategy to be used. Possible values are "dlx", "delayed_message_plugin" or service id')
                 ->end()
         ;
     }
 
+    /**
+     * {@inheritdoc}
+     */
+    public function createConnectionFactory(ContainerBuilder $container, array $config)
+    {
+        $factoryId = parent::createConnectionFactory($container, $config);
+
+        $this->registerDelayStrategy($container, $config, $factoryId, $this->getName());
+
+        return $factoryId;
+    }
+
     /**
      * {@inheritdoc}
      */

pkg/amqp-ext/Tests/Symfony/RabbitMqAmqpTransportFactoryTest.php

@@ -59,7 +59,7 @@ public function testShouldAllowAddConfiguration()
             'pass' => 'guest',
             'vhost' => '/',
             'persisted' => false,
-            'delay_plugin_installed' => false,
+            'delay_strategy' => 'dlx',
             'lazy' => true,
             'receive_method' => 'basic_get',
         ], $config);
@@ -78,7 +78,7 @@ public function testShouldCreateConnectionFactory()
             'pass' => 'guest',
             'vhost' => '/',
             'persisted' => false,
-            'delay_plugin_installed' => false,
+            'delay_strategy' => null,
         ]);
 
         $this->assertTrue($container->hasDefinition($serviceId));
@@ -91,7 +91,7 @@ public function testShouldCreateConnectionFactory()
             'pass' => 'guest',
             'vhost' => '/',
             'persisted' => false,
-            'delay_plugin_installed' => false,
+            'delay_strategy' => null,
         ]], $factory->getArguments());
     }
 
@@ -108,7 +108,7 @@ public function testShouldCreateContext()
             'pass' => 'guest',
             'vhost' => '/',
             'persisted' => false,
-            'delay_plugin_installed' => false,
+            'delay_strategy' => null,
         ]);
 
         $this->assertEquals('enqueue.transport.rabbitmq_amqp.context', $serviceId);

pkg/amqp-lib/Symfony/RabbitMqAmqpLibTransportFactory.php

@@ -2,6 +2,7 @@
 
 namespace Enqueue\AmqpLib\Symfony;
 
+use Enqueue\AmqpTools\DelayStrategyTransportFactoryTrait;
 use Enqueue\Client\Amqp\RabbitMqDriver;
 use Symfony\Component\Config\Definition\Builder\ArrayNodeDefinition;
 use Symfony\Component\DependencyInjection\ContainerBuilder;
@@ -10,6 +11,8 @@
 
 class RabbitMqAmqpLibTransportFactory extends AmqpLibTransportFactory
 {
+    use DelayStrategyTransportFactoryTrait;
+
     /**
      * @param string $name
      */
@@ -27,13 +30,25 @@ public function addConfiguration(ArrayNodeDefinition $builder)
 
         $builder
             ->children()
-                ->booleanNode('delay_plugin_installed')
-                    ->defaultFalse()
-                    ->info('The option tells whether RabbitMQ broker has delay plugin installed or not')
+                ->scalarNode('delay_strategy')
+                    ->defaultValue('dlx')
+                    ->info('The delay strategy to be used. Possible values are "dlx", "delayed_message_plugin" or service id')
                 ->end()
         ;
     }
 
+    /**
+     * {@inheritdoc}
+     */
+    public function createConnectionFactory(ContainerBuilder $container, array $config)
+    {
+        $factoryId = parent::createConnectionFactory($container, $config);
+
+        $this->registerDelayStrategy($container, $config, $factoryId, $this->getName());
+
+        return $factoryId;
+    }
+
     /**
      * {@inheritdoc}
      */

pkg/amqp-lib/Tests/AmqpConsumerTest.php

@@ -102,6 +102,7 @@ public function testShouldReturnMessageOnReceiveNoWait()
         $amqpMessage = new \PhpAmqpLib\Message\AMQPMessage('body');
         $amqpMessage->delivery_info['delivery_tag'] = 'delivery-tag';
         $amqpMessage->delivery_info['redelivered'] = true;
+        $amqpMessage->delivery_info['routing_key'] = 'routing-key';
 
         $channel = $this->createChannelMock();
         $channel
@@ -120,13 +121,15 @@ public function testShouldReturnMessageOnReceiveNoWait()
         $this->assertInstanceOf(AmqpMessage::class, $message);
         $this->assertSame('body', $message->getBody());
         $this->assertSame('delivery-tag', $message->getDeliveryTag());
+        $this->assertSame('routing-key', $message->getRoutingKey());
         $this->assertTrue($message->isRedelivered());
     }
 
     public function testShouldReturnMessageOnReceiveWithReceiveMethodBasicGet()
     {
         $amqpMessage = new \PhpAmqpLib\Message\AMQPMessage('body');
         $amqpMessage->delivery_info['delivery_tag'] = 'delivery-tag';
+        $amqpMessage->delivery_info['routing_key'] = 'routing-key';
         $amqpMessage->delivery_info['redelivered'] = true;
 
         $channel = $this->createChannelMock();
@@ -146,6 +149,7 @@ public function testShouldReturnMessageOnReceiveWithReceiveMethodBasicGet()
         $this->assertInstanceOf(AmqpMessage::class, $message);
         $this->assertSame('body', $message->getBody());
         $this->assertSame('delivery-tag', $message->getDeliveryTag());
+        $this->assertSame('routing-key', $message->getRoutingKey());
         $this->assertTrue($message->isRedelivered());
     }
 

pkg/amqp-lib/Tests/AmqpContextTest.php

@@ -8,6 +8,7 @@
 use Interop\Amqp\Impl\AmqpTopic;
 use PhpAmqpLib\Channel\AMQPChannel;
 use PhpAmqpLib\Connection\AbstractConnection;
+use PhpAmqpLib\Wire\AMQPTable;
 use PHPUnit\Framework\TestCase;
 
 class AmqpContextTest extends TestCase
@@ -26,7 +27,7 @@ public function testShouldDeclareTopic()
                 $this->isTrue(),
                 $this->isTrue(),
                 $this->isTrue(),
-                $this->identicalTo(['key' => 'value']),
+                $this->isInstanceOf(AMQPTable::class),
                 $this->isNull()
             )
         ;
@@ -94,7 +95,7 @@ public function testShouldDeclareQueue()
                 $this->isTrue(),
                 $this->isTrue(),
                 $this->isTrue(),
-                $this->identicalTo(['key' => 'value']),
+                $this->isInstanceOf(AMQPTable::class),
                 $this->isNull()
             )
         ;

pkg/amqp-lib/Tests/AmqpProducerTest.php

@@ -2,6 +2,7 @@
 
 namespace Enqueue\AmqpLib\Tests;
 
+use Enqueue\AmqpLib\AmqpContext;
 use Enqueue\AmqpLib\AmqpProducer;
 use Enqueue\Test\ClassExtensionTrait;
 use Interop\Amqp\Impl\AmqpMessage;
@@ -23,7 +24,7 @@ class AmqpProducerTest extends TestCase
 
     public function testCouldBeConstructedWithRequiredArguments()
     {
-        new AmqpProducer($this->createAmqpChannelMock());
+        new AmqpProducer($this->createAmqpChannelMock(), $this->createContextMock());
     }
 
     public function testShouldImplementPsrProducerInterface()
@@ -33,7 +34,7 @@ public function testShouldImplementPsrProducerInterface()
 
     public function testShouldThrowExceptionWhenDestinationTypeIsInvalid()
     {
-        $producer = new AmqpProducer($this->createAmqpChannelMock());
+        $producer = new AmqpProducer($this->createAmqpChannelMock(), $this->createContextMock());
 
         $this->expectException(InvalidDestinationException::class);
         $this->expectExceptionMessage('The destination must be an instance of Interop\Amqp\AmqpQueue but got');
@@ -43,7 +44,7 @@ public function testShouldThrowExceptionWhenDestinationTypeIsInvalid()
 
     public function testShouldThrowExceptionWhenMessageTypeIsInvalid()
     {
-        $producer = new AmqpProducer($this->createAmqpChannelMock());
+        $producer = new AmqpProducer($this->createAmqpChannelMock(), $this->createContextMock());
 
         $this->expectException(InvalidMessageException::class);
         $this->expectExceptionMessage('The message must be an instance of Interop\Amqp\AmqpMessage but it is');
@@ -70,7 +71,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);
 
         $this->assertEquals('body', $amqpMessage->getBody());
@@ -92,7 +93,7 @@ public function testShouldPublishMessageToQueue()
 
         $queue = new AmqpQueue('queue');
 
-        $producer = new AmqpProducer($channel);
+        $producer = new AmqpProducer($channel, $this->createContextMock());
         $producer->send($queue, new AmqpMessage('body'));
 
         $this->assertEquals('body', $amqpMessage->getBody());
@@ -111,7 +112,7 @@ public function testShouldSetMessageHeaders()
             }))
         ;
 
-        $producer = new AmqpProducer($channel);
+        $producer = new AmqpProducer($channel, $this->createContextMock());
         $producer->send(new AmqpTopic('name'), new AmqpMessage('body', [], ['content_type' => 'text/plain']));
 
         $this->assertEquals(['content_type' => 'text/plain'], $amqpMessage->get_properties());
@@ -130,7 +131,7 @@ public function testShouldSetMessageProperties()
             }))
         ;
 
-        $producer = new AmqpProducer($channel);
+        $producer = new AmqpProducer($channel, $this->createContextMock());
         $producer->send(new AmqpTopic('name'), new AmqpMessage('body', ['key' => 'value']));
 
         $properties = $amqpMessage->get_properties();
@@ -163,4 +164,12 @@ private function createAmqpChannelMock()
     {
         return $this->createMock(AMQPChannel::class);
     }
+
+    /**
+     * @return \PHPUnit_Framework_MockObject_MockObject|AmqpContext
+     */
+    private function createContextMock()
+    {
+        return $this->createMock(AmqpContext::class);
+    }
 }