Merge pull request #266 from tiborb/rabbitmq-dynamic-consumer

videlalvaro · videlalvaro · commit 915f68fa974f · 2015-05-11T17:32:36.000+02:00
Added rabbitmq:dynamic-consumer command
diff --git a/Command/BaseConsumerCommand.php b/Command/BaseConsumerCommand.php
@@ -87,14 +87,18 @@ protected function execute(InputInterface $input, OutputInterface $output)
         if (0 > $this->amount) {
             throw new \InvalidArgumentException("The -m option should be null or greater than 0");
         }
-
+        $this->initConsumer($input);
+        
+        $this->consumer->consume($this->amount);   
+    }
+    
+    protected function initConsumer($input) {
         $this->consumer = $this->getContainer()
-            ->get(sprintf($this->getConsumerService(), $input->getArgument('name')));
+                ->get(sprintf($this->getConsumerService(), $input->getArgument('name')));
 
-        if (!is_null($input->getOption('memory-limit')) && ctype_digit((string)$input->getOption('memory-limit')) && $input->getOption('memory-limit') > 0) {
+        if (!is_null($input->getOption('memory-limit')) && ctype_digit((string) $input->getOption('memory-limit')) && $input->getOption('memory-limit') > 0) {
             $this->consumer->setMemoryLimit($input->getOption('memory-limit'));
         }
         $this->consumer->setRoutingKey($input->getOption('route'));
-        $this->consumer->consume($this->amount);
     }
 }
diff --git a/Command/DynamicConsumerCommand.php b/Command/DynamicConsumerCommand.php
@@ -0,0 +1,39 @@
+<?php
+
+/**
+ * DynamicConsumerCommand
+ * 
+ * The context argument is passed to the consumer instance
+ * which can decide about the queue and routings it uses.
+ *
+ * @author Tibor Barna <tibor.barna@wiredminds.de>
+ */
+namespace OldSound\RabbitMqBundle\Command;
+
+use OldSound\RabbitMqBundle\Command\BaseConsumerCommand;
+use Symfony\Component\Console\Input\InputArgument;
+
+class DynamicConsumerCommand extends BaseConsumerCommand
+{
+    protected function configure()
+    {
+        parent::configure();
+        
+        $this
+            ->setName('rabbitmq:dynamic-consumer')
+            ->setDescription('Context-aware consumer')
+            ->addArgument('context', InputArgument::REQUIRED, 'Context the consumer runs in')
+            ;
+    }
+
+    protected function getConsumerService()
+    {
+        return 'old_sound_rabbit_mq.%s_dynamic';
+    }
+    
+    protected function initConsumer($input)
+    {
+        parent::initConsumer($input);
+        $this->consumer->setContext($input->getArgument('context'));
+    }
+}
diff --git a/DependencyInjection/Configuration.php b/DependencyInjection/Configuration.php
@@ -31,6 +31,7 @@ public function getConfigTreeBuilder()
         $this->addProducers($rootNode);
         $this->addConsumers($rootNode);
         $this->addMultipleConsumers($rootNode);
+        $this->addDynamicConsumers($rootNode);
         $this->addAnonConsumers($rootNode);
         $this->addRpcClients($rootNode);
         $this->addRpcServers($rootNode);
@@ -152,6 +153,37 @@ protected function addMultipleConsumers(ArrayNodeDefinition $node)
             ->end()
         ;
     }
+    
+    protected function addDynamicConsumers(ArrayNodeDefinition $node)
+    {
+        $node
+            ->fixXmlConfig('dynamic_consumer')
+            ->children()
+                ->arrayNode('dynamic_consumers')
+                    ->canBeUnset()
+                    ->useAttributeAsKey('key')
+                    ->prototype('array')
+                        ->append($this->getExchangeConfiguration())
+                        ->children()
+                            ->scalarNode('connection')->defaultValue('default')->end()
+                            ->scalarNode('callback')->isRequired()->end()
+                            ->scalarNode('idle_timeout')->end()
+                            ->scalarNode('auto_setup_fabric')->defaultTrue()->end()
+                            ->arrayNode('qos_options')
+                                ->canBeUnset()
+                                ->children()
+                                    ->scalarNode('prefetch_size')->defaultValue(0)->end()
+                                    ->scalarNode('prefetch_count')->defaultValue(0)->end()
+                                    ->booleanNode('global')->defaultFalse()->end()
+                                ->end()
+                            ->end()
+                            ->scalarNode('queue_options_provider')->isRequired()->end()
+                        ->end()
+                    ->end()
+                ->end()
+            ->end()
+        ;
+    }
 
     protected function addAnonConsumers(ArrayNodeDefinition $node)
     {
diff --git a/DependencyInjection/OldSoundRabbitMqExtension.php b/DependencyInjection/OldSoundRabbitMqExtension.php
@@ -49,6 +49,7 @@ public function load(array $configs, ContainerBuilder $container)
         $this->loadProducers();
         $this->loadConsumers();
         $this->loadMultipleConsumers();
+        $this->loadDynamicConsumers();
         $this->loadAnonConsumers();
         $this->loadRpcClients();
         $this->loadRpcServers();
@@ -233,6 +234,57 @@ protected function loadMultipleConsumers()
             }
         }
     }
