Merge pull request #230 from bburnichon/feature/dequeuer-aware-interface

add dequeuer and dequeuer aware interface

Author: videlalvaro

DependencyInjection/OldSoundRabbitMqExtension.php

@@ -158,14 +158,17 @@ protected function loadConsumers()
                 $this->injectLoggedChannel($definition, $key, $consumer['connection']);
             }
 
-            $this->container->setDefinition(sprintf('old_sound_rabbit_mq.%s_consumer', $key), $definition);
+            $name = sprintf('old_sound_rabbit_mq.%s_consumer', $key);
+            $this->container->setDefinition($name, $definition);
+            $this->addDequeuerAwareCall($consumer['callback'], $name);
         }
     }
 
     protected function loadMultipleConsumers()
     {
         foreach ($this->config['multiple_consumers'] as $key => $consumer) {
             $queues = array();
+            $callbacks = array();
 
             if (empty($consumer['queues']) && empty($consumer['queues_provider'])) {
                 throw new InvalidConfigurationException(
@@ -177,6 +180,7 @@ protected function loadMultipleConsumers()
             foreach ($consumer['queues'] as $queueName => $queueOptions) {
                 $queues[$queueOptions['name']]  = $queueOptions;
                 $queues[$queueOptions['name']]['callback'] = array(new Reference($queueOptions['callback']), 'execute');
+                $callbacks[] = new Reference($queueOptions['callback']);
             }
 
             $definition = new Definition('%old_sound_rabbit_mq.multi_consumer.class%');
@@ -213,7 +217,14 @@ protected function loadMultipleConsumers()
                 $this->injectLoggedChannel($definition, $key, $consumer['connection']);
             }
 
-            $this->container->setDefinition(sprintf('old_sound_rabbit_mq.%s_multiple', $key), $definition);
+            $name = sprintf('old_sound_rabbit_mq.%s_multiple', $key);
+            $this->container->setDefinition($name, $definition);
+            if ($consumer['queues_provider']) {
+                $this->addDequeuerAwareCall($consumer['queues_provider'], $name);
+            }
+            foreach ($callbacks as $callback) {
+                $this->addDequeuerAwareCall($callback, $name);
+            }
         }
     }
 
@@ -231,7 +242,9 @@ protected function loadAnonConsumers()
                 $this->injectLoggedChannel($definition, $key, $anon['connection']);
             }
 
-            $this->container->setDefinition(sprintf('old_sound_rabbit_mq.%s_anon', $key), $definition);
+            $name = sprintf('old_sound_rabbit_mq.%s_anon', $key);
+            $this->container->setDefinition($name, $definition);
+            $this->addDequeuerAwareCall($anon['callback'], $name);
         }
     }
 
@@ -355,4 +368,23 @@ public function getAlias()
     {
         return 'old_sound_rabbit_mq';
     }
+
+    /**
+     * Add proper dequeuer aware call
+     *
+     * @param string $callback
+     * @param string $name
+     */
+    protected function addDequeuerAwareCall($callback, $name)
+    {
+        if (! $this->container->has($callback)) {
+            return;
+        }
+
+        $callbackDefinition = $this->container->findDefinition($callback);
+        $refClass           = new \ReflectionClass($callbackDefinition->getClass());
+        if ($refClass->implementsInterface('OldSound\RabbitMqBundle\RabbitMq\DequeuerAwareInterface')) {
+            $callbackDefinition->addMethodCall('setDequeuer', array(new Reference($name)));
+        }
+    }
 }

README.md

@@ -516,7 +516,11 @@ All the options of `queues-options` in the consumer are available for each queue
 Be aware that all queues are under the same exchange, it's up to you to set the correct routing for callbacks.
 
 The `queues_provider` is a optional service that dynamically provides queues.
-It must implement `QueuesProviderInterface`
+It must implement `QueuesProviderInterface`.
+
+Be aware that queues providers are responsible for the proper calls to `setDequeuer` and that callbacks are callables 
+(not `ConsumerInterface`). In case service providing queues implements `DequeuerAwareInterface`, a call to
+`setDequeuer` is added to the definition of the service with a `DequeuerInterface` currently being a `MultipleConsumer`.
 
 ### Anonymous Consumers ###
 

RabbitMq/BaseConsumer.php

@@ -2,9 +2,7 @@
 
 namespace OldSound\RabbitMqBundle\RabbitMq;
 
-use OldSound\RabbitMqBundle\RabbitMq\BaseAmqp;
-
-abstract class BaseConsumer extends BaseAmqp
+abstract class BaseConsumer extends BaseAmqp implements DequeuerInterface
 {
     protected $target;
 

RabbitMq/DequeuerAwareInterface.php

@@ -0,0 +1,8 @@
+<?php
+
+namespace OldSound\RabbitMqBundle\RabbitMq;
+
+interface DequeuerAwareInterface
+{
+    public function setDequeuer(DequeuerInterface $dequeuer);
+}

RabbitMq/DequeuerInterface.php

@@ -0,0 +1,29 @@
+<?php
+
+namespace OldSound\RabbitMqBundle\RabbitMq;
+
+interface DequeuerInterface
+{
+    /**
+     * Stop dequeuing messages.
+     *
+     * @return void
+     */
+    public function forceStopConsumer();
+
+    /**
+     * Set idle timeout
+     *
+     * @param $idleTimeout
+     *
+     * @return void
+     */
+    public function setIdleTimeout($idleTimeout);
+
+    /**
+     * Get current idle timeout
+     *
+     * @return int
+     */
+    public function getIdleTimeout();
+}

Tests/RabbitMq/BaseConsumerTest.php

@@ -0,0 +1,46 @@
+<?php
+
+namespace OldSound\RabbitMqBundle\Tests\RabbitMq;
+
+use OldSound\RabbitMqBundle\RabbitMq\BaseConsumer;
+
+class BaseConsumerTest extends \PHPUnit_Framework_TestCase
+{
+    /** @var BaseConsumer */
+    protected $consumer;
+
+    protected function setUp()
+    {
+        $amqpConnection =  $this->getMockBuilder('\PhpAmqpLib\Connection\AMQPConnection')
+            ->disableOriginalConstructor()
+            ->getMock();
+
+        $this->consumer = $this->getMockBuilder('\OldSound\RabbitMqBundle\RabbitMq\BaseConsumer')
+            ->setConstructorArgs(array($amqpConnection))
+            ->getMockForAbstractClass();
+    }
+
+    public function testItExtendsBaseAmqpInterface()
+    {
+        $this->assertInstanceOf('OldSound\RabbitMqBundle\RabbitMq\BaseAmqp', $this->consumer);
+    }
+
+    public function testItImplementsDequeuerInterface()
+    {
+        $this->assertInstanceOf('OldSound\RabbitMqBundle\RabbitMq\DequeuerInterface', $this->consumer);
+    }
+
+    public function testItsIdleTimeoutIsMutable()
+    {
+        $this->assertEquals(0, $this->consumer->getIdleTimeout());
+        $this->consumer->setIdleTimeout(42);
+        $this->assertEquals(42, $this->consumer->getIdleTimeout());
+    }
+
+    public function testForceStopConsumer()
+    {
+        $this->assertAttributeEquals(false, 'forceStop', $this->consumer);
+        $this->consumer->forceStopConsumer();
+        $this->assertAttributeEquals(true, 'forceStop', $this->consumer);
+    }
+}