if no queue was consumed, then wait the polling interval. Otherweise continue processing the queues!

Author: Henning

pkg/dbal/DbalConsumerHelperTrait.php

@@ -29,6 +29,18 @@ protected function fetchMessage(array $queues, int $redeliveryDelay): ?DbalMessa
 
         $endAt = microtime(true) + 0.2; // add 200ms
 
+        $select = $this->getConnection()->createQueryBuilder()
+            ->select('id')
+            ->from($this->getContext()->getTableName())
+            ->andWhere('queue IN (:queues)')
+            ->andWhere('delayed_until IS NULL OR delayed_until <= :delayedUntil')
+            ->andWhere('delivery_id IS NULL')
+            ->addOrderBy('priority', 'asc')
+            ->addOrderBy('published_at', 'asc')
+            ->setParameter('queues', $queues, Connection::PARAM_STR_ARRAY)
+            ->setParameter('delayedUntil', $now, DbalType::INTEGER)
+            ->setMaxResults(1);
+
         $update = $this->getConnection()->createQueryBuilder()
             ->update($this->getContext()->getTableName())
             ->set('delivery_id', ':deliveryId')
@@ -41,7 +53,7 @@ protected function fetchMessage(array $queues, int $redeliveryDelay): ?DbalMessa
 
         while (microtime(true) < $endAt) {
             try {
-                $result = $this->getResultByQueueList($queues, $now);
+                $result = $select->execute()->fetch();
                 if (empty($result)) {
                     return null;
                 }
@@ -143,30 +155,4 @@ private function deleteMessage(string $deliveryId): void
             ['delivery_id' => DbalType::GUID]
         );
     }
-
-    private function getResultByQueueList(array $queues, int $now): ?array
-    {
-        $select = $this->getConnection()->createQueryBuilder()
-            ->select('id')
-            ->from($this->getContext()->getTableName())
-            ->andWhere('queue = :queue')
-            ->andWhere('delayed_until IS NULL OR delayed_until <= :delayedUntil')
-            ->andWhere('delivery_id IS NULL')
-            ->addOrderBy('priority', 'asc')
-            ->addOrderBy('published_at', 'asc')
-            ->setParameter('delayedUntil', $now, DbalType::INTEGER)
-            ->setMaxResults(1);
-
-        foreach ($queues as $queue) {
-            $select->setParameter('queue', $queue, DbalType::STRING);
-
-            $result = $select->execute()->fetch();
-
-            if (!empty($result)) {
-                return $result;
-            }
-        }
-
-        return null;
-    }
 }

pkg/dbal/DbalSubscriptionConsumer.php

@@ -87,21 +87,29 @@ public function consume(int $timeout = 0): void
 
         $queueNames = [];
         foreach (array_keys($this->subscribers) as $queueName) {
-            $queueNames[] = $queueName;
+            $queueNames[$queueName] = $queueName;
         }
-        $queueNames = array_unique($queueNames);
 
         $timeout /= 1000;
         $now = time();
         $redeliveryDelay = $this->getRedeliveryDelay() / 1000; // milliseconds to seconds
 
+        $currentQueueNames = [];
+        $queueConsumed = false;
         while (true) {
+            if (empty($currentQueueNames)) {
+                $currentQueueNames = $queueNames;
+                $queueConsumed = false;
+            }
+
             $this->removeExpiredMessages();
             $this->redeliverMessages();
 
-            if ($message = $this->fetchMessage($queueNames, $redeliveryDelay)) {
+            if ($message = $this->fetchMessage($currentQueueNames, $redeliveryDelay)) {
+                $queueConsumed = true;
+
                 /**
-                 * @var DbalConsumer
+                 * @var DbalConsumer $consumer
                  * @var callable     $callback
                  */
                 [$consumer, $callback] = $this->subscribers[$message->getQueue()];
@@ -110,12 +118,13 @@ public function consume(int $timeout = 0): void
                     return;
                 }
 
-                $queueNames = array_filter($queueNames, static function ($queueName) use ($message) {
-                    return $message->getQueue() !== $queueName;
-                });
-                $queueNames[] = $message->getQueue();
+                unset($currentQueueNames[$message->getQueue()]);
             } else {
-                usleep($this->getPollingInterval() * 1000);
+                $currentQueueNames = [];
+
+                if (!$queueConsumed) {
+                    usleep($this->getPollingInterval() * 1000);
+                }
             }
 
             if ($timeout && microtime(true) >= $now + $timeout) {