Qless sharding work incorrect (#23)

Author: Vitalii Bondiuk

src/Queue/QlessConnectionHandler.php

@@ -61,14 +61,17 @@ public function getCurrentClient(): Client
 
     public function getNextClient(): Client
     {
-        if ($this->clientIterator->current() === null) {
+        $this->clientIterator->next();
+
+        if (!$this->clientIterator->valid()) {
             $this->clientIterator->rewind();
         }
 
-        $currentClient = $this->clientIterator->current();
-        $this->clientIterator->next();
-
-        return $currentClient;
+        return $this->clientIterator->current();
     }
 
+    public function getClientCount(): int
+    {
+        return $this->clientIterator->count();
+    }
 }

src/Queue/QlessQueue.php

@@ -9,7 +9,6 @@
 use LaravelQless\Job\AbstractJob;
 use LaravelQless\Job\QlessJob;
 use Qless\Client;
-use Qless\Jobs\BaseJob;
 use Qless\Topics\Topic;
 
 /**
@@ -163,13 +162,21 @@ public function recur(int $interval, string $job, array $data, ?string $queueNam
      */
     public function pop($queueName = null)
     {
-        $connection = $this->getNextConnection();
+        $connectionCount = $this->getClientCount();
 
-        /** @var \Qless\Queues\Queue $queue */
-        $queue = $connection->queues[$queueName];
+        for ($i = 0; $i < $connectionCount; $i++) {
+            $connection = $this->getNextConnection();
 
-        /** @var BaseJob $job */
-        $job = $queue->pop(self::WORKER_PREFIX . $connection->getWorkerName());
+            /** @var \Qless\Queues\Queue $queue */
+            $queue = $connection->queues[$queueName];
+
+            /** @var \Qless\Jobs\BaseJob $job */
+            $job = $queue->pop(self::WORKER_PREFIX . $connection->getWorkerName());
+
+            if ($job) {
+                break;
+            }
+        }
 
         if (!$job) {
             return null;
@@ -319,4 +326,9 @@ public function getConnection(): Client
     {
         return $this->getCurrentConnection();
     }
+
+    public function getClientCount(): int
+    {
+        return $this->clients->getClientCount();
+    }
 }