PDO based queues: Fix excessively low poll interval, and make it conf…

…igurable
AllenJB · Oct 20, 2021 · a524cdc · a524cdc
1 parent 96d6ba3
commit a524cdc
Show file tree

Hide file tree

Showing 6 changed files with 8 additions and 8 deletions.
diff --git a/src/Pdo/Queue.php b/src/Pdo/Queue.php
@@ -75,7 +75,7 @@ public function publish(QueueMessage $message, \DateTimeImmutable $delayTo = nul
     }
 
 
-    public function consume(callable $callback, float $timeoutSecs): void
+    public function consume(callable $callback, float $timeoutSecs, float $pollIntervalSecs = 0.5): void
     {
         $tsLimit = microtime(true) + $timeoutSecs;
         $waitingReplies = 0;
@@ -125,7 +125,7 @@ public function consume(callable $callback, float $timeoutSecs): void
                 continue;
             }
 
-            usleep(10);
+            usleep($pollIntervalSecs * 1000000);
         }
     }
 

diff --git a/src/Pdo/RPCQueue.php b/src/Pdo/RPCQueue.php
@@ -33,7 +33,7 @@ public function publish(QueueMessage $message): PromiseInterface
     }
 
 
-    public function consume(callable $callback, float $timeoutSecs): void
+    public function consume(callable $callback, float $timeoutSecs, float $pollIntervalSecs = 0.1): void
     {
         $this->replyQueue->consume($callback, $timeoutSecs);
     }

diff --git a/src/Pdo/ReplyQueue.php b/src/Pdo/ReplyQueue.php
@@ -110,7 +110,7 @@ public function getName(): string
     }
 
 
-    public function consume(callable $callback, float $timeoutSecs): void
+    public function consume(callable $callback, float $timeoutSecs, float $pollIntervalSecs = 0.1): void
     {
         $tsLimit = microtime(true) + $timeoutSecs;
         $waitingReplies = 0;
@@ -169,7 +169,7 @@ public function consume(callable $callback, float $timeoutSecs): void
                 continue;
             }
 
-            usleep(10);
+            usleep($pollIntervalSecs * 1000000);
         }
     }
 

diff --git a/src/QueueInterface.php b/src/QueueInterface.php
@@ -14,7 +14,7 @@ public function publish(QueueMessage $message, \DateTimeImmutable $delayTo = nul
     /**
      * @param callable $callback Callback with parameters: QueueInterface $this, QueueMessage
      */
-    public function consume(callable $callback, float $timeoutSecs): void;
+    public function consume(callable $callback, float $timeoutSecs, float $pollIntervalSecs): void;
 
 
     public function ack(QueueMessage $message): void;

diff --git a/src/RPCQueueInterface.php b/src/RPCQueueInterface.php
@@ -14,7 +14,7 @@ public function publish(QueueMessage $message): PromiseInterface;
     /**
      * @param callable $callback Callback with parameters: RPCQueueInterface $this, QueueMessage
      */
-    public function consume(callable $callback, float $timeoutSecs): void;
+    public function consume(callable $callback, float $timeoutSecs, float $pollIntervalSecs): void;
 
 
     public function setExpectedResponseCount(?int $count): void;

diff --git a/src/Rabbit/Queue.php b/src/Rabbit/Queue.php
@@ -80,7 +80,7 @@ public function publish(QueueMessage $message, \DateTimeImmutable $delayTo = nul
     }
 
 
-    public function consume(callable $callback, float $timeoutSecs): void
+    public function consume(callable $callback, float $timeoutSecs, float $pollIntervalSecs = 0.1): void
     {
         throw new UnsupportedOperationException("Consumption is not supported by RabbitQueue");
     }