feat(redis): add support for useMessageBuffer option when using cluster

Author: Julien-R44

src/transports/redis.ts

@@ -17,6 +17,7 @@ import type {
   Serializable,
   SubscribeHandler,
   RedisTransportConfig,
+  RedisTransportOptions,
 } from '../types/main.js'
 
 export function redis(config: RedisTransportConfig, encoder?: TransportEncoder) {
@@ -33,10 +34,15 @@ export class RedisTransport implements Transport {
 
   constructor(path: string, encoder?: TransportEncoder)
   constructor(options: RedisTransportConfig, encoder?: TransportEncoder)
-  constructor(connection: Redis | Cluster, encoder?: TransportEncoder)
+  constructor(
+    connection: Redis | Cluster,
+    encoder?: TransportEncoder,
+    options?: RedisTransportOptions
+  )
   constructor(
     options: RedisTransportConfig | string | Redis | Cluster,
-    encoder?: TransportEncoder
+    encoder?: TransportEncoder,
+    transportOptions?: RedisTransportOptions
   ) {
     this.#encoder = encoder ?? new JsonEncoder()
 
@@ -47,6 +53,7 @@ export class RedisTransport implements Transport {
     if (options instanceof Redis || options instanceof Cluster) {
       this.#publisher = options.duplicate()
       this.#subscriber = options.duplicate()
+      this.#useMessageBuffer = transportOptions?.useMessageBuffer ?? false
       return
     }
 

src/types/main.ts

@@ -39,6 +39,15 @@ export interface RedisTransportConfig extends RedisOptions {
   useMessageBuffer?: boolean
 }
 
+export interface RedisTransportOptions {
+  /**
+   * If true, we will use `messageBuffer` event instead of `message` event
+   * that is emitted by ioredis. `messageBuffer` will returns a buffer instead
+   * of a string and this is useful when you are dealing with binary data.
+   */
+  useMessageBuffer?: boolean
+}
+
 export enum MqttProtocol {
   MQTT = 'mqtt',
   MQTTS = 'mqtts',

tests/drivers/redis_transport.spec.ts

@@ -206,4 +206,54 @@ test.group('Redis Transport', (group) => {
   })
     .waitForDone()
     .skip(!!process.env.CI, 'Skipping cluster test on CI')
+
+  test('send binary data using useMessageBuffer with existing redis instance', async ({
+    assert,
+    cleanup,
+  }, done) => {
+    assert.plan(1)
+
+    class BinaryEncoder implements TransportEncoder {
+      encode(message: TransportMessage<any>) {
+        return Buffer.from(JSON.stringify(message))
+      }
+
+      decode(data: string | Buffer) {
+        const buffer = Buffer.isBuffer(data) ? data : Buffer.from(data, 'binary')
+        return JSON.parse(buffer.toString())
+      }
+    }
+
+    const redisInstance = new Redis({
+      host: container.getHost(),
+      port: container.getMappedPort(6379),
+    })
+
+    cleanup(async () => {
+      await redisInstance.quit()
+    })
+
+    const transport1 = new RedisTransport(redisInstance, new BinaryEncoder(), {
+      useMessageBuffer: true,
+    }).setId('bus1')
+
+    const transport2 = new RedisTransport(redisInstance, new BinaryEncoder(), {
+      useMessageBuffer: true,
+    }).setId('bus2')
+
+    cleanup(async () => {
+      await transport1.disconnect()
+      await transport2.disconnect()
+    })
+
+    const data = ['foo', '👍']
+
+    await transport1.subscribe('testing-channel', (payload) => {
+      assert.deepEqual(payload, data)
+      done()
+    })
+
+    await setTimeout(200)
+    await transport2.publish('testing-channel', data)
+  }).waitForDone()
 })