Add notify to rpc client. To enable progress functionality. (#351)

* Add notify to rpc client. To enable progress capabilities

* Typo

* Test exception throwing

Author: fon60

RabbitMq/RpcClient.php

@@ -10,6 +10,7 @@ class RpcClient extends BaseAmqp
     protected $replies = array();
     protected $expectSerializedResponse;
     protected $timeout = 0;
+    protected $notifyCallback;
 
     private $queueName;
     private $unserializer = 'unserialize';
@@ -80,6 +81,9 @@ public function processMessage(AMQPMessage $msg)
         if ($this->expectSerializedResponse) {
             $messageBody = call_user_func($this->unserializer, $messageBody);
         }
+        if ($this->notifyCallback !== null) {
+            call_user_func($this->notifyCallback, $messageBody);
+        }
 
         $this->replies[$msg->get('correlation_id')] = $messageBody;
     }
@@ -98,6 +102,15 @@ public function setUnserializer($unserializer)
         $this->unserializer = $unserializer;
     }
 
+    public function notify($callback)
+    {
+        if (is_callable($callback)) {
+            $this->notifyCallback = $callback;
+        } else {
+            throw new \InvalidArgumentException('First parameter expects to be callable');
+        }
+    }
+
     public function setDirectReplyTo($directReplyTo)
     {
         $this->directReplyTo = $directReplyTo;

Tests/RabbitMq/RpcClientTest.php

@@ -22,4 +22,38 @@ public function testProcessMessageWithCustomUnserializer()
         });
         $client->processMessage($message);
     }
+    
+    public function testProcessMessageWithNotifyMethod()
+    {
+        /** @var RpcClient $client */
+        $client = $this->getMockBuilder('\OldSound\RabbitMqBundle\RabbitMq\RpcClient')
+            ->setMethods(array('sendReply', 'maybeStopConsumer'))
+            ->disableOriginalConstructor()
+            ->getMock();
+        $expectedNotify = 'message';
+        $message = $this->getMock('\PhpAmqpLib\Message\AMQPMessage', array('get'), array($expectedNotify));
+        $notified = false;
+        $client->notify(function ($message) use (&$notified) {
+            $notified = $message;
+        });
+
+        $client->initClient(false);
+        $client->processMessage($message);
+
+        $this->assertSame($expectedNotify, $notified);
+    }
+
+    /**
+     * @expectedException \InvalidArgumentException
+     */
+    public function testInvalidParameterOnNotify()
+    {
+        /** @var RpcClient $client */
+        $client = $this->getMockBuilder('\OldSound\RabbitMqBundle\RabbitMq\RpcClient')
+            ->setMethods(array('sendReply', 'maybeStopConsumer'))
+            ->disableOriginalConstructor()
+            ->getMock();
+
+        $client->notify('not a callable');
+    }
 }