[12.x] RedisBroadcaster for RedisCuster

Author: vadimonus

.github/workflows/queues.yml

@@ -127,6 +127,12 @@ jobs:
           REDIS_CLIENT: ${{ matrix.client }}
           QUEUE_CONNECTION: redis
 
+      - name: Execute broadcasting tests
+        run: vendor/bin/phpunit tests/Integration/Broadcasting
+        env:
+          REDIS_CLIENT: ${{ matrix.client }}
+          QUEUE_CONNECTION: redis
+
   redis-cluster:
     runs-on: ubuntu-24.04
 
@@ -175,6 +181,12 @@ jobs:
           REDIS_CLUSTER_HOSTS_AND_PORTS: 127.0.0.1:7000,127.0.0.1:7001,127.0.0.1:7002
           REDIS_QUEUE: '{default}'
 
+      - name: Execute broadcasting tests
+        run: vendor/bin/phpunit tests/Integration/Broadcasting
+        env:
+          REDIS_CLIENT: ${{ matrix.client }}
+          REDIS_CLUSTER_HOSTS_AND_PORTS: 127.0.0.1:7000,127.0.0.1:7001,127.0.0.1:7002
+
   beanstalkd:
     runs-on: ubuntu-24.04
 

src/Illuminate/Broadcasting/Broadcasters/RedisBroadcaster.php

@@ -4,7 +4,11 @@
 
 use Illuminate\Broadcasting\BroadcastException;
 use Illuminate\Contracts\Redis\Factory as Redis;
+use Illuminate\Redis\Connections\PhpRedisClusterConnection;
+use Illuminate\Redis\Connections\PredisClusterConnection;
+use Illuminate\Redis\Connections\PredisConnection;
 use Illuminate\Support\Arr;
+use Predis\Connection\Cluster\RedisCluster;
 use Predis\Connection\ConnectionException;
 use RedisException;
 use Symfony\Component\HttpKernel\Exception\AccessDeniedHttpException;
@@ -125,10 +129,29 @@ public function broadcast(array $channels, $event, array $payload = [])
         ]);
 
         try {
-            $connection->eval(
-                $this->broadcastMultipleChannelsScript(),
-                0, $payload, ...$this->formatChannels($channels)
-            );
+            if ($connection instanceof PhpRedisClusterConnection) {
+                foreach ($channels as $channel) {
+                    $connection->publish($channel, $payload);
+                }
+            } elseif ($connection instanceof PredisClusterConnection
+                && $connection->client()->getConnection() instanceof RedisCluster
+            ) {
+                $randomClusterNodeConnection = new PredisConnection(
+                    $connection->client()->getClientBy('slot', mt_rand(0, 16383))
+                );
+                if ($events = $connection->getEventDispatcher()) {
+                    $randomClusterNodeConnection->setEventDispatcher($events);
+                }
+                $randomClusterNodeConnection->eval(
+                    $this->broadcastMultipleChannelsScript(),
+                    0, $payload, ...$this->formatChannels($channels)
+                );
+            } else {
+                $connection->eval(
+                    $this->broadcastMultipleChannelsScript(),
+                    0, $payload, ...$this->formatChannels($channels)
+                );
+            }
         } catch (ConnectionException|RedisException $e) {
             throw new BroadcastException(
                 sprintf('Redis error: %s.', $e->getMessage())

tests/Integration/Broadcasting/RedisBroadcasterTest.php

@@ -0,0 +1,54 @@
+<?php
+
+namespace Illuminate\Tests\Integration\Broadcasting;
+
+use Illuminate\Broadcasting\Broadcasters\RedisBroadcaster;
+use Illuminate\Foundation\Testing\Concerns\InteractsWithRedis;
+use Orchestra\Testbench\TestCase;
+use PHPUnit\Framework\Attributes\DataProvider;
+use PHPUnit\Framework\Attributes\RequiresPhpExtension;
+
+#[RequiresPhpExtension('redis')]
+class RedisBroadcasterTest extends TestCase
+{
+    use InteractsWithRedis;
+
+    /**
+     * @param  string  $driver
+     */
+    #[DataProvider('redisDriverProvider')]
+    #[RequiresPhpExtension('pcntl')]
+    public function testBroadcast($driver)
+    {
+        $this->beforeApplicationDestroyed(function () {
+            $this->tearDownRedis();
+        });
+
+        if ($pid = pcntl_fork() > 0) {
+            $this->setUpRedis();
+            /** @var \Redis|\RedisCluster $redisClient */
+            $redisClient = $this->redis['phpredis']->client();
+            $redisClient->subscribe(['channel-1'], function ($redis, $channel, $message) {
+                $redis->unsubscribe(['channel-1']);
+                $redis->close();
+                $receivedPayload = json_decode($message, true);
+                $this->assertEquals('test_channel-1', $channel);
+                $this->assertEquals([
+                    'event' => 'test.event',
+                    'data' => ['foo' => 'bar'],
+                    'socket' => null,
+                ], $receivedPayload);
+            });
+        } elseif ($pid == 0) {
+            $this->setUpRedis();
+            $redis = $this->redis[$driver];
+            $broadcaster = new RedisBroadcaster($redis, null, 'test_');
+            $channels = ['channel-1', 'channel-2'];
+            usleep(1000);
+            $broadcaster->broadcast($channels, 'test.event', ['foo' => 'bar']);
+            exit;
+        } else {
+            $this->fail('Cannot fork');
+        }
+    }
+}