feat(redis): add redis cluster support in

Author: Julien-R44

compose.yml

@@ -0,0 +1,19 @@
+services:
+  redis:
+    image: redis:7.2-alpine
+    ports:
+      - "6379:6379"
+
+  redis-cluster:
+    profiles:
+      - cluster
+    image: grokzen/redis-cluster:7.0.8
+    environment:
+      - IP=0.0.0.0
+      - INITIAL_PORT=7000
+      - MASTERS=3
+      - SLAVES_PER_MASTER=0
+    ports:
+      - "7000:7000"
+      - "7001:7001"
+      - "7002:7002"

src/transports/redis.ts

@@ -5,7 +5,7 @@
  * @copyright BoringNode
  */
 
-import { Redis } from 'ioredis'
+import { Redis, Cluster } from 'ioredis'
 import { assert } from '@poppinss/utils/assert'
 
 import debug from '../debug.js'
@@ -24,21 +24,36 @@ export function redis(config: RedisTransportConfig, encoder?: TransportEncoder)
 }
 
 export class RedisTransport implements Transport {
-  readonly #publisher: Redis
-  readonly #subscriber: Redis
+  readonly #publisher: Redis | Cluster
+  readonly #subscriber: Redis | Cluster
   readonly #encoder: TransportEncoder
   readonly #useMessageBuffer: boolean = false
 
   #id: string | undefined
 
   constructor(path: string, encoder?: TransportEncoder)
   constructor(options: RedisTransportConfig, encoder?: TransportEncoder)
-  constructor(options: RedisTransportConfig | string, encoder?: TransportEncoder) {
+  constructor(connection: Redis | Cluster, encoder?: TransportEncoder)
+  constructor(
+    options: RedisTransportConfig | string | Redis | Cluster,
+    encoder?: TransportEncoder
+  ) {
+    this.#encoder = encoder ?? new JsonEncoder()
+
+    /**
+     * If an existing Redis or Cluster instance is passed, we duplicate it
+     * to have separate connections for publisher and subscriber
+     */
+    if (options instanceof Redis || options instanceof Cluster) {
+      this.#publisher = options.duplicate()
+      this.#subscriber = options.duplicate()
+      return
+    }
+
     // @ts-expect-error - merged definitions of overloaded constructor is not public
     this.#publisher = new Redis(options)
     // @ts-expect-error - merged definitions of overloaded constructor is not public
     this.#subscriber = new Redis(options)
-    this.#encoder = encoder ?? new JsonEncoder()
 
     if (typeof options === 'object') {
       this.#useMessageBuffer = options.useMessageBuffer ?? false

src/types/main.ts

@@ -7,6 +7,8 @@
 
 import type { RedisOptions } from 'ioredis'
 import type { IClientOptions } from 'mqtt'
+
+export type { Redis, Cluster } from 'ioredis'
 export type TransportFactory = () => Transport
 
 /**

tests/drivers/redis_transport.spec.ts

@@ -7,6 +7,7 @@
 
 import { setTimeout } from 'node:timers/promises'
 import { test } from '@japa/runner'
+import { Redis, Cluster } from 'ioredis'
 import { RedisContainer, StartedRedisContainer } from '@testcontainers/redis'
 import { RedisTransport } from '../../src/transports/redis.js'
 import { JsonEncoder } from '../../src/encoders/json_encoder.js'
@@ -146,4 +147,63 @@ test.group('Redis Transport', (group) => {
     await setTimeout(200)
     await transport2.publish('testing-channel', data)
   }).waitForDone()
+
+  test('should work with an existing redis instance', async ({ assert, cleanup }, done) => {
+    assert.plan(1)
+
+    const redisInstance = new Redis({
+      host: container.getHost(),
+      port: container.getMappedPort(6379),
+    })
+
+    cleanup(async () => {
+      await redisInstance.quit()
+    })
+
+    const transport1 = new RedisTransport(redisInstance).setId('bus1')
+    const transport2 = new RedisTransport(redisInstance).setId('bus2')
+
+    cleanup(async () => {
+      await transport1.disconnect()
+      await transport2.disconnect()
+    })
+
+    await transport1.subscribe('testing-channel', (payload) => {
+      assert.equal(payload, 'test')
+      done()
+    })
+
+    await setTimeout(200)
+
+    await transport2.publish('testing-channel', 'test')
+  }).waitForDone()
+
+  test('should work with an existing cluster instance', async ({ assert, cleanup }, done) => {
+    assert.plan(1)
+
+    const cluster = new Cluster([{ host: '127.0.0.1', port: 7000 }])
+
+    cleanup(async () => {
+      await cluster.quit()
+    })
+
+    const transport1 = new RedisTransport(cluster).setId('bus1')
+    const transport2 = new RedisTransport(cluster).setId('bus2')
+
+    cleanup(async () => {
+      await transport1.disconnect()
+      await transport2.disconnect()
+    })
+
+    await transport1.subscribe('testing-channel', (payload) => {
+      assert.equal(payload, 'test')
+      done()
+    })
+
+    await setTimeout(200)
+
+    await transport2.publish('testing-channel', 'test')
+  })
+    .waitForDone()
+    .skip(!!process.env.CI, 'Skipping cluster test on CI')
 })