+    
+    protected function loadDynamicConsumers()
+    {   
+        foreach ($this->config['dynamic_consumers'] as $key => $consumer) {
+            
+            if (empty($consumer['queue_options_provider'])) {
+                throw new InvalidConfigurationException(
+                    "Error on loading $key dynamic consumer. " .
+                    "'queue_provider' parameter should be defined."
+                );
+            }
+            
+            $definition = new Definition('%old_sound_rabbit_mq.dynamic_consumer.class%');
+            $definition
+                ->addTag('old_sound_rabbit_mq.base_amqp')
+                ->addTag('old_sound_rabbit_mq.consumer')
+                ->addTag('old_sound_rabbit_mq.dynamic_consumer')
+                ->addMethodCall('setExchangeOptions', array($this->normalizeArgumentKeys($consumer['exchange_options'])))
+                ->addMethodCall('setCallback', array(array(new Reference($consumer['callback']), 'execute')));
+
+            if (array_key_exists('qos_options', $consumer)) {
+                $definition->addMethodCall('setQosOptions', array(
+                    $consumer['qos_options']['prefetch_size'],
+                    $consumer['qos_options']['prefetch_count'],
+                    $consumer['qos_options']['global']
+                ));
+            }
+            
+            $definition->addMethodCall(
+                    'setQueueOptionsProvider',
+                    array(new Reference($consumer['queue_options_provider']))
+                );
+            
+            if(isset($consumer['idle_timeout'])) {
+                $definition->addMethodCall('setIdleTimeout', array($consumer['idle_timeout']));
+            }
+            if (!$consumer['auto_setup_fabric']) {
+                $definition->addMethodCall('disableAutoSetupFabric');
+            }
+
+            $this->injectConnection($definition, $consumer['connection']);
+            if ($this->collectorEnabled) {
+                $this->injectLoggedChannel($definition, $key, $consumer['connection']);
+            }
+
+            $name = sprintf('old_sound_rabbit_mq.%s_dynamic', $key);
+            $this->container->setDefinition($name, $definition);
+            $this->addDequeuerAwareCall($consumer['callback'], $name);
+            $this->addDequeuerAwareCall($consumer['queue_options_provider'], $name);
+        }
+    }
 
     protected function loadAnonConsumers()
     {
diff --git a/Provider/QueueOptionsProviderInterface.php b/Provider/QueueOptionsProviderInterface.php
@@ -0,0 +1,26 @@
+<?php
+
+namespace OldSound\RabbitMqBundle\Provider;
+
+/**
+ * Queue provider interface
+ *
+ * @author Tibor Barna <tibor.barna@wiredminds.de>
+ */
+interface QueueOptionsProviderInterface
+{
+    /**
+     * Return queue options
+     * 
+     * Example:
+     * array(
+     *   'name' => 'example_context',
+     *   'durable' => true,
+     *   'routing_keys' => array('key.*')
+     * )
+     * 
+     * @return array
+     * 
+     */
+    public function getQueueOptions($context = null);
+}
diff --git a/README.md b/README.md
@@ -516,6 +516,32 @@ Be aware that queues providers are responsible for the proper calls to `setDeque
 (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`.
 
+### Dynamic Consumers ###
+
+Sometimes you have to change the consumer's configuration on the fly.
+Dynamic consumers allow you to define the consumers queue options programatically, based on the context.
+
+e.g. In a scenario when the defined consumer must be responsible for a dynamic number of topics and you do not want (or can't) change it's configuration every time.
+
+Define a service `queue_options_provider` that implements the `QueueOptionsProviderInterface`, and add it to your `dynamic_consumers` configuration.
+
+```yaml
+dynamic_consumers:
+    proc_logs:
+        connection: default
+        exchange_options: {name: 'logs', type: topic}
+        callback: parse_logs_service
+        queue_options_provider: queue_options_provider_service
+```
+
+Example Usage:
+
+```bash
+$ ./app/console rabbitmq:dynamic-consumer proc_logs server1
+```
+
+In this case the `proc_logs` consumer runs for `server1` and it can decide over the queue options it uses.
+
 ### Anonymous Consumers ###
 
 Now, why will we ever need anonymous consumers? This sounds like some internet threat or something… Keep reading.
diff --git a/RabbitMq/DynamicConsumer.php b/RabbitMq/DynamicConsumer.php
@@ -0,0 +1,52 @@
+<?php
+
+namespace OldSound\RabbitMqBundle\RabbitMq;
+
+use OldSound\RabbitMqBundle\Provider\QueueOptionsProviderInterface;
+
+class DynamicConsumer extends Consumer{
+
+    /**
+     * Queue provider
+     *
+     * @var QueueOptionsProviderInterface
+     */
+    protected $queueOptionsProvider = null;
+    
+    /**
+     * Context the consumer runs in
+     * 
+     * @var string
+     */
+    protected $context = null;
+
+    /**
+     * QueueOptionsProvider setter
+     *
+     * @param QueueOptionsProviderInterface $queueOptionsProvider
+     *
+     * @return self
+     */
+    public function setQueueOptionsProvider(QueueOptionsProviderInterface $queueOptionsProvider)
+    {
+        $this->queueOptionsProvider = $queueOptionsProvider;
+        return $this;
+    }
+    
+    public function setContext($context)
+    {
+        $this->context = $context;
+    }
+
+
+    protected function setupConsumer()
+    {   
+        $this->mergeQueueOptions();
+        parent::setupConsumer();
+    }
+    
+    protected function mergeQueueOptions()
+    {   
+        $this->queueOptions = array_merge($this->queueOptions, $this->queueOptionsProvider->getQueueOptions($this->context));
+    }
+}
diff --git a/Resources/config/rabbitmq.xml b/Resources/config/rabbitmq.xml
@@ -10,6 +10,7 @@
         <parameter key="old_sound_rabbit_mq.producer.class">OldSound\RabbitMqBundle\RabbitMq\Producer</parameter>
         <parameter key="old_sound_rabbit_mq.consumer.class">OldSound\RabbitMqBundle\RabbitMq\Consumer</parameter>
         <parameter key="old_sound_rabbit_mq.multi_consumer.class">OldSound\RabbitMqBundle\RabbitMq\MultipleConsumer</parameter>
+        <parameter key="old_sound_rabbit_mq.dynamic_consumer.class">OldSound\RabbitMqBundle\RabbitMq\DynamicConsumer</parameter>
         <parameter key="old_sound_rabbit_mq.anon_consumer.class">OldSound\RabbitMqBundle\RabbitMq\AnonConsumer</parameter>
         <parameter key="old_sound_rabbit_mq.rpc_client.class">OldSound\RabbitMqBundle\RabbitMq\RpcClient</parameter>
         <parameter key="old_sound_rabbit_mq.rpc_server.class">OldSound\RabbitMqBundle\RabbitMq\RpcServer</parameter>
diff --git a/Tests/Command/DynamicConsumerCommandTest.php b/Tests/Command/DynamicConsumerCommandTest.php
@@ -0,0 +1,54 @@
+<?php
+
+namespace OldSound\RabbitMqBundle\Tests\Command;
+
+use OldSound\RabbitMqBundle\Command\DynamicConsumerCommand;
+
+use Symfony\Component\Console\Input\InputOption;
+
+class DynamicConsumerCommandTest extends BaseCommandTest{
+    
+    protected function setUp()
+    {
+        parent::setUp();
+        $this->definition->expects($this->any())
+            ->method('getOptions')
+            ->will($this->returnValue(array(
+                new InputOption('--verbose', '-v', InputOption::VALUE_NONE, 'Increase verbosity of messages.'),
+                new InputOption('--env', '-e', InputOption::VALUE_REQUIRED, 'The Environment name.', 'dev'),
+                new InputOption('--no-debug', null, InputOption::VALUE_NONE, 'Switches off debug mode.'),
+            )));
+        $this->application->expects($this->once())
+            ->method('getHelperSet')
+            ->will($this->returnValue($this->helperSet));
+
+        $this->command = new DynamicConsumerCommand();
+        $this->command->setApplication($this->application);
+    }
+    
+    /**
+     * testInputsDefinitionCommand
+     */
+    public function testInputsDefinitionCommand()
+    {
+        // check argument
+        $this->assertTrue($this->command->getDefinition()->hasArgument('name'));
+        $this->assertTrue($this->command->getDefinition()->getArgument('name')->isRequired()); // Name is required to find the service
+        
+        $this->assertTrue($this->command->getDefinition()->hasArgument('context'));
+        $this->assertTrue($this->command->getDefinition()->getArgument('context')->isRequired()); // Context is required for the queue options provider
+
+        //check options
+        $this->assertTrue($this->command->getDefinition()->hasOption('messages'));
+        $this->assertTrue($this->command->getDefinition()->getOption('messages')->isValueOptional()); // It should accept value
+
+        $this->assertTrue($this->command->getDefinition()->hasOption('route'));
+        $this->assertTrue($this->command->getDefinition()->getOption('route')->isValueOptional()); // It should accept value
+
+        $this->assertTrue($this->command->getDefinition()->hasOption('without-signals'));
+        $this->assertFalse($this->command->getDefinition()->getOption('without-signals')->acceptValue()); // It shouldn't accept value because it is a true/false input
+
+        $this->assertTrue($this->command->getDefinition()->hasOption('debug'));
+        $this->assertFalse($this->command->getDefinition()->getOption('debug')->acceptValue()); // It shouldn't accept value because it is a true/false input
+    }
+}
diff --git a/Tests/DependencyInjection/Fixtures/test.yml b/Tests/DependencyInjection/Fixtures/test.yml
@@ -122,7 +122,23 @@ old_sound_rabbit_mq:
                       - 'iphone.upload'
                     callback:        foo.multiple_test2.callback
             queues_provider: foo.queues_provider
-
+            
+    dynamic_consumers:
+        foo_dyn_consumer:
+            connection: foo_default
+            exchange_options:
+                name:           foo_dynamic_exchange
+                type:           direct
+            callback: foo.dynamic.callback
+            queue_options_provider: foo.dynamic.provider
+        bar_dyn_consumer:
+            connection: bar_default
+            exchange_options:
+                name:           bar_dynamic_exchange
+                type:           direct
+            callback: bar.dynamic.callback
+            queue_options_provider: bar.dynamic.provider
+            
     anon_consumers:
         foo_anon_consumer:
             connection:      foo_connection
diff --git a/Tests/DependencyInjection/OldSoundRabbitMqExtensionTest.php b/Tests/DependencyInjection/OldSoundRabbitMqExtensionTest.php
diff --git a/Tests/RabbitMq/ConsumerTest.php b/Tests/RabbitMq/ConsumerTest.php
diff --git a/Tests/RabbitMq/DynamicConsumerTest.php b/Tests/RabbitMq/DynamicConsumerTest.php

Original file line number	Diff line number	Diff line change
`@@ -87,14 +87,18 @@ protected function execute(InputInterface $input, OutputInterface $output)`
`87`	`87`	`if (0 > $this->amount) {`
`88`	`88`	`throw new \InvalidArgumentException("The -m option should be null or greater than 0");`
`89`	`89`	`}`
`90`		`-`
	`90`	`+ $this->initConsumer($input);`
	`91`	`+`
	`92`	`+ $this->consumer->consume($this->amount);`
	`93`	`+ }`
	`94`	`+`
	`95`	`+ protected function initConsumer($input) {`
`91`	`96`	`$this->consumer = $this->getContainer()`
`92`		`- ->get(sprintf($this->getConsumerService(), $input->getArgument('name')));`
	`97`	`+ ->get(sprintf($this->getConsumerService(), $input->getArgument('name')));`
`93`	`98`
`94`		`- if (!is_null($input->getOption('memory-limit')) && ctype_digit((string)$input->getOption('memory-limit')) && $input->getOption('memory-limit') > 0) {`
	`99`	`+ if (!is_null($input->getOption('memory-limit')) && ctype_digit((string) $input->getOption('memory-limit')) && $input->getOption('memory-limit') > 0) {`
`95`	`100`	`$this->consumer->setMemoryLimit($input->getOption('memory-limit'));`
`96`	`101`	`}`
`97`	`102`	`$this->consumer->setRoutingKey($input->getOption('route'));`
`98`		`- $this->consumer->consume($this->amount);`
`99`	`103`	`}`
`100`	`104`	`}`