allow queue to bind to multiple exchanges

fixes: #20
prolic · Jul 16, 2016 · 64d5fc6 · 64d5fc6
1 parent cd191d8
commit 64d5fc6
Show file tree

Hide file tree

Showing 7 changed files with 214 additions and 37 deletions.
diff --git a/src/Container/QueueFactory.php b/src/Container/QueueFactory.php
@@ -140,23 +140,37 @@ public function __invoke(ContainerInterface $container) : Queue
 
         if ($this->autoSetupFabric || $options['auto_setup_fabric']) {
             // auto setup fabric depended exchange
-            $exchangeName = $options['exchange'];
-            ExchangeFactory::$exchangeName($container, $this->channel, true);
-
             if (isset($options['arguments']['x-dead-letter-exchange'])) {
                 // auto setup fabric dead letter exchange
                 $exchangeName = $options['arguments']['x-dead-letter-exchange'];
                 ExchangeFactory::$exchangeName($container, $this->channel, true);
             }
 
+            $exchanges = $options['exchanges'];
+
+            if ($exchanges instanceof \Traversable) {
+                $exchanges = iterator_to_array($exchanges);
+            }
+
+            if (! is_array($exchanges) || empty($exchanges)) {
+                throw new Exception\InvalidArgumentException('Expected an array or traversable of exchanges');
+            }
+
+            foreach ($exchanges as $exchange => $exchangeOptions) {
+                ExchangeFactory::$exchange($container, $this->channel, true);
+            }
+
             $queue->declareQueue();
 
-            $routingKeys = $options['routing_keys'];
-            if (empty($routingKeys)) {
-                $queue->bind($options['exchange'], '', $options['bind_arguments']);
-            } else {
-                foreach ($routingKeys as $routingKey) {
-                    $queue->bind($options['exchange'], $routingKey, $options['bind_arguments']);
+            foreach ($exchanges as $exchange => $exchangeOptions) {
+                if (empty($exchangeOptions)) {
+                    $this->bindQueue($queue, $exchange, [], []);
+                } else {
+                    foreach ($exchangeOptions as $exchangeOption) {
+                        $routingKeys = $exchangeOption['routing_keys'] ?? [];
+                        $bindArguments = $exchangeOption['bind_arguments'] ?? [];
+                        $this->bindQueue($queue, $exchange, $routingKeys, $bindArguments);
+                    }
                 }
             }
         }
@@ -184,8 +198,6 @@ public function defaultOptions()
             'exclusive' => false,
             'auto_delete' => false,
             'arguments' => [],
-            'routing_keys' => [],
-            'bind_arguments' => [],
             // factory configs
             'auto_setup_fabric' => false
         ];
@@ -198,7 +210,7 @@ public function mandatoryOptions()
     {
         return [
             'connection',
-            'exchange',
+            'exchanges',
         ];
     }
 
@@ -216,4 +228,21 @@ private function getFlags($options)
 
         return $flags;
     }
+
+    /**
+     * @param Queue $queue
+     * @param string $exchange
+     * @param array $routingKeys
+     * @param array $bindArguments
+     */
+    private function bindQueue(Queue $queue, string $exchange, array $routingKeys, array $bindArguments)
+    {
+        if (empty($routingKeys)) {
+            $queue->bind($exchange, '', $bindArguments);
+        } else {
+            foreach ($routingKeys as $routingKey) {
+                $queue->bind($exchange, $routingKey, $bindArguments);
+            }
+        }
+    }
 }
diff --git a/tests/Console/Command/PurgeQueueCommandTest.php b/tests/Console/Command/PurgeQueueCommandTest.php
@@ -114,7 +114,9 @@ public function it_purges_the_queue()
                         'queue' => [
                             'foo' => [
                                 'name' => 'foo',
-                                'exchange' => 'demo',
+                                'exchanges' => [
+                                    'demo' => [],
+                                ],
                                 'connection' => 'default',
                             ],
                         ],

diff --git a/tests/Console/Command/SetupFabricCommandTest.php b/tests/Console/Command/SetupFabricCommandTest.php
@@ -77,13 +77,14 @@ public function it_declares_exchanges_and_queues()
                         'queue' => [
                             'foo' => [
                                 'name' => 'foo',
-                                'exchange' => 'demo',
+                                'exchanges' => [
+                                    'demo' => [],
+                                ],
                                 'connection' => 'default',
                             ],
                         ],
                         'connection' => [
-                            'default' => [
-                            ],
+                            'default' => [],
                         ],
                     ]
                 ]

diff --git a/tests/Container/CallbackConsumerFactoryTest.php b/tests/Container/CallbackConsumerFactoryTest.php
@@ -63,7 +63,9 @@ public function it_creates_callback_consumer()
                         'my_queue' => [
                             'connection' => 'my_connection',
                             'name' => 'my_queue',
-                            'exchange' => 'my_exchange',
+                            'exchanges' => [
+                                'my_exchange' => [],
+                            ],
                         ],
                     ],
                     'callback_consumer' => [
@@ -123,7 +125,9 @@ public function it_creates_callback_consumer_with_call_static_and_defined_logger
                         'my_queue' => [
                             'connection' => 'my_connection',
                             'name' => 'my_queue',
-                            'exchange' => 'my_exchange',
+                            'exchanges' => [
+                                'my_exchange' => [],
+                            ],
                         ],
                     ],
                     'callback_consumer' => [

diff --git a/tests/Container/JsonRpcClientFactoryTest.php b/tests/Container/JsonRpcClientFactoryTest.php
@@ -63,7 +63,9 @@ public function it_creates_json_rpc_client()
                         'my_queue' => [
                             'connection' => 'my_connection',
                             'name' => 'my_queue',
-                            'exchange' => 'my_exchange',
+                            'exchanges' => [
+                                'my_exchange' => [],
+                            ],
                         ],
                     ],
                     'json_rpc_client' => [
@@ -129,7 +131,9 @@ public function it_creates_json_rpc_client_with_call_static()
                         'my_queue' => [
                             'connection' => 'my_connection',
                             'name' => 'my_queue',
-                            'exchange' => 'my_exchange',
+                            'exchanges' => [
+                                'my_exchange' => [],
+                            ],
                         ],
                     ],
                     'json_rpc_client' => [
@@ -210,7 +214,9 @@ public function it_throws_exception_when_empty_exchanges_given()
                         'my_queue' => [
                             'connection' => 'my_connection',
                             'name' => 'my_queue',
-                            'exchange' => 'my_exchange',
+                            'exchanges' => [
+                                'my_exchange' => [],
+                            ],
                         ],
                     ],
                     'json_rpc_client' => [

diff --git a/tests/Container/JsonRpcServerFactoryTest.php b/tests/Container/JsonRpcServerFactoryTest.php
@@ -64,7 +64,9 @@ public function it_creates_json_rpc_server()
                         'my_queue' => [
                             'connection' => 'my_connection',
                             'name' => 'my_queue',
-                            'exchange' => 'my_exchange',
+                            'exchanges' => [
+                                'my_exchange' => [],
+                            ],
                         ],
                     ],
                     'json_rpc_server' => [
@@ -133,7 +135,9 @@ public function it_creates_json_rpc_server_with_call_static_and_defined_logger()
                         'my_queue' => [
                             'connection' => 'my_connection',
                             'name' => 'my_queue',
-                            'exchange' => 'my_exchange',
+                            'exchanges' => [
+                                'my_exchange' => [],
+                            ],
                         ],
                     ],
                     'json_rpc_server' => [