diff --git a/src/bus.ts b/src/bus.ts index ac7e849..cbb3eb6 100644 --- a/src/bus.ts +++ b/src/bus.ts @@ -54,10 +54,10 @@ export class Bus { await this.processErrorRetryQueue() } - subscribe(channel: string, handler: SubscribeHandler) { + async subscribe(channel: string, handler: SubscribeHandler) { 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) } } diff --git a/src/bus_manager.ts b/src/bus_manager.ts index fa8b494..12e01d8 100644 --- a/src/bus_manager.ts +++ b/src/bus_manager.ts @@ -58,12 +58,12 @@ export class BusManager> return this.use().publish(channel, message) } - subscribe(channel: string, handler: SubscribeHandler) { - return this.use().subscribe(channel, handler) + async subscribe(channel: string, handler: SubscribeHandler) { + 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() { diff --git a/src/transports/redis.ts b/src/transports/redis.ts index af89683..c968f73 100644 --- a/src/transports/redis.ts +++ b/src/transports/redis.ts @@ -89,11 +89,7 @@ export class RedisTransport implements Transport { channel: string, handler: SubscribeHandler ): Promise { - 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) => { diff --git a/test_helpers/chaos_transport.ts b/test_helpers/chaos_transport.ts index ba1a936..b175929 100644 --- a/test_helpers/chaos_transport.ts +++ b/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() {