Navy.gif/wrappers
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Fix potential bug

Browse Source
This commit is contained in:
Erik 2024-10-13 00:56:27 +03:00
parent 4e9c63856c
commit 88dc5dfb4d
1 changed files with 8 additions and 7 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

15
src/MessageBroker.ts
View File

@ -145,6 +145,7 @@ class MessageBroker
this.#connection.on('connect', async ({ url }) => this.#connection.on('connect', async ({ url }) =>
{ {
this.#logger?.status(`Message broker connected to ${url}`); this.#logger?.status(`Message broker connected to ${url}`);
this.ensureChannel();
}); });
await new Promise<void>((resolve, reject) => await new Promise<void>((resolve, reject) =>
@ -157,7 +158,7 @@ class MessageBroker
this.#connection?.once('connectFailed', reject); this.#connection?.once('connectFailed', reject);
}); });
await this.awaitChannel(); await this.ensureChannel();
} }
async close () async close ()
@ -235,7 +236,7 @@ class MessageBroker
await this.#_processQueues(); await this.#_processQueues();
} }
awaitChannel () ensureChannel ()
{ {
if (this.#channel) if (this.#channel)
return Promise.resolve(this.#channel); return Promise.resolve(this.#channel);
@ -270,7 +271,7 @@ class MessageBroker
private async _consume<T> (queue: string, consumer: Consumer<T>, options?: Options.Consume): Promise<void> private async _consume<T> (queue: string, consumer: Consumer<T>, options?: Options.Consume): Promise<void>
{ {
await this.awaitChannel(); await this.ensureChannel();
await this.#channel!.consume(queue, async (msg: ConsumeMessage) => await this.#channel!.consume(queue, async (msg: ConsumeMessage) =>
{ {
this.#logger?.debug(`Consumer for ${queue} (${consumer.name}) fired`); this.#logger?.debug(`Consumer for ${queue} (${consumer.name}) fired`);
@ -296,7 +297,7 @@ class MessageBroker
private async _subscribe<T> (name: string, listener: Subscriber<T>, options: SubscriptionOptions): Promise<void> private async _subscribe<T> (name: string, listener: Subscriber<T>, options: SubscriptionOptions): Promise<void>
{ {
await this.awaitChannel(); await this.ensureChannel();
const type = options.exchangeType ?? 'fanout'; const type = options.exchangeType ?? 'fanout';
await this.#channel!.assertExchange(name, type, { durable: true }); await this.#channel!.assertExchange(name, type, { durable: true });
@ -378,20 +379,20 @@ class MessageBroker
async assertExchange (exchange: string, props?: ExchangeDef) async assertExchange (exchange: string, props?: ExchangeDef)
{ {
await this.awaitChannel(); await this.ensureChannel();
return this.#channel!.assertExchange(exchange, props?.type ?? 'fanout', props); return this.#channel!.assertExchange(exchange, props?.type ?? 'fanout', props);
} }
async assertQueue (queue: string, opts?: QueueDef) async assertQueue (queue: string, opts?: QueueDef)
{ {
await this.awaitChannel(); await this.ensureChannel();
return this.#channel!.assertQueue(queue, opts); return this.#channel!.assertQueue(queue, opts);
} }
// Processes messages queued up while the broker was unreachable // Processes messages queued up while the broker was unreachable
async #_processQueues () async #_processQueues ()
{ {
await this.awaitChannel(); await this.ensureChannel();
this.#logger?.status('Processing queues of unsent messages'); this.#logger?.status('Processing queues of unsent messages');
const pQ = [ ...this.#_pQueue ]; const pQ = [ ...this.#_pQueue ];