From 2c0d109ded7517695db373070c7fcefde5662c45 Mon Sep 17 00:00:00 2001
From: AllenJB <github@allenjb.me.uk>
Date: Fri, 18 Dec 2020 10:10:52 +0000
Subject: [PATCH] PDO based queues: Fix timeout not being a hard limit (as it
 is with RabbitMQ based queues)

---
 src/Pdo/Queue.php      | 13 +++++++------
 src/Pdo/ReplyQueue.php |  9 +++++----
 2 files changed, 12 insertions(+), 10 deletions(-)

diff --git a/src/Pdo/Queue.php b/src/Pdo/Queue.php
index 9c0a305..5e8d395 100644
--- a/src/Pdo/Queue.php
+++ b/src/Pdo/Queue.php
@@ -111,7 +111,11 @@ 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();
             }
@@ -119,9 +123,6 @@ public function consume(callable $callback, float $timeoutSecs): void
                 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
index 7ffaf1d..8d899ab 100644
--- a/src/Pdo/ReplyQueue.php
+++ b/src/Pdo/ReplyQueue.php
@@ -155,7 +155,11 @@ 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();
             }
@@ -163,9 +167,6 @@ public function consume(callable $callback, float $timeoutSecs): void
                 continue;
             }
 
-            if (microtime(true) > $tsLimit) {
-                break;
-            }
             usleep(10);
         }
     }