prevent process to loop when stopConsuming is called

Author: Nicola Turato

RabbitMq/BatchConsumer.php

@@ -9,11 +9,6 @@
 
 final class BatchConsumer extends BaseAmqp implements DequeuerInterface
 {
-    /**
-     * @var int
-     */
-    private $consumed = 0;
-
     /**
      * @var \Closure|callable
      */
@@ -75,43 +70,32 @@ public function consume()
     {
         $this->setupConsumer();
 
-        $isConsuming = false;
-        $timeoutWanted = $this->getTimeoutWait();
         while (count($this->getChannel()->callbacks)) {
-            $this->maybeStopConsumer();
-            if (!$this->forceStop) {
-                try {
-                    $this->getChannel()->wait(null, false, $timeoutWanted);
-                    $isConsuming = true;
-                } catch (AMQPTimeoutException $e) {
-                    $this->batchConsume();
-                    if ($isConsuming) {
-                        $isConsuming = false;
-                    } elseif (null !== $this->getIdleTimeoutExitCode()) {
-                        return $this->getIdleTimeoutExitCode();
-                    } else {
-                        throw $e;
-                    }
-                }
-            } else {
+            if ($this->isCompleteBatch()) {
                 $this->batchConsume();
             }
 
-            if ($this->isCompleteBatch($isConsuming)) {
-                $this->batchConsume();
-            }
+            $this->maybeStopConsumer();
+
+            $timeout = $this->isEmptyBatch() ? $this->getIdleTimeout() : $this->getTimeoutWait();
 
-            $timeoutWanted = $isConsuming ? $this->getTimeoutWait() : $this->getIdleTimeout();
+            try {
+                $this->getChannel()->wait(null, false, $timeout);
+            } catch (AMQPTimeoutException $e) {
+                if (!$this->isEmptyBatch()) {
+                    $this->batchConsume();
+                } elseif (null !== $this->getIdleTimeoutExitCode()) {
+                    return $this->getIdleTimeoutExitCode();
+                } else {
+                    throw $e;
+                }
+            }
         }
     }
 
-    public function batchConsume()
+    private function batchConsume()
     {
-        if ($this->batchCounter === 0) {
-            return;
-        }
-
-        try  {
+        try {
             $processFlags = call_user_func($this->callback, $this->messages);
             $this->handleProcessMessages($processFlags);
             $this->logger->debug('Queue message processed', array(
@@ -129,6 +113,7 @@ public function batchConsume()
                     'stacktrace' => $e->getTraceAsString()
                 )
             ));
+            $this->resetBatch();
             $this->stopConsuming();
         } catch (\Exception $e) {
             $this->logger->error($e->getMessage(), array(
@@ -166,9 +151,6 @@ protected function handleProcessMessages($processFlags = null)
         foreach ($processFlags as $deliveryTag => $processFlag) {
             $this->handleProcessFlag($deliveryTag, $processFlag);
         }
-
-        $this->consumed++;
-        $this->maybeStopConsumer();
     }
 
     /**
@@ -177,7 +159,7 @@ protected function handleProcessMessages($processFlags = null)
      *
      * @return  void
      */
-    private function handleProcessFlag ($deliveryTag, $processFlag)
+    private function handleProcessFlag($deliveryTag, $processFlag)
     {
         if ($processFlag === ConsumerInterface::MSG_REJECT_REQUEUE || false === $processFlag) {
             // Reject and requeue message to RabbitMQ
@@ -195,13 +177,19 @@ private function handleProcessFlag ($deliveryTag, $processFlag)
     }
 
     /**
-     * @param   bool    $isConsuming
-     *
      * @return  bool
      */
-    protected function isCompleteBatch($isConsuming)
+    protected function isCompleteBatch()
     {
-        return $isConsuming && $this->batchCounter === $this->prefetchCount;
+        return $this->batchCounter === $this->prefetchCount;
+    }
+
+    /**
+     * @return  bool
+     */
+    protected function isEmptyBatch()
+    {
+        return $this->batchCounter === 0;
     }
 
     /**
@@ -300,7 +288,9 @@ private function getMessageChannel($deliveryTag)
      */
     public function stopConsuming()
     {
-        $this->batchConsume();
+        if (!$this->isEmptyBatch()) {
+            $this->batchConsume();
+        }
 
         $this->getChannel()->basic_cancel($this->getConsumerTag());
     }