[amqp] fix signal handler if consume called from consume

makasim · makasim · commit eeb6967b5e90 · 2018-01-12T16:43:59.000+02:00
diff --git a/pkg/amqp-bunny/AmqpContext.php b/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();
         }
     }
 
diff --git a/pkg/amqp-lib/AmqpContext.php b/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();
         }
     }
 

Original file line number	Diff line number	Diff line change
`@@ -55,11 +55,6 @@ class AmqpContext implements InteropAmqpContext, DelayStrategyAware`
`55`	`55`	`*/`
`56`	`56`	`private $subscribers;`
`57`	`57`
`58`		`- /**`
`59`		`- * @var SignalSocketHelper`
`60`		`- */`
`61`		`- private $signalSocketHandler;`
`62`		`-`
`63`	`58`	`/**`
`64`	`59`	`* Callable must return instance of \Bunny\Channel once called.`
`65`	`60`	`*`
`@@ -85,7 +80,6 @@ public function __construct($bunnyChannel, $config = [])`
`85`	`80`
`86`	`81`	`$this->buffer = new Buffer();`
`87`	`82`	`$this->subscribers = [];`
`88`		`- $this->signalSocketHandler = new SignalSocketHelper();`
`89`	`83`	`}`
`90`	`84`
`91`	`85`	`/**`
`@@ -396,18 +390,19 @@ public function consume($timeout = 0)`
`396`	`390`	`throw new \LogicException('There is no subscribers. Consider calling basicConsumeSubscribe before consuming');`
`397`	`391`	`}`
`398`	`392`
`399`		`- $this->signalSocketHandler->beforeSocket();`
	`393`	`+ $signalHandler = new SignalSocketHelper();`
	`394`	`+ $signalHandler->beforeSocket();`
`400`	`395`
`401`	`396`	`try {`
`402`	`397`	`$this->getBunnyChannel()->getClient()->run(0 !== $timeout ? $timeout / 1000 : null);`
`403`	`398`	`} catch (ClientException $e) {`
`404`		`- if ('stream_select() failed.' == $e->getMessage() && $this->signalSocketHandler->wasThereSignal()) {`
	`399`	`+ if ('stream_select() failed.' == $e->getMessage() && $signalHandler->wasThereSignal()) {`
`405`	`400`	`return;`
`406`	`401`	`}`
`407`	`402`
`408`	`403`	`throw $e;`
`409`	`404`	`} finally {`
`410`		`- $this->signalSocketHandler->afterSocket();`
	`405`	`+ $signalHandler->afterSocket();`
`411`	`406`	`}`
`412`	`407`	`}`
`413`	`408`