Added direct reply-to clients support

Author: danut007ro

DependencyInjection/Configuration.php

@@ -218,6 +218,7 @@ protected function addRpcClients(ArrayNodeDefinition $node)
                             ->scalarNode('connection')->defaultValue('default')->end()
                             ->booleanNode('expect_serialized_response')->defaultTrue()->end()
                             ->scalarNode('unserializer')->defaultValue('unserialize')->end()
+                            ->booleanNode('direct_reply_to')->defaultFalse()->end()
                         ->end()
                     ->end()
                 ->end()

DependencyInjection/OldSoundRabbitMqExtension.php

@@ -373,6 +373,9 @@ protected function loadRpcClients()
             if (array_key_exists('unserializer', $client)) {
                 $definition->addMethodCall('setUnserializer', array($client['unserializer']));
             }
+            if (array_key_exists('direct_reply_to', $client)) {
+                $definition->addMethodCall('setDirectReplyTo', array($client['direct_reply_to']));
+            }
 
             $this->container->setDefinition(sprintf('old_sound_rabbit_mq.%s_rpc', $key), $definition);
         }

README.md

@@ -376,6 +376,7 @@ rpc_clients:
     integer_store:
         connection: default
         unserializer: json_decode
+        direct_reply_to: false
 rpc_servers:
     random_int:
         connection: default
@@ -476,6 +477,12 @@ public function indexAction($name)
 
 Is very similar to the previous example, we just have an extra `addRequest` call. Also we provide meaningful request identifiers so later will be easier for us to find the reply we want in the __$replies__ array.
 
+### Direct Reply-To clients ###
+
+To enable [direct reply-to clients](https://www.rabbitmq.com/direct-reply-to.html) you just have to enable option __direct_reply_to__ on the __rpc_clients__ configuration for the client.
+
+This option will use pseudo-queue __amq.rabbitmq.reply-to__ when doing RPC calls. On the RPC server there is no modification needed.
+
 ### Multiple Consumers ###
 
 It's a good practice to have a lot of queues for logic separation. With a simple consumer you will have to create one worker (consumer) per queue and it can be hard to manage when dealing

RabbitMq/RpcClient.php

@@ -14,6 +14,8 @@ class RpcClient extends BaseAmqp
 
     private $queueName;
     private $unserializer = 'unserialize';
+    private $directReplyTo;
+    private $directConsumerTag;
 
     public function initClient($expectSerializedResponse = true)
     {
@@ -26,8 +28,20 @@ public function addRequest($msgBody, $server, $requestId = null, $routingKey = '
             throw new \InvalidArgumentException('You must provide a $requestId');
         }
 
+        if (0 == $this->requests) {
+            // On first addRequest() call, clear all replies
+            $this->replies = array();
+
+            if ($this->directReplyTo) {
+                // On direct reply-to mode, make initial consume call
+                $this->directConsumerTag = $this->getChannel()->basic_consume('amq.rabbitmq.reply-to', '', false, true, false, false, array($this, 'processMessage'));
+            }
+        }
+
         $msg = new AMQPMessage($msgBody, array('content_type' => 'text/plain',
-                                               'reply_to' => $this->getQueueName(),
+                                               'reply_to' => $this->directReplyTo
+                                                   ? 'amq.rabbitmq.reply-to' // On direct reply-to mode, use predefined queue name
+                                                   : $this->getQueueName(),
                                                'delivery_mode' => 1, // non durable
                                                'expiration' => $expiration*1000,
                                                'correlation_id' => $requestId));
@@ -43,14 +57,18 @@ public function addRequest($msgBody, $server, $requestId = null, $routingKey = '
 
     public function getReplies()
     {
-        $this->replies = array();
-        $consumer_tag = $this->getChannel()->basic_consume($this->getQueueName(), '', false, true, false, false, array($this, 'processMessage'));
+        if ($this->directReplyTo) {
+            $consumer_tag = $this->directConsumerTag;
+        } else {
+            $consumer_tag = $this->getChannel()->basic_consume($this->getQueueName(), '', false, true, false, false, array($this, 'processMessage'));
+        }
 
         while (count($this->replies) < $this->requests) {
             $this->getChannel()->wait(null, false, $this->timeout);
         }
 
         $this->getChannel()->basic_cancel($consumer_tag);
+        $this->directConsumerTag = null;
         $this->requests = 0;
         $this->timeout = 0;
 
@@ -80,4 +98,9 @@ public function setUnserializer($unserializer)
     {
         $this->unserializer = $unserializer;
     }
+
+    public function setDirectReplyTo($directReplyTo)
+    {
+        $this->directReplyTo = $directReplyTo;
+    }
 }

Tests/DependencyInjection/Fixtures/test.yml

@@ -164,6 +164,7 @@ old_sound_rabbit_mq:
         foo_client:
             connection:      foo_connection
             unserializer:    json_decode
+            direct_reply_to: true
 
         default_client:
 

Tests/DependencyInjection/OldSoundRabbitMqExtensionTest.php

@@ -518,7 +518,8 @@ public function testFooRpcClientDefinition()
         $this->assertEquals(
             array(
                 array('initClient', array(true)),
-                array('setUnserializer', array('json_decode'))
+                array('setUnserializer', array('json_decode')),
+                array('setDirectReplyTo', array(true)),
             ),
             $definition->getMethodCalls()
         );
@@ -536,7 +537,8 @@ public function testDefaultRpcClientDefinition()
         $this->assertEquals(
             array(
                 array('initClient', array(true)),
-                array('setUnserializer', array('unserialize'))
+                array('setUnserializer', array('unserialize')),
+                array('setDirectReplyTo', array(false)),
             ),
             $definition->getMethodCalls()
         );