fix 100% CPU usage and stop-when-empty flag of rabbitmq:consume command

Author: vyuldashev

CHANGELOG.md

@@ -2,6 +2,11 @@
 
 All notable changes to this project will be documented in this file.
 
+## [10.1.3 (2020-01-12)](https://github.com/vyuldashev/laravel-queue-rabbitmq/compare/v10.1.2...v10.1.3)
+
+- Fix 100% CPU usage of `rabbitmq:consume` command by adding sleep to consumer when no messages are got from the queue.
+- Fix `stop-on-empty` flag for `rabbitmq:consume` command.
+
 ## [10.1.2 (2019-12-24)](https://github.com/vyuldashev/laravel-queue-rabbitmq/compare/v10.1.1...v10.1.2)
 
 - Fix `rabbitmq:queue-bind` command. [#294](https://github.com/vyuldashev/laravel-queue-rabbitmq/pull/294)

src/Consumer.php

@@ -32,6 +32,9 @@ class Consumer extends Worker
     /** @var AMQPChannel */
     protected $channel;
 
+    /** @var boolean */
+    protected $gotJob = false;
+
     public function setContainer(Container $value): void
     {
         $this->container = $value;
@@ -79,6 +82,8 @@ public function daemon($connectionName, $queue, WorkerOptions $options): void
             false,
             false,
             function (AMQPMessage $message) use ($connection, $options, $connectionName, $queue): void {
+                $this->gotJob = true;
+
                 $job = new RabbitMQJob(
                     $this->container,
                     $connection,
@@ -104,9 +109,7 @@ function (AMQPMessage $message) use ($connection, $options, $connectionName, $qu
                 continue;
             }
 
-            // If the daemon should run (not in maintenance mode, etc.), then we can run
-            // fire off this job for processing. Otherwise, we will need to sleep the
-            // worker so no more jobs are processed until they should be processed.
+            // If the daemon should run (not in maintenance mode, etc.), then we can wait for a job.
             try {
                 $this->channel->wait(null, true, (int) $options->timeout);
             } catch (AMQPRuntimeException $exception) {
@@ -123,10 +126,17 @@ function (AMQPMessage $message) use ($connection, $options, $connectionName, $qu
                 $this->stopWorkerIfLostConnection($exception);
             }
 
+            // If no job is got off the queue, we will need to sleep the worker.
+            if (! $this->gotJob) {
+                $this->sleep($options->sleep);
+            }
+
             // Finally, we will check to see if we have exceeded our memory limits or if
             // the queue should restart based on other indications. If so, we'll stop
             // this worker and let whatever is "monitoring" it restart the process.
-            $this->stopIfNecessary($options, $lastRestart, null);
+            $this->stopIfNecessary($options, $lastRestart, $this->gotJob ? true : null);
+
+            $this->gotJob = false;
         }
     }