Close RPC client channel on exceptions (#31)

Co-authored-by: Hafsah Haynes <hafsah@surfsharkbiz.com>

Author: notrix

RabbitMq/RpcClient.php

@@ -63,11 +63,14 @@ public function getReplies()
             $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);
+        try {
+            while (count($this->replies) < $this->requests) {
+                $this->getChannel()->wait(null, false, $this->timeout);
+            }
+        } finally {
+            $this->getChannel()->basic_cancel($consumer_tag);
         }
 
-        $this->getChannel()->basic_cancel($consumer_tag);
         $this->directConsumerTag = null;
         $this->requests = 0;
         $this->timeout = 0;
@@ -87,7 +90,7 @@ public function processMessage(AMQPMessage $msg)
 
         $this->replies[$msg->get('correlation_id')] = $messageBody;
     }
-    
+
     protected function getQueueName()
     {
         if (null === $this->queueName) {

Tests/RabbitMq/RpcClientTest.php

@@ -3,6 +3,7 @@
 namespace OldSound\RabbitMqBundle\Tests\RabbitMq;
 
 use OldSound\RabbitMqBundle\RabbitMq\RpcClient;
+use PhpAmqpLib\Exception\AMQPTimeoutException;
 use PhpAmqpLib\Message\AMQPMessage;
 use PHPUnit\Framework\TestCase;
 
@@ -55,9 +56,6 @@ public function testProcessMessageWithNotifyMethod()
         $this->assertSame($expectedNotify, $notified);
     }
 
-    /**
-     * @expectedException \InvalidArgumentException
-     */
     public function testInvalidParameterOnNotify()
     {
         /** @var RpcClient $client */
@@ -66,6 +64,34 @@ public function testInvalidParameterOnNotify()
             ->disableOriginalConstructor()
             ->getMock();
 
+        $this->expectException('\InvalidArgumentException');
+
         $client->notify('not a callable');
     }
+
+    public function testChannelCancelOnGetRepliesException()
+    {
+        $client = $this->getMockBuilder('\OldSound\RabbitMqBundle\RabbitMq\RpcClient')
+            ->setMethods(null)
+            ->disableOriginalConstructor()
+            ->getMock();
+
+        $channel = $this->createMock('\PhpAmqpLib\Channel\AMQPChannel');
+        $channel->expects($this->any())
+            ->method('getChannelId')
+            ->willReturn('test');
+        $channel->expects($this->once())
+            ->method('wait')
+            ->willThrowException(new AMQPTimeoutException());
+
+        $this->expectException('\PhpAmqpLib\Exception\AMQPTimeoutException');
+
+        $channel->expects($this->once())
+            ->method('basic_cancel');
+
+        $client->setChannel($channel);
+        $client->addRequest('a', 'b', 'c');
+
+        $client->getReplies();
+    }
 }