Skip to content

refactor: update RedisChannelConfig to use ConnectionOptions #2

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
wants to merge 3 commits into
base: main
Choose a base branch
from
+51 −28
Open

refactor: update RedisChannelConfig to use ConnectionOptions #2

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
cf4d219
refactor: update RedisChannelConfig to use ConnectionOptions
kastov Jul 8, 2025
7977d6d
refactor: rename connection property to connectionOptions and enhance…
kastov Jul 12, 2025
636c5b4
refactor: remove unused comments
kastov Jul 12, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
39 changes: 25 additions & 14 deletions src/channel/redis.channel-config.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -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;
Comment on lines +28 to +35
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Hey, change looks like improvement to supports passowrd/db which is cool, but at the same time it is a BC.

Can you make something like:

Suggested change
interface ConnectionOptions {
redis: {
host: string;
port: number;
password?: string;
db?: number;
};
prefix?: string;
interface ConnectionOptions {
host: string;
port: number;
password?: string;
db?: number;
prefix?: string;

which not provide a BC in contract, just supports additional properties.

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

is prefix is a part of connection (I think no 😄 but I can be wrong)?
May you can move it to RedisChannelConfig beyond ConnectionOptions
and changes in RedisChannel, RedisMessagingConsumer can be reverted, because everything will be in connection.

Copy link
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

is prefix is a part of connection (I think no 😄 but I can be wrong)? May you can move it to RedisChannelConfig beyond ConnectionOptions and changes in RedisChannel, RedisMessagingConsumer can be reverted, because everything will be in connection.

Yeah, it is part of connection, but with BullMQ we need to use separate one.

Link: taskforcesh/bullmq#1219 (comment)

}
5 changes: 4 additions & 1 deletion src/channel/redis.channel.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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,
});
}
}
19 changes: 15 additions & 4 deletions src/consumer/redis-messaging.consumer.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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();
}

Expand Down
11 changes: 6 additions & 5 deletions src/message-bus/redis-message-bus-factory.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -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);
}
Expand Down
5 changes: 1 addition & 4 deletions src/message-bus/redis-message.bus.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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);
Expand Down