[amqp-ext] Restore default read timeout inside consume callback.

Author: makasim

pkg/amqp-ext/AmqpContext.php

@@ -360,17 +360,24 @@ public function consume($timeout = 0)
 
             $extQueue = new \AMQPQueue($this->getExtChannel());
             $extQueue->setName($consumer->getQueue()->getQueueName());
-            $extQueue->consume(function (\AMQPEnvelope $extEnvelope, \AMQPQueue $q) {
-                $message = $this->convertMessage($extEnvelope);
-                $message->setConsumerTag($q->getConsumerTag());
-
-                /**
-                 * @var AmqpConsumer
-                 * @var callable     $callback
-                 */
-                list($consumer, $callback) = $this->subscribers[$q->getConsumerTag()];
-
-                return call_user_func($callback, $message, $consumer);
+            $extQueue->consume(function (\AMQPEnvelope $extEnvelope, \AMQPQueue $q) use ($originalTimeout, $extConnection) {
+                $consumeTimeout = $extConnection->getReadTimeout();
+                try {
+                    $extConnection->setReadTimeout($originalTimeout);
+
+                    $message = $this->convertMessage($extEnvelope);
+                    $message->setConsumerTag($q->getConsumerTag());
+
+                    /**
+                     * @var AmqpConsumer
+                     * @var callable     $callback
+                     */
+                    list($consumer, $callback) = $this->subscribers[$q->getConsumerTag()];
+
+                    return call_user_func($callback, $message, $consumer);
+                } finally {
+                    $extConnection->setReadTimeout($consumeTimeout);
+                }
             }, AMQP_JUST_CONSUME);
         } catch (\AMQPQueueException $e) {
             if ('Consumer timeout exceed' == $e->getMessage()) {