Merge remote-tracking branch 'upstream/master'

Conflicts:
	README.md
	RabbitMq/RpcClient.php

Author: steveYeah

README.md

@@ -393,13 +393,7 @@ public function indexAction($name)
 {
     $client = $this->get('old_sound_rabbit_mq.integer_store_rpc');
     $client->addRequest(serialize(array('min' => 0, 'max' => 10)), 'random_int', 'request_id');
-    $timeout = 5; // seconds
-    try {
-        $replies = $client->getReplies($timeout);
-        // process $replies['request_id'];
-    } catch (\PhpAmqpLib\Exception\AMQPTimeoutException $e) {
-        // handle timeout
-    }
+    $replies = $client->getReplies();
 }
 ```
 
@@ -421,7 +415,24 @@ rpc_clients:
         connection: default
         expect_serialized_response: false
 ```
-        
+
+You can also set a expiration for request in seconds, after which message will no longer be handled by server and client request will simply time out. Setting expiration for messages works only for RabbitMQ 3.x and above. Visit http://www.rabbitmq.com/ttl.html#per-message-ttl for more information.
+
+```php
+public function indexAction($name)
+{
+    $expiration = 5; // seconds
+    $client = $this->get('old_sound_rabbit_mq.integer_store_rpc');
+    $client->addRequest($body, $server, $requestId, $expiration);
+    try {
+        $replies = $client->getReplies();
+        // process $replies['request_id'];
+    } catch (\PhpAmqpLib\Exception\AMQPTimeoutException $e) {
+        // handle timeout
+    }
+}
+```
+
 As you can guess, we can also make __parallel RPC calls__.
 
 ### Parallel RPC ###

RabbitMq/RpcClient.php

@@ -11,39 +11,47 @@ class RpcClient extends BaseAmqp
     protected $replies = array();
     protected $queueName;
     protected $expectSerializedResponse;
+    protected $timeout = 0;
 
     public function initClient($expectSerializedResponse = true)
     {
         list($this->queueName, ,) = $this->getChannel()->queue_declare("", false, false, true, true);
         $this->expectSerializedResponse = $expectSerializedResponse;
     }
 
-    public function addRequest($msgBody, $server, $requestId = null, $routingKey = '')
+    public function addRequest($msgBody, $server, $requestId = null, $routingKey = '', $expiration = 0)
     {
         if (empty($requestId)) {
             throw new \InvalidArgumentException('You must provide a $requestId');
         }
 
         $msg = new AMQPMessage($msgBody, array('content_type' => 'text/plain',
                                                'reply_to' => $this->queueName,
+                                               'delivery_mode' => 1, // non durable
+                                               'expiration' => $expiration*1000,
                                                'correlation_id' => $requestId));
 
         $this->getChannel()->basic_publish($msg, $server, $routingKey);
 
         $this->requests++;
+
+        if ($expiration > $this->timeout) {
+            $this->timeout = $expiration;
+        }
     }
 
-    public function getReplies($timeout = 0)
+    public function getReplies()
     {
         $this->replies = array();
         $this->getChannel()->basic_consume($this->queueName, '', false, true, false, false, array($this, 'processMessage'));
 
         while (count($this->replies) < $this->requests) {
-            $this->getChannel()->wait(null, false, $timeout);
+            $this->getChannel()->wait(null, false, $this->timeout);
         }
 
         $this->getChannel()->basic_cancel($this->queueName);
         $this->requests = 0;
+        $this->timeout = 0;
 
         return $this->replies;
     }