refactor(BaseRedis): group as constructor param and cleanup subscribers

Author: nizans

packages/brokers/src/brokers/Broker.ts

@@ -44,11 +44,11 @@ export interface IBaseBroker<TEvents extends Record<string, any>> {
 	/**
 	 * Subscribes to the given events, grouping them by the given group name
 	 */
-	subscribe(group: string, events: (keyof TEvents)[]): Promise<void>;
+	subscribe(events: (keyof TEvents)[]): Promise<void>;
 	/**
 	 * Unsubscribes from the given events - it's required to pass the same group name as when subscribing for proper cleanup
 	 */
-	unsubscribe(group: string, events: (keyof TEvents)[]): Promise<void>;
+	unsubscribe(events: (keyof TEvents)[]): Promise<void>;
 }
 
 export interface IPubSubBroker<TEvents extends Record<string, any>>

packages/brokers/src/brokers/redis/BaseRedis.ts

@@ -23,10 +23,19 @@ export interface RedisBrokerOptions extends BaseBrokerOptions {
 	 * How long to block for messages when polling
 	 */
 	blockTimeout?: number;
+
+	/**
+	 * Consumer group name to use for this broker
+	 *
+	 * @see {@link https://redis.io/commands/xreadgroup/}
+	 */
+	group: string;
+
 	/**
 	 * Max number of messages to poll at once
 	 */
 	maxChunk?: number;
+
 	/**
 	 * Unique consumer name.
 	 *
@@ -43,7 +52,7 @@ export const DefaultRedisBrokerOptions = {
 	name: randomBytes(20).toString('hex'),
 	maxChunk: 10,
 	blockTimeout: 5_000,
-} as const satisfies Required<RedisBrokerOptions>;
+} as const satisfies Required<Omit<RedisBrokerOptions, 'group'>>;
 
 /**
  * Helper class with shared Redis logic
@@ -93,32 +102,32 @@ export abstract class BaseRedisBroker<TEvents extends Record<string, any>>
 	/**
 	 * {@inheritDoc IBaseBroker.subscribe}
 	 */
-	public async subscribe(group: string, events: (keyof TEvents)[]): Promise<void> {
+	public async subscribe(events: (keyof TEvents)[]): Promise<void> {
 		await Promise.all(
 			// @ts-expect-error: Intended
 			events.map(async (event) => {
 				this.subscribedEvents.add(event as string);
 				try {
-					return await this.redisClient.xgroup('CREATE', event as string, group, 0, 'MKSTREAM');
+					return await this.redisClient.xgroup('CREATE', event as string, this.options.group, 0, 'MKSTREAM');
 				} catch (error) {
 					if (!(error instanceof ReplyError)) {
 						throw error;
 					}
 				}
 			}),
 		);
-		void this.listen(group);
+		void this.listen();
 	}
 
 	/**
 	 * {@inheritDoc IBaseBroker.unsubscribe}
 	 */
-	public async unsubscribe(group: string, events: (keyof TEvents)[]): Promise<void> {
+	public async unsubscribe(events: (keyof TEvents)[]): Promise<void> {
 		const commands: unknown[][] = Array.from({ length: events.length * 2 });
 		for (let idx = 0; idx < commands.length; idx += 2) {
 			const event = events[idx / 2];
-			commands[idx] = ['xgroup', 'delconsumer', event as string, group, this.options.name];
-			commands[idx + 1] = ['xcleangroup', event as string, group];
+			commands[idx] = ['xgroup', 'delconsumer', event as string, this.options.group, this.options.name];
+			commands[idx + 1] = ['xcleangroup', event as string, this.options.group];
 		}
 
 		await this.redisClient.pipeline(commands).exec();
@@ -131,7 +140,7 @@ export abstract class BaseRedisBroker<TEvents extends Record<string, any>>
 	/**
 	 * Begins polling for events, firing them to {@link BaseRedisBroker.listen}
 	 */
-	protected async listen(group: string): Promise<void> {
+	protected async listen(): Promise<void> {
 		if (this.listening) {
 			return;
 		}
@@ -142,7 +151,7 @@ export abstract class BaseRedisBroker<TEvents extends Record<string, any>>
 			try {
 				const data = await this.streamReadClient.xreadgroupBuffer(
 					'GROUP',
-					group,
+					this.options.group,
 					this.options.name,
 					'COUNT',
 					String(this.options.maxChunk),
@@ -169,7 +178,7 @@ export abstract class BaseRedisBroker<TEvents extends Record<string, any>>
 							continue;
 						}
 
-						this.emitEvent(id, group, event.toString('utf8'), this.options.decode(data));
+						this.emitEvent(id, this.options.group, event.toString('utf8'), this.options.decode(data));
 					}
 				}
 			} catch (error) {
@@ -185,6 +194,7 @@ export abstract class BaseRedisBroker<TEvents extends Record<string, any>>
 	 * Destroys the broker, closing all connections
 	 */
 	public async destroy() {
+		await this.unsubscribe(Array.from(this.subscribedEvents));
 		this.streamReadClient.disconnect();
 		this.redisClient.disconnect();
 		this.removeAllListeners();