feat: update RedisChannelConfig to use ConnectionOptions

src/channel/redis.channel-config.ts

@@ -1,25 +1,36 @@
 import { ChannelConfig } from '@nestjstools/messaging';
 
 export class RedisChannelConfig extends ChannelConfig {
-  public readonly connection: Connection;
+  public readonly connectionOptions: ConnectionOptions;
   public readonly queue: string;
 
   constructor({
-                name,
-                connection,
-                queue,
-                enableConsumer,
-                avoidErrorsForNotExistedHandlers,
-                middlewares,
-                normalizer,
-              }: RedisChannelConfig) {
-    super(name, avoidErrorsForNotExistedHandlers, middlewares, enableConsumer, normalizer)
-    this.connection = connection;
+    name,
+    connectionOptions,
+    queue,
+    enableConsumer,
+    avoidErrorsForNotExistedHandlers,
+    middlewares,
+    normalizer,
+  }: RedisChannelConfig) {
+    super(
+      name,
+      avoidErrorsForNotExistedHandlers,
+      middlewares,
+      enableConsumer,
+      normalizer,
+    );
+    this.connectionOptions = connectionOptions;
     this.queue = queue;
   }
 }
 
-interface Connection {
-  host: string;
-  port: number;
+interface ConnectionOptions {
+  redis: {
+    host: string;
+    port: number;
+    password?: string;
+    db?: number;
+  };
+  prefix?: string;
 }

src/channel/redis.channel.ts

@@ -8,6 +8,9 @@ export class RedisChannel extends Channel<RedisChannelConfig> {
 
   constructor(config: RedisChannelConfig) {
     super(config);
-    this.queue = new Queue(config.queue, { connection: config.connection });
+    this.queue = new Queue(config.queue, {
+      connection: config.connectionOptions.redis,
+      prefix: config.connectionOptions.prefix,
+    });
   }
 }

src/consumer/redis-messaging.consumer.ts

@@ -8,25 +8,36 @@ import { Worker } from 'bullmq';
 
 @Injectable()
 @MessageConsumer(RedisChannel)
-export class RedisMessagingConsumer implements IMessagingConsumer<RedisChannel>, OnApplicationShutdown {
+export class RedisMessagingConsumer
+  implements IMessagingConsumer<RedisChannel>, OnApplicationShutdown
+{
   private channel?: RedisChannel = undefined;
   private worker?: Worker = undefined;
 
-  async consume(dispatcher: ConsumerMessageDispatcher, channel: RedisChannel): Promise<void> {
+  async consume(
+    dispatcher: ConsumerMessageDispatcher,
+    channel: RedisChannel,
+  ): Promise<void> {
     this.channel = channel;
 
     this.worker = new Worker(
       channel.config.queue,
       async (job) => {
         dispatcher.dispatch(new ConsumerMessage(job.data, job.name));
       },
-      { connection: channel.config.connection }
+      {
+        connection: channel.config.connectionOptions.redis,
+        prefix: channel.config.connectionOptions.prefix,
+      },
     );
 
     return Promise.resolve();
   }
 
-  onError(errored: ConsumerDispatchedMessageError, channel: RedisChannel): Promise<void> {
+  onError(
+    errored: ConsumerDispatchedMessageError,
+    channel: RedisChannel,
+  ): Promise<void> {
     return Promise.resolve();
   }
 

src/message-bus/redis-message-bus-factory.ts

@@ -1,14 +1,15 @@
 import { Injectable } from '@nestjs/common';
 import { RedisMessageBus } from './redis-message.bus';
 import { RedisChannel } from '../channel/redis.channel';
-import {IMessageBusFactory} from "@nestjstools/messaging";
-import {MessageBusFactory} from "@nestjstools/messaging";
-import {IMessageBus} from "@nestjstools/messaging";
+import { IMessageBusFactory } from '@nestjstools/messaging';
+import { MessageBusFactory } from '@nestjstools/messaging';
+import { IMessageBus } from '@nestjstools/messaging';
 
 @Injectable()
 @MessageBusFactory(RedisChannel)
-export class RedisMessageBusFactory implements IMessageBusFactory<RedisChannel> {
-
+export class RedisMessageBusFactory
+  implements IMessageBusFactory<RedisChannel>
+{
   create(channel: RedisChannel): IMessageBus {
     return new RedisMessageBus(channel);
   }

src/message-bus/redis-message.bus.ts

@@ -5,10 +5,7 @@ import { RedisChannel } from '../channel/redis.channel';
 
 @Injectable()
 export class RedisMessageBus implements IMessageBus {
-  constructor(
-    private readonly redisChannel: RedisChannel,
-  ) {
-  }
+  constructor(private readonly redisChannel: RedisChannel) {}
 
   async dispatch(message: RoutingMessage): Promise<object | void> {
     this.redisChannel.queue.add(message.messageRoutingKey, message.message);