add logic to walk through the different queues after each message

Author: Henning

pkg/dbal/DbalConsumerHelperTrait.php

@@ -30,16 +30,17 @@ protected function fetchMessage(array $queues, int $redeliveryDelay): ?DbalMessa
         $endAt = microtime(true) + 0.2; // add 200ms
 
         $select = $this->getConnection()->createQueryBuilder()
-            ->select('id')
+            ->select('MIN(id)')
+            ->addSelect('queue')
             ->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')
+            ->groupBy('queue')
             ->setParameter('queues', $queues, Connection::PARAM_STR_ARRAY)
-            ->setParameter('delayedUntil', $now, DbalType::INTEGER)
-            ->setMaxResults(1);
+            ->setParameter('delayedUntil', $now, DbalType::INTEGER);
 
         $update = $this->getConnection()->createQueryBuilder()
             ->update($this->getContext()->getTableName())
@@ -53,7 +54,12 @@ protected function fetchMessage(array $queues, int $redeliveryDelay): ?DbalMessa
 
         while (microtime(true) < $endAt) {
             try {
-                $result = $select->execute()->fetch();
+                $results = $select->execute()->fetch();
+                if (empty($results)) {
+                    return null;
+                }
+
+                $result = $this->getResultByQueueList($results, $queues);
                 if (empty($result)) {
                     return null;
                 }
@@ -155,4 +161,17 @@ private function deleteMessage(string $deliveryId): void
             ['delivery_id' => DbalType::GUID]
         );
     }
+
+    private function getResultByQueueList(array $results, array $queues): ?array
+    {
+        $results = array_combine(array_column($results, 'queue'), $results);
+
+        foreach ($queues as $queue) {
+            if (isset($results[$queue])) {
+                return $results[$queue];
+            }
+        }
+
+        return null;
+    }
 }

pkg/dbal/DbalSubscriptionConsumer.php

@@ -87,8 +87,9 @@ public function consume(int $timeout = 0): void
 
         $queueNames = [];
         foreach (array_keys($this->subscribers) as $queueName) {
-            $queueNames[$queueName] = $queueName;
+            $queueNames[] = $queueName;
         }
+        $queueNames = array_unique($queueNames);
 
         $timeout /= 1000;
         $now = time();
@@ -108,6 +109,11 @@ public function consume(int $timeout = 0): void
                 if (false === call_user_func($callback, $message, $consumer)) {
                     return;
                 }
+
+                $queueNames = array_filter($queueNames, static function ($queueName) use ($message) {
+                    return $message->getQueue() !== $queueName;
+                });
+                $queueNames[] = $message->getQueue();
             } else {
                 usleep($this->getPollingInterval() * 1000);
             }