Merge pull request #176 from papoola/patch

timeout migrated from getReplies() to addRequest() as expiration #2

Author: videlalvaro

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();
 }
 ```
 
@@ -413,6 +407,23 @@ The arguments we are sending are the __min__ and __max__ values for the `rand()`
 
 The final piece is to get the reply. Our PHP script will block till the server returns a value. The __$replies__ variable will be an associative array where each reply from the server will contained in the respective __request\_id__ key.
 
+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

@@ -10,38 +10,46 @@ class RpcClient extends BaseAmqp
     protected $requests = 0;
     protected $replies = array();
     protected $queueName;
+    protected $timeout = 0;
 
     public function initClient()
     {
         list($this->queueName, ,) = $this->getChannel()->queue_declare("", false, false, true, true);
     }
 
-    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;
     }