Merge pull request #22 from M6Web/fix/normalize-keys-queue-arguments

disabled normalize keys for queue arguments

composer.json

@@ -16,7 +16,8 @@
     "require-dev" : {
         "atoum/atoum-bundle" : "dev-master",
         "m6web/coke" : "~1.2",
-        "m6web/symfony2-coding-standard" : "~1.1"
+        "m6web/symfony2-coding-standard" : "~1.1",
+        "symfony/yaml" : "~2.7.5"
     },
     "suggest": {
         "ocramius/proxy-manager": "Required for lazy connections"

src/AmqpBundle/DependencyInjection/Configuration.php

@@ -77,6 +77,19 @@ protected function addProducers(ArrayNodeDefinition $node)
                                     ->booleanNode('passive')->defaultFalse()->end()
                                     ->booleanNode('durable')->defaultTrue()->end()
                                     ->booleanNode('auto_delete')->defaultFalse()->end()
+
+                                    // args
+                                    ->arrayNode('arguments')
+                                    ->prototype('scalar')->end()
+                                    ->defaultValue(array())
+                                    ->normalizeKeys(false)
+                                    ->end()
+
+                                    // binding
+                                    ->arrayNode('routing_keys')
+                                    ->prototype('scalar')->end()
+                                    ->defaultValue(array())
+                                    ->end()
                                 ->end()
                             ->end()
                             ->arrayNode('exchange_options')
@@ -148,6 +161,7 @@ protected function addConsumers(ArrayNodeDefinition $node)
                                     ->arrayNode('arguments')
                                         ->prototype('scalar')->end()
                                         ->defaultValue(array())
+                                        ->normalizeKeys(false)
                                     ->end()
 
                                     // binding

src/AmqpBundle/Factory/ProducerFactory.php

@@ -106,13 +106,19 @@ public function get($class, $connexion, array $exchangeOptions, array $queueOpti
             /** @var \AMQPQueue $queue */
             $queue = new $this->queueClass($channel);
             $queue->setName($queueOptions['name']);
+            $queue->setArguments($queueOptions['arguments']);
             $queue->setFlags(
                 ($queueOptions['passive'] ? AMQP_PASSIVE : AMQP_NOPARAM) |
                 ($queueOptions['durable'] ? AMQP_DURABLE : AMQP_NOPARAM) |
                 ($queueOptions['auto_delete'] ? AMQP_AUTODELETE : AMQP_NOPARAM)
             );
             $queue->declareQueue();
             $queue->bind($exchangeOptions['name']);
+
+            // Bind the queue to some routing keys
+            foreach ($queueOptions['routing_keys'] as $routingKey) {
+                $queue->bind($exchangeOptions['name'], $routingKey);
+            }
         }
 
         // Create the producer

src/AmqpBundle/Tests/Fixtures/queue-arguments-config.yml

@@ -0,0 +1,37 @@
+m6_web_amqp:
+    connections:
+        default:
+            host:     'localhost'
+            port:     5672
+            timeout:  2
+            login:    'guest'
+            password: 'guest'
+            vhost:    '/'
+            lazy:     true
+    producers:
+        producer_1:
+            connection: default
+            exchange_options:
+                name: 'exchange_1'
+                type: direct
+                routing_keys: ['super_routing_key']
+        producer_2:
+            connection: default
+            exchange_options:
+                name: 'exchange_2'
+                type: direct
+                routing_keys: ['super_routing_key']
+            queue_options:
+                name: 'ha.queue_exchange_2'
+                arguments:
+                    x-message-ttl: 1000
+    consumers:
+        consumer_1:
+            connection: default
+            exchange_options:
+                name: 'exchange_1'
+            queue_options:
+                name: 'ha.queue_exchange_1'
+                routing_keys: ['super_routing_key']
+                arguments:
+                    x-dead-letter-exchange: 'exchange_2'

src/AmqpBundle/Tests/Units/DependencyInjection/M6WebAmqpExtension.php

@@ -0,0 +1,86 @@
+<?php
+namespace M6Web\Bundle\AmqpBundle\Tests\Units\DependencyInjection;
+
+use M6Web\Bundle\AmqpBundle\DependencyInjection\M6WebAmqpExtension as Base;
+
+use Symfony\Component\DependencyInjection\ParameterBag\ParameterBag;
+use Symfony\Component\DependencyInjection\ContainerBuilder;
+use Symfony\Component\Config\FileLocator;
+use Symfony\Component\DependencyInjection\Loader\YamlFileLoader;
+
+use atoum\test;
+
+/**
+ * Class M6WebAmqpExtension
+ *
+ * @package M6Web\Bundle\AmqpBundle\Tests\Units\DependencyInjection
+ */
+class M6WebAmqpExtension extends test
+{
+    protected function getContainerForConfiguration($fixtureName)
+    {
+        $extension = new Base();
+
+        $parameterBag = new ParameterBag(array('kernel.debug' => true));
+        $container = new ContainerBuilder($parameterBag);
+        $container->set('event_dispatcher', new \mock\Symfony\Component\EventDispatcher\EventDispatcherInterface());
+        $container->registerExtension($extension);
+
+        $loader = new YamlFileLoader($container, new FileLocator(__DIR__.'/../../Fixtures/'));
+        $loader->load($fixtureName.'.yml');
+
+        return $container;
+    }
+
+    public function testQueueArgumentsConfig()
+    {
+        $container = $this->getContainerForConfiguration('queue-arguments-config');
+        $container->compile();
+
+        // test producer queue options
+        $this
+            ->boolean($container->has('m6_web_amqp.producer.producer_2'))
+                ->isTrue()
+            ->array($queueOptions = $container->getDefinition('m6_web_amqp.producer.producer_2')->getArgument(3))
+                ->hasSize(6)
+            ->string($queueOptions['name'])
+                ->isEqualTo('ha.queue_exchange_2')
+            ->array($arguments = $queueOptions['arguments'])
+                ->hasSize(1)
+                ->hasKey('x-message-ttl')
+                ->contains(1000)
+            ->boolean($queueOptions['passive'])
+                ->isFalse()
+            ->boolean($queueOptions['durable'])
+                ->isTrue()
+            ->boolean($queueOptions['auto_delete'])
+                ->isFalse()
+            ->array($queueOptions['routing_keys'])
+                ->isEmpty()
+        ;
+
+        // test consumer queue options
+        $this
+            ->boolean($container->has('m6_web_amqp.consumer.consumer_1'))
+                ->isTrue()
+            ->array($queueOptions = $container->getDefinition('m6_web_amqp.consumer.consumer_1')->getArgument(3))
+                ->hasSize(7)
+            ->string($queueOptions['name'])
+                ->isEqualTo('ha.queue_exchange_1')
+            ->array($arguments = $queueOptions['arguments'])
+                ->hasSize(1)
+                ->hasKey('x-dead-letter-exchange')
+                ->contains('exchange_2')
+            ->boolean($queueOptions['passive'])
+                ->isFalse()
+            ->boolean($queueOptions['durable'])
+                ->isTrue()
+            ->boolean($queueOptions['auto_delete'])
+                ->isFalse()
+            ->array($queueOptions['routing_keys'])
+                ->hasSize(1)
+                ->contains('super_routing_key')
+        ;
+    }
+
+}

src/AmqpBundle/Tests/Units/Factory/ProducerFactory.php

@@ -82,6 +82,8 @@ public function testFactory()
                 'passive' => false,
                 'durable' => true,
                 'auto_delete' => false,
+                'arguments'   => [],
+                'routing_keys' => []
             ])
             ->and($factory = new Base($channelClass, $exchangeClass, $queueClass))
                 ->object($factory->get($producerClass, $connexion, $exchangeOptions, $queueOptions))