PDO based queues: Fix timeout not being a hard limit (as it is with R…

…abbitMQ based queues)
AllenJB · Dec 18, 2020 · 2c0d109 · 2c0d109
1 parent e49c568
commit 2c0d109
Show file tree

Hide file tree

Showing 2 changed files with 12 additions and 10 deletions.
diff --git a/src/Pdo/Queue.php b/src/Pdo/Queue.php
@@ -111,17 +111,18 @@ public function consume(callable $callback, float $timeoutSecs): void
                 $callback($this, $message);
             }
 
-            // Always process all the waiting responses
+            if (microtime(true) > $tsLimit) {
+                break;
+            }
+
+            // If more queue items are pending, loop immediately
             if ($waitingReplies < 1) {
                 $waitingReplies = $this->getMessageCount();
             }
             if ($waitingReplies > 0) {
                 continue;
             }
 
-            if (microtime(true) > $tsLimit) {
-                break;
-            }
             usleep(10);
         }
     }
@@ -158,7 +159,7 @@ public function nack(QueueMessage $message): void
      */
     public function getMessageCount(): ?int
     {
-        $sql = "SELECT COUNT(queueid) AS `c` 
+        $sql = "SELECT COUNT(queueid) AS `c`
             FROM `q_{$this->name}`
             WHERE `locked` = 0
                 AND (`dt_scheduled` IS NULL OR `dt_scheduled` < NOW())";
@@ -172,7 +173,7 @@ public function getMessageCount(): ?int
      */
     public function getTotalMessageCount(): ?int
     {
-        $sql = "SELECT COUNT(queueid) AS `c` 
+        $sql = "SELECT COUNT(queueid) AS `c`
             FROM `q_{$this->name}`
             WHERE `locked` = 0";
         $stmt = $this->pdo->query($sql);

diff --git a/src/Pdo/ReplyQueue.php b/src/Pdo/ReplyQueue.php
@@ -155,17 +155,18 @@ public function consume(callable $callback, float $timeoutSecs): void
                 }
             }
 
-            // Always process all the waiting responses
+            if (microtime(true) > $tsLimit) {
+                break;
+            }
+
+            // If more queue items are pending, loop immediately
             if ($waitingReplies < 1) {
                 $waitingReplies = $this->getMessageCount();
             }
             if ($waitingReplies > 0) {
                 continue;
             }
 
-            if (microtime(true) > $tsLimit) {
-                break;
-            }
             usleep(10);
         }
     }