AMQP Key Routing

Command/AMQPRemoteCommand.php

@@ -18,6 +18,7 @@ class AMQPRemoteCommand extends ContainerAwareCommand
             ->setHelp('Run a command on an AMQP Consumer')
             ->addArgument('name', InputOption::VALUE_REQUIRED, 'Command to execute')
             ->addOption('args', null, InputOption::VALUE_OPTIONAL|InputOption::VALUE_IS_ARRAY, 'Optional Commands arguments. e.g. --args=key:value --args=key1:value1')
+            ->addOption('route', null, InputOption::VALUE_OPTIONAL, 'AMQP Route Key', '')
         ;
     }
 
@@ -35,7 +36,8 @@ class AMQPRemoteCommand extends ContainerAwareCommand
                 'args' => $args,
             );
             $producer = $this->getContainer()->get('old_sound_rabbit_mq.command_producer_producer');
-            $producer->publish(serialize(compact('name', 'args')));
+	    $route = $input->getOption('route');
+            $producer->publish(serialize(compact('name', 'args')), $route);
             
             $output->writeln('<info>Command executed!</info>');
         } else {

Resources/config/rabbit_mq/config.yml

@@ -22,23 +22,23 @@ old_sound_rabbit_mq:
     producers:
         flowdat_tasklogger:
             connection:       default
-            exchange_options: {name: 'send', type: direct}
+            exchange_options: {name: 'send', type: topic}
             service_alias:    fd_tasklogger_service # no alias by default
         
         command_producer:
             connection:       default
-            exchange_options: {name: 'command', type: direct}
+            exchange_options: {name: 'command', type: topic}
             service_alias:    command_producer_service # no alias by default
             
     consumers:
         flowdat_tasklogger:
             connection:       default
-            exchange_options: {name: 'send', type: direct}
+            exchange_options: {name: 'send', type: topic}
             queue_options:    {name: 'send'}
             callback:         flowdat_tasklogger_service # sf service id
             
         command_consumer:
             connection:       default
-            exchange_options: {name: 'command', type: direct}
-            queue_options:    {name: 'command'}
+            exchange_options: {name: 'command', type: topic}
+            queue_options:    {name: 'command', routing_keys : ["%env(AMQP_KEY)%"]}
             callback:         command_consumer_service # sf service id

Services/CommandConsumer.php

@@ -34,7 +34,7 @@ class CommandConsumer
      */
     public function execute(AMQPMessage $msg)
     {
-        $msgBody = unserialize($msg->getBody());
+        $msgBody = unserialize($msg->body);
         if (isset($msgBody['name']) && isset($msgBody['args'])) {
             // command name and args
             $name = $msgBody['name'];