feat: update RedisChannelConfig to use ConnectionOptions (#2)

* refactor: update RedisChannelConfig to use ConnectionOptions

* refactor: rename connection property to connectionOptions and enhance ConnectionOptions interface in RedisChannelConfig

* refactor: remove unused comments

* refactor: update RedisChannelConfig to use a simplified Connection interface and adjust related components

Author: kastov

src/channel/redis.channel-config.ts

@@ -1,19 +1,24 @@
 import { ChannelConfig } from '@nestjstools/messaging';
-
 export class RedisChannelConfig extends ChannelConfig {
   public readonly connection: Connection;
   public readonly queue: string;
 
   constructor({
-                name,
-                connection,
-                queue,
-                enableConsumer,
-                avoidErrorsForNotExistedHandlers,
-                middlewares,
-                normalizer,
-              }: RedisChannelConfig) {
-    super(name, avoidErrorsForNotExistedHandlers, middlewares, enableConsumer, normalizer)
+    name,
+    connection,
+    queue,
+    enableConsumer,
+    avoidErrorsForNotExistedHandlers,
+    middlewares,
+    normalizer,
+  }: RedisChannelConfig) {
+    super(
+      name,
+      avoidErrorsForNotExistedHandlers,
+      middlewares,
+      enableConsumer,
+      normalizer,
+    );
     this.connection = connection;
     this.queue = queue;
   }
@@ -22,4 +27,11 @@ export class RedisChannelConfig extends ChannelConfig {
 interface Connection {
   host: string;
   port: number;
+  password?: string;
+  db?: number;
+  /**
+   * This prefix is not used as RedisOptions keyPrefix, it is used as prefix for BullMQ
+   * Read more: https://github.com/taskforcesh/bullmq/issues/1219#issuecomment-1113903785
+   */
+  keyPrefix?: string;
 }

src/channel/redis.channel-factory.ts

@@ -1,11 +1,13 @@
 import { RedisChannel } from './redis.channel';
-import {Injectable} from "@nestjs/common";
+import { Injectable } from '@nestjs/common';
 import { ChannelFactory, IChannelFactory } from '@nestjstools/messaging';
 import { RedisChannelConfig } from './redis.channel-config';
 
 @Injectable()
 @ChannelFactory(RedisChannelConfig)
-export class RedisChannelFactory implements IChannelFactory<RedisChannelConfig> {
+export class RedisChannelFactory
+  implements IChannelFactory<RedisChannelConfig>
+{
   create(channelConfig: RedisChannelConfig): RedisChannel {
     return new RedisChannel(channelConfig);
   }

src/channel/redis.channel.ts

@@ -8,6 +8,14 @@ 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: {
+        host: config.connection.host,
+        port: config.connection.port,
+        password: config.connection.password,
+        db: config.connection.db,
+      },
+      prefix: config.connection.keyPrefix,
+    });
   }
 }

src/consumer/redis-messaging.consumer.ts

@@ -8,25 +8,41 @@ 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: {
+          host: channel.config.connection.host,
+          port: channel.config.connection.port,
+          password: channel.config.connection.password,
+          db: channel.config.connection.db,
+        },
+        prefix: channel.config.connection.keyPrefix,
+      },
     );
 
     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);