Await on redis.subscribe() to ensure that the method returns only after Redis has acknowledged the registration of the subscriber.

Author: ifullgaz

src/bus.ts

@@ -54,10 +54,10 @@ export class Bus {
     await this.processErrorRetryQueue()
   }
 
-  subscribe<T extends Serializable>(channel: string, handler: SubscribeHandler<T>) {
+  async subscribe<T extends Serializable>(channel: string, handler: SubscribeHandler<T>) {
     debug(`subscribing to channel ${channel}`)
 
-    return this.#transport.subscribe(channel, async (message) => {
+    return await this.#transport.subscribe(channel, async (message) => {
       debug('received message %j from bus', message)
       // @ts-expect-error - TODO: Weird typing issue
       handler(message)
@@ -94,7 +94,7 @@ export class Bus {
     return this.#transport.disconnect()
   }
 
-  unsubscribe(channel: string) {
-    return this.#transport.unsubscribe(channel)
+  async unsubscribe(channel: string) {
+    return await this.#transport.unsubscribe(channel)
   }
 }

src/bus_manager.ts

@@ -58,12 +58,12 @@ export class BusManager<KnownTransports extends Record<string, TransportConfig>>
     return this.use().publish(channel, message)
   }
 
-  subscribe<T extends Serializable>(channel: string, handler: SubscribeHandler<T>) {
-    return this.use().subscribe(channel, handler)
+  async subscribe<T extends Serializable>(channel: string, handler: SubscribeHandler<T>) {
+    return await this.use().subscribe(channel, handler)
   }
 
-  unsubscribe(channel: string) {
-    return this.use().unsubscribe(channel)
+  async unsubscribe(channel: string) {
+    return await this.use().unsubscribe(channel)
   }
 
   disconnect() {

src/transports/redis.ts

@@ -89,11 +89,7 @@ export class RedisTransport implements Transport {
     channel: string,
     handler: SubscribeHandler<T>
   ): Promise<void> {
-    this.#subscriber.subscribe(channel, (err) => {
-      if (err) {
-        throw err
-      }
-    })
+    await this.#subscriber.subscribe(channel)
 
     const event = this.#useMessageBuffer ? 'messageBuffer' : 'message'
     this.#subscriber.on(event, (receivedChannel: Buffer | string, message: Buffer | string) => {

test_helpers/chaos_transport.ts

@@ -59,8 +59,8 @@ export class ChaosTransport implements Transport {
     return this.#innerTransport.subscribe(channel, handler)
   }
 
-  unsubscribe(channel: string) {
-    return this.#innerTransport.unsubscribe(channel)
+  async unsubscribe(channel: string) {
+    return await this.#innerTransport.unsubscribe(channel)
   }
 
   disconnect() {