Merge pull request #328 from php-enqueue/amqp-fix

[amqp] fix signal handler if consume called from consume

Author: makasim

pkg/amqp-bunny/AmqpContext.php

@@ -55,11 +55,6 @@ class AmqpContext implements InteropAmqpContext, DelayStrategyAware
      */
     private $subscribers;
 
-    /**
-     * @var SignalSocketHelper
-     */
-    private $signalSocketHandler;
-
     /**
      * Callable must return instance of \Bunny\Channel once called.
      *
@@ -85,7 +80,6 @@ public function __construct($bunnyChannel, $config = [])
 
         $this->buffer = new Buffer();
         $this->subscribers = [];
-        $this->signalSocketHandler = new SignalSocketHelper();
     }
 
     /**
@@ -396,18 +390,19 @@ public function consume($timeout = 0)
             throw new \LogicException('There is no subscribers. Consider calling basicConsumeSubscribe before consuming');
         }
 
-        $this->signalSocketHandler->beforeSocket();
+        $signalHandler = new SignalSocketHelper();
+        $signalHandler->beforeSocket();
 
         try {
             $this->getBunnyChannel()->getClient()->run(0 !== $timeout ? $timeout / 1000 : null);
         } catch (ClientException $e) {
-            if ('stream_select() failed.' == $e->getMessage() && $this->signalSocketHandler->wasThereSignal()) {
+            if ('stream_select() failed.' == $e->getMessage() && $signalHandler->wasThereSignal()) {
                 return;
             }
 
             throw $e;
         } finally {
-            $this->signalSocketHandler->afterSocket();
+            $signalHandler->afterSocket();
         }
     }
 

pkg/amqp-lib/AmqpContext.php

@@ -57,11 +57,6 @@ class AmqpContext implements InteropAmqpContext, DelayStrategyAware
      */
     private $subscribers;
 
-    /**
-     * @var SignalSocketHelper
-     */
-    private $signalSocketHandler;
-
     /**
      * @param AbstractConnection $connection
      * @param array              $config
@@ -78,7 +73,6 @@ public function __construct(AbstractConnection $connection, $config = [])
         $this->connection = $connection;
         $this->buffer = new Buffer();
         $this->subscribers = [];
-        $this->signalSocketHandler = new SignalSocketHelper();
     }
 
     /**
@@ -390,7 +384,8 @@ public function consume($timeout = 0)
             throw new \LogicException('There is no subscribers. Consider calling basicConsumeSubscribe before consuming');
         }
 
-        $this->signalSocketHandler->beforeSocket();
+        $signalHandler = new SignalSocketHelper();
+        $signalHandler->beforeSocket();
 
         try {
             while (true) {
@@ -413,13 +408,13 @@ public function consume($timeout = 0)
         } catch (AMQPTimeoutException $e) {
         } catch (StopBasicConsumptionException $e) {
         } catch (AMQPIOWaitException $e) {
-            if ($this->signalSocketHandler->wasThereSignal()) {
+            if ($signalHandler->wasThereSignal()) {
                 return;
             }
 
             throw $e;
         } finally {
-            $this->signalSocketHandler->afterSocket();
+            $signalHandler->afterSocket();
         }
     }