Merge pull request #415 from tiborb/multiple_consumer_context

Added optional context argument to multiple consumer

Author: tiborb

Command/MultipleConsumerCommand.php

@@ -2,18 +2,28 @@
 
 namespace OldSound\RabbitMqBundle\Command;
 
+use Symfony\Component\Console\Input\InputArgument;
+
 class MultipleConsumerCommand extends BaseConsumerCommand
 {
     protected function configure()
     {
         parent::configure();
 
-        $this->setDescription('Executes a consumer that uses multiple queues');
-        $this->setName('rabbitmq:multiple-consumer');
+        $this->setDescription('Executes a consumer that uses multiple queues')
+                ->setName('rabbitmq:multiple-consumer')
+                ->addArgument('context', InputArgument::OPTIONAL, 'Context the consumer runs in')
+        ;
     }
 
     protected function getConsumerService()
     {
         return 'old_sound_rabbit_mq.%s_multiple';
     }
-}
+
+    protected function initConsumer($input)
+    {
+        parent::initConsumer($input);
+        $this->consumer->setContext($input->getArgument('context'));
+    }
+}

Provider/QueuesProviderInterface.php

@@ -12,7 +12,22 @@ interface QueuesProviderInterface
     /**
      * Return array of queues
      *
+     * Example:
+     * array(
+     *    'queue_name' => array(
+     *       'durable' => false,
+     *       'exclusive' => false,
+     *       'passive' => false,
+     *       'nowait' => false,
+     *       'auto_delete' => false,
+     *       'routing_keys' => array('key.1', 'key.2'),
+     *       'arguments' => array(),
+     *       'ticket' => '',
+     *       'callback' => array($callback, 'execute')
+     *    )
+     * );
      * @return array
+     * 
      */
     public function getQueues();
 }

RabbitMq/MultipleConsumer.php

@@ -16,6 +16,13 @@ class MultipleConsumer extends Consumer
      * @var QueuesProviderInterface
      */
     protected $queuesProvider = null;
+    
+    /**
+     * Context the consumer runs in
+     *
+     * @var string
+     */
+    protected $context = null;
 
     /**
      * QueuesProvider setter
@@ -39,6 +46,11 @@ public function setQueues(array $queues)
     {
         $this->queues = $queues;
     }
+    
+    public function setContext($context)
+    {
+        $this->context = $context;
+    }
 
     protected function setupConsumer()
     {
@@ -102,7 +114,7 @@ protected function mergeQueues()
         if ($this->queuesProvider) {
             $this->queues = array_merge(
                 $this->queues,
-                $this->queuesProvider->getQueues()
+                $this->queuesProvider->getQueues($this->context)
             );
         }
     }

Tests/Command/MultipleConsumerCommandTest.php

@@ -0,0 +1,55 @@
+<?php
+
+namespace OldSound\RabbitMqBundle\Tests\Command;
+
+use OldSound\RabbitMqBundle\Command\MultipleConsumerCommand;
+
+use Symfony\Component\Console\Input\InputOption;
+
+class MultipleConsumerCommandTest 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 MultipleConsumerCommand();
+        $this->command->setApplication($this->application);
+    }
+    
+    /**
+     * testInputsDefinitionCommand
+     */
+    public function testInputsDefinitionCommand()
+    {
+        // check argument
+        $definition = $this->command->getDefinition();
+        $this->assertTrue($definition->hasArgument('name'));
+        $this->assertTrue($definition->getArgument('name')->isRequired()); // Name is required to find the service
+        
+        $this->assertTrue($definition->hasArgument('context'));
+        $this->assertFalse($definition->getArgument('context')->isRequired()); // Context is required for the queue options provider
+
+        //check options
+        $this->assertTrue($definition->hasOption('messages'));
+        $this->assertTrue($definition->getOption('messages')->isValueOptional()); // It should accept value
+
+        $this->assertTrue($definition->hasOption('route'));
+        $this->assertTrue($definition->getOption('route')->isValueOptional()); // It should accept value
+
+        $this->assertTrue($definition->hasOption('without-signals'));
+        $this->assertFalse($definition->getOption('without-signals')->acceptValue()); // It shouldn't accept value because it is a true/false input
+
+        $this->assertTrue($definition->hasOption('debug'));
+        $this->assertFalse($definition->getOption('debug')->acceptValue()); // It shouldn't accept value because it is a true/false input
+    }
+}

Tests/RabbitMq/MultipleConsumerTest.php

@@ -111,6 +111,29 @@ public function testQueuesProvider($processFlag, $expectedMethod, $expectedReque
         $this->multipleConsumer->processQueueMessage('test-1', $amqpMessage);
         $this->multipleConsumer->processQueueMessage('test-2', $amqpMessage);
     }
+    
+    public function testQueuesPrivider()
+    {
+        $amqpConnection = $this->prepareAMQPConnection();
+        $amqpChannel = $this->prepareAMQPChannel();
+        $this->multipleConsumer->setContext('foo');
+        
+        $queuesProvider = $this->prepareQueuesProvider();
+        $queuesProvider->expects($this->once())
+            ->method('getQueues')
+            ->will($this->returnValue(
+                array(
+                    'queue_foo' => array()
+                )
+            ));
+        
+        $this->multipleConsumer->setQueuesProvider($queuesProvider);
+        
+        $reflectionClass = new \ReflectionClass(get_class($this->multipleConsumer));
+        $reflectionMethod = $reflectionClass->getMethod('mergeQueues');
+        $reflectionMethod->setAccessible(true);
+        $reflectionMethod->invoke($this->multipleConsumer);        
+    }
 
     /**
      * Check queues provider works well with static queues together