[11.x] Job Batches with Redis Cluster (#54522)

* Job Batches with Redis Cluster

* Update RedisQueue.php

---------

Co-authored-by: Vadim Dvorovenko <[email protected]>
Co-authored-by: Taylor Otwell <[email protected]>

Author: 3 people

src/Illuminate/Queue/RedisQueue.php

@@ -6,6 +6,8 @@
 use Illuminate\Contracts\Queue\Queue as QueueContract;
 use Illuminate\Contracts\Redis\Factory as Redis;
 use Illuminate\Queue\Jobs\RedisJob;
+use Illuminate\Redis\Connections\PhpRedisClusterConnection;
+use Illuminate\Redis\Connections\PredisClusterConnection;
 use Illuminate\Support\Str;
 
 class RedisQueue extends Queue implements QueueContract, ClearableQueue
@@ -118,17 +120,25 @@ public function size($queue = null)
      */
     public function bulk($jobs, $data = '', $queue = null)
     {
-        $this->getConnection()->pipeline(function () use ($jobs, $data, $queue) {
-            $this->getConnection()->transaction(function () use ($jobs, $data, $queue) {
-                foreach ((array) $jobs as $job) {
-                    if (isset($job->delay)) {
-                        $this->later($job->delay, $job, $data, $queue);
-                    } else {
-                        $this->push($job, $data, $queue);
-                    }
+        $connection = $this->getConnection();
+
+        $bulk = function () use ($jobs, $data, $queue) {
+            foreach ((array) $jobs as $job) {
+                if (isset($job->delay)) {
+                    $this->later($job->delay, $job, $data, $queue);
+                } else {
+                    $this->push($job, $data, $queue);
                 }
-            });
-        });
+            }
+        };
+
+        if ($connection instanceof PhpRedisClusterConnection) {
+            $connection->transaction($bulk);
+        } elseif ($connection instanceof PredisClusterConnection) {
+            $connection->pipeline($bulk);
+        } else {
+            $connection->pipeline(fn () => $connection->transaction($bulk));
+        }
     }
 
     /**

tests/Integration/Queue/RedisQueueTest.php

@@ -9,8 +9,6 @@
 use Illuminate\Queue\Events\JobQueueing;
 use Illuminate\Queue\Jobs\RedisJob;
 use Illuminate\Queue\RedisQueue;
-use Illuminate\Redis\Connections\PhpRedisClusterConnection;
-use Illuminate\Redis\Connections\PredisClusterConnection;
 use Illuminate\Support\InteractsWithTime;
 use Illuminate\Support\Str;
 use Mockery as m;
@@ -517,11 +515,6 @@ public function testPushJobQueueingAndJobQueuedEvents($driver)
     #[DataProvider('redisDriverProvider')]
     public function testBulkJobQueuedEvent($driver)
     {
-        if ($this->redis[$driver]->connection() instanceof PhpRedisClusterConnection
-            || $this->redis[$driver]->connection() instanceof PredisClusterConnection
-        ) {
-            $this->markTestSkipped('RedisQueue::bulk currently does not support cluster connections');
-        }
         $events = m::mock(Dispatcher::class);
         $events->shouldReceive('dispatch')->with(m::type(JobQueueing::class))->andReturnNull()->times(3);
         $events->shouldReceive('dispatch')->with(m::type(JobQueued::class))->andReturnNull()->times(3);