Added per-queue message TTL in AMQPBackend (#223)

sonata-project · Dec 22, 2016 · 96eafb2 · 96eafb2
1 parent 1484b93
commit 96eafb2
Show file tree

Hide file tree

Showing 4 changed files with 134 additions and 55 deletions.
diff --git a/Backend/AMQPBackend.php b/Backend/AMQPBackend.php
@@ -57,27 +57,34 @@ class AMQPBackend implements BackendInterface
      */
     protected $deadLetterRoutingKey;
 
+    /**
+     * @var null|int
+     */
+    protected $ttl;
+
     /**
      * @var AMQPBackendDispatcher
      */
     protected $dispatcher = null;
 
     /**
-     * @param string $exchange
-     * @param string $queue
-     * @param string $recover
-     * @param string $key
-     * @param string $deadLetterExchange
-     * @param string $deadLetterRoutingKey
+     * @param string   $exchange
+     * @param string   $queue
+     * @param string   $recover
+     * @param string   $key
+     * @param string   $deadLetterExchange
+     * @param string   $deadLetterRoutingKey
+     * @param null|int $ttl
      */
-    public function __construct($exchange, $queue, $recover, $key, $deadLetterExchange = null, $deadLetterRoutingKey = null)
+    public function __construct($exchange, $queue, $recover, $key, $deadLetterExchange = null, $deadLetterRoutingKey = null, $ttl = null)
     {
         $this->exchange = $exchange;
         $this->queue = $queue;
         $this->recover = $recover;
         $this->key = $key;
         $this->deadLetterExchange = $deadLetterExchange;
         $this->deadLetterRoutingKey = $deadLetterRoutingKey;
+        $this->ttl = $ttl;
 
         if (!class_exists('PhpAmqpLib\Message\AMQPMessage')) {
             throw new \RuntimeException('Please install php-amqplib/php-amqplib dependency');
@@ -107,6 +114,10 @@ public function initialize()
             }
         }
 
+        if ($this->ttl !== null) {
+            $args['x-message-ttl'] = array('I', $this->ttl);
+        }
+
         /*
          * name: $queue
          * passive: false

diff --git a/DependencyInjection/Configuration.php b/DependencyInjection/Configuration.php
@@ -213,13 +213,19 @@ protected function getQueueNode()
 If is set, failed messages will be rejected and sent to this exchange
 EOF;
 
-        $deadLetterRoutingKey = <<<'EOF'
+        $deadLetterRoutingKeyInfo = <<<'EOF'
 Only used by RabbitMQ
 
 If set, failed messages will be routed to the queue using this key by dead-letter-exchange,
 otherwise it will be requeued to the original queue if `dead-letter-exchange` is set.
 
 If set, the queue must be configured with this key as `routing_key`.
+EOF;
+
+        $ttlInfo = <<<'EOF'
+Only used by RabbitMQ
+
+Defines the per-queue message time-to-live (milliseconds)
 EOF;
 
         $typesInfo = <<<'EOF'
@@ -260,7 +266,12 @@ protected function getQueueNode()
                     ->defaultValue(null)
                 ->end()
                 ->scalarNode('dead_letter_routing_key')
-                    ->info($deadLetterRoutingKey)
+                    ->info($deadLetterRoutingKeyInfo)
+                    ->defaultValue(null)
+                ->end()
+                ->integerNode('ttl')
+                    ->info($ttlInfo)
+                    ->min(0)
                     ->defaultValue(null)
                 ->end()
 

diff --git a/DependencyInjection/SonataNotificationExtension.php b/DependencyInjection/SonataNotificationExtension.php
@@ -330,6 +330,7 @@ protected function configureRabbitmq(ContainerBuilder $container, array $config)
                 'recover' => false,
                 'dead_letter_exchange' => null,
                 'dead_letter_routing_key' => null,
+                'ttl' => null,
             ));
         }
 
@@ -375,7 +376,8 @@ protected function configureRabbitmq(ContainerBuilder $container, array $config)
                 $queue['recover'],
                 $queue['routing_key'],
                 $queue['dead_letter_exchange'],
-                $queue['dead_letter_routing_key']
+                $queue['dead_letter_routing_key'],
+                $queue['ttl']
             );
 
             $amqBackends[$pos] = array(
@@ -412,10 +414,11 @@ protected function configureRabbitmq(ContainerBuilder $container, array $config)
      * @param string           $key
      * @param string           $deadLetterExchange
      * @param string           $deadLetterRoutingKey
+     * @param int|null         $ttl
      *
      * @return string
      */
-    protected function createAMQPBackend(ContainerBuilder $container, $exchange, $name, $recover, $key = '', $deadLetterExchange = null, $deadLetterRoutingKey = null)
+    protected function createAMQPBackend(ContainerBuilder $container, $exchange, $name, $recover, $key = '', $deadLetterExchange = null, $deadLetterRoutingKey = null, $ttl = null)
     {
         $id = 'sonata.notification.backend.rabbitmq.'.$this->amqpCounter++;
 
@@ -428,6 +431,7 @@ protected function createAMQPBackend(ContainerBuilder $container, $exchange, $na
                 $key,
                 $deadLetterExchange,
                 $deadLetterRoutingKey,
+                $ttl,
             )
         );
         $definition->setPublic(false);

diff --git a/Tests/Backend/AMQPBackendTest.php b/Tests/Backend/AMQPBackendTest.php
@@ -18,6 +18,7 @@ class AMQPBackendTest extends \PHPUnit_Framework_TestCase
     const KEY = 'message.type.foo';
     const DEAD_LETTER_EXCHANGE = 'dlx';
     const DEAD_LETTER_ROUTING_KEY = 'message.type.dl';
+    const TTL = 60000;
 
     protected function setUp()
     {
@@ -32,27 +33,30 @@ public function testInitializeWithNoDeadLetterExchangeAndNoDeadLetterRoutingKey(
 
         $channelMock->expects($this->once())
             ->method('exchange_declare')
-            ->with($this->equalTo(self::EXCHANGE),
-                   $this->equalTo('direct'),
-                   $this->isType('boolean'),
-                   $this->isType('boolean'),
-                   $this->isType('boolean')
+            ->with(
+                $this->equalTo(self::EXCHANGE),
+                $this->equalTo('direct'),
+                $this->isType('boolean'),
+                $this->isType('boolean'),
+                $this->isType('boolean')
              );
         $channelMock->expects($this->once())
             ->method('queue_declare')
-            ->with($this->equalTo(self::QUEUE),
-                   $this->isType('boolean'),
-                   $this->isType('boolean'),
-                   $this->isType('boolean'),
-                   $this->isType('boolean'),
-                   $this->isType('boolean'),
-                   $this->equalTo(array())
+            ->with(
+                $this->equalTo(self::QUEUE),
+                $this->isType('boolean'),
+                $this->isType('boolean'),
+                $this->isType('boolean'),
+                $this->isType('boolean'),
+                $this->isType('boolean'),
+                $this->equalTo(array())
              );
         $channelMock->expects($this->once())
             ->method('queue_bind')
-            ->with($this->equalTo(self::QUEUE),
-                   $this->equalTo(self::EXCHANGE),
-                   $this->equalTo(self::KEY)
+            ->with(
+                $this->equalTo(self::QUEUE),
+                $this->equalTo(self::EXCHANGE),
+                $this->equalTo(self::KEY)
              );
 
         $backend->initialize();
@@ -82,15 +86,16 @@ public function testInitializeWithDeadLetterExchangeAndNoDeadLetterRoutingKey()
              );
         $channelMock->expects($this->once())
             ->method('queue_declare')
-            ->with($this->equalTo(self::QUEUE),
-                   $this->isType('boolean'),
-                   $this->isType('boolean'),
-                   $this->isType('boolean'),
-                   $this->isType('boolean'),
-                   $this->isType('boolean'),
-                   $this->equalTo(array(
-                       'x-dead-letter-exchange' => array('S', self::DEAD_LETTER_EXCHANGE),
-                   ))
+            ->with(
+                $this->equalTo(self::QUEUE),
+                $this->isType('boolean'),
+                $this->isType('boolean'),
+                $this->isType('boolean'),
+                $this->isType('boolean'),
+                $this->isType('boolean'),
+                $this->equalTo(array(
+                    'x-dead-letter-exchange' => array('S', self::DEAD_LETTER_EXCHANGE),
+                ))
              );
         $channelMock->expects($this->exactly(2))
             ->method('queue_bind')
@@ -116,39 +121,87 @@ public function testInitializeWithDeadLetterExchangeAndDeadLetterRoutingKey()
 
         $channelMock->expects($this->once())
             ->method('exchange_declare')
-            ->with($this->equalTo(self::EXCHANGE),
-                   $this->equalTo('direct'),
-                   $this->isType('boolean'),
-                   $this->isType('boolean'),
-                   $this->isType('boolean')
+            ->with(
+                $this->equalTo(self::EXCHANGE),
+                $this->equalTo('direct'),
+                $this->isType('boolean'),
+                $this->isType('boolean'),
+                $this->isType('boolean')
              );
         $channelMock->expects($this->once())
             ->method('queue_declare')
-            ->with($this->equalTo(self::QUEUE),
-                   $this->isType('boolean'),
-                   $this->isType('boolean'),
-                   $this->isType('boolean'),
-                   $this->isType('boolean'),
-                   $this->isType('boolean'),
-                   $this->equalTo(array(
-                       'x-dead-letter-exchange' => array('S', self::DEAD_LETTER_EXCHANGE),
-                       'x-dead-letter-routing-key' => array('S', self::DEAD_LETTER_ROUTING_KEY),
-                   ))
+            ->with(
+                $this->equalTo(self::QUEUE),
+                $this->isType('boolean'),
+                $this->isType('boolean'),
+                $this->isType('boolean'),
+                $this->isType('boolean'),
+                $this->isType('boolean'),
+                $this->equalTo(array(
+                   'x-dead-letter-exchange' => array('S', self::DEAD_LETTER_EXCHANGE),
+                   'x-dead-letter-routing-key' => array('S', self::DEAD_LETTER_ROUTING_KEY),
+                ))
              );
         $channelMock->expects($this->once())
             ->method('queue_bind')
-            ->with($this->equalTo(self::QUEUE),
-                   $this->equalTo(self::EXCHANGE),
-                   $this->equalTo(self::KEY)
+            ->with(
+                $this->equalTo(self::QUEUE),
+                $this->equalTo(self::EXCHANGE),
+                $this->equalTo(self::KEY)
+             );
+
+        $backend->initialize();
+    }
+
+    public function testInitializeWithTTL()
+    {
+        list($backend, $channelMock) = $this->getBackendAndChannelMock(false, null, null, self::TTL);
+
+        $channelMock->expects($this->once())
+            ->method('exchange_declare')
+            ->with(
+                $this->equalTo(self::EXCHANGE),
+                $this->equalTo('direct'),
+                $this->isType('boolean'),
+                $this->isType('boolean'),
+                $this->isType('boolean')
+             );
+        $channelMock->expects($this->once())
+            ->method('queue_declare')
+            ->with(
+                $this->equalTo(self::QUEUE),
+                $this->isType('boolean'),
+                $this->isType('boolean'),
+                $this->isType('boolean'),
+                $this->isType('boolean'),
+                $this->isType('boolean'),
+                $this->equalTo(array(
+                    'x-message-ttl' => array('I', self::TTL),
+                ))
+             );
+        $channelMock->expects($this->once())
+            ->method('queue_bind')
+            ->with(
+                $this->equalTo(self::QUEUE),
+                $this->equalTo(self::EXCHANGE),
+                $this->equalTo(self::KEY)
              );
 
         $backend->initialize();
     }
 
-    protected function getBackendAndChannelMock($recover = false, $deadLetterExchange = null, $deadLetterRoutingKey = null)
+    protected function getBackendAndChannelMock($recover = false, $deadLetterExchange = null, $deadLetterRoutingKey = null, $ttl = null)
     {
         $mock = $this->getMockBuilder('\Sonata\NotificationBundle\Backend\AMQPBackend')
-            ->setConstructorArgs(array(self::EXCHANGE, self::QUEUE, $recover, self::KEY, $deadLetterExchange, $deadLetterRoutingKey))
+            ->setConstructorArgs(array(
+                self::EXCHANGE,
+                self::QUEUE,
+                $recover,
+                self::KEY,
+                $deadLetterExchange,
+                $deadLetterRoutingKey,
+                $ttl,
+            ))
             ->setMethods(array('getIterator'))
             ->getMock();