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

hopefully fix missing channel errors

Browse Source
This commit is contained in:
Erik 2024-10-09 12:50:10 +03:00
parent b267921d48
commit c1df312b2c
1 changed files with 44 additions and 31 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

75
src/MessageBroker.ts
View File

@ -30,7 +30,8 @@ export interface BrokerOptions {
pass: string, pass: string,
vhost: string, vhost: string,
exchanges?: Record<string, ExchangeDef>, exchanges?: Record<string, ExchangeDef>,
queues?: Record<string, QueueDef> queues?: Record<string, QueueDef>,
heartbeat?: number
} }
interface InternalMessage { interface InternalMessage {
@ -79,6 +80,8 @@ class MessageBroker
#_pQueue: InternalPublishMsg[]; #_pQueue: InternalPublishMsg[];
#_qQueue: InternalQueueMsg[]; #_qQueue: InternalQueueMsg[];
#_qTO: NodeJS.Timeout | null; #_qTO: NodeJS.Timeout | null;
#heartbeat: number | undefined;
#creatingChannel?: Promise<ChannelWrapper> | null;
constructor (server: IServer, options: BrokerOptions) constructor (server: IServer, options: BrokerOptions)
{ {
@ -88,6 +91,7 @@ class MessageBroker
this.#password = options.pass; this.#password = options.pass;
this.#vhost = options.vhost || ''; this.#vhost = options.vhost || '';
this.#port = options.port; this.#port = options.port;
this.#heartbeat = options.heartbeat;
if (!this.#hosts.length) if (!this.#hosts.length)
throw new Error('Missing hosts configuration'); throw new Error('Missing hosts configuration');
@ -122,7 +126,7 @@ class MessageBroker
this.#logger?.info('Initialising message broker'); this.#logger?.info('Initialising message broker');
const credentials = this.#username ? `${this.#username}:${this.#password}@` : ''; const credentials = this.#username ? `${this.#username}:${this.#password}@` : '';
const connectionStrings = this.#hosts.map(host => `amqp://${credentials}${host}:${this.#port}/${this.#vhost}`); const connectionStrings = this.#hosts.map(host => `amqp://${credentials}${host}:${this.#port}/${this.#vhost}`);
this.#connection = amqp.connect(connectionStrings); this.#connection = amqp.connect(connectionStrings, { heartbeatIntervalInSeconds: this.#heartbeat });
this.#connection.on('disconnect', async ({ err }) => this.#connection.on('disconnect', async ({ err }) =>
{ {
@ -153,8 +157,7 @@ class MessageBroker
this.#connection?.once('connectFailed', reject); this.#connection?.once('connectFailed', reject);
}); });
await this.createChannel(); await this.awaitChannel();
this.#connection.on('connect', this.createChannel.bind(this));
} }
async close () async close ()
@ -232,14 +235,31 @@ class MessageBroker
await this.#_processQueues(); await this.#_processQueues();
} }
awaitChannel ()
{
if (this.#channel)
return Promise.resolve(this.#channel);
if (this.#creatingChannel)
return this.#creatingChannel;
this.#logger?.debug('Creating channel create promise');
const promise = new Promise<ChannelWrapper>((resolve, reject) =>
{
this.createChannel().then(() =>
{
this.#creatingChannel = null;
resolve(this.#channel!);
}).catch(reject);
});
this.#creatingChannel = promise;
return promise;
}
/** /**
* Consume queue * Consume queue
*/ */
async consume<T = unknown> (queue: string, consumer: Consumer<T>, options?: Options.Consume) async consume<T = unknown> (queue: string, consumer: Consumer<T>, options?: Options.Consume)
{ {
if (!this.#channel)
throw new Error('Channel does not exist');
this.#logger?.debug(`Adding queue consumer for ${queue}`); this.#logger?.debug(`Adding queue consumer for ${queue}`);
await this._consume<T>(queue, consumer, options); await this._consume<T>(queue, consumer, options);
@ -250,9 +270,8 @@ 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>
{ {
if (!this.#channel) await this.awaitChannel();
return Promise.reject(new Error('Channel doesn\'t exist')); 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`);
if (msg.content) if (msg.content)
@ -266,9 +285,6 @@ class MessageBroker
*/ */
async subscribe<T> (name: string, subscriber: Subscriber<T>, options: SubscriptionOptions = {}) async subscribe<T> (name: string, subscriber: Subscriber<T>, options: SubscriptionOptions = {})
{ {
if (!this.#channel)
throw new Error('Channel does not exist');
this.#logger?.debug(`Subscribing to ${name}`); this.#logger?.debug(`Subscribing to ${name}`);
// if (!this.#subscribers.has(name)) // if (!this.#subscribers.has(name))
await this._subscribe<T>(name, subscriber, options); await this._subscribe<T>(name, subscriber, options);
@ -280,15 +296,14 @@ 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>
{ {
if (!this.#channel) await this.awaitChannel();
return Promise.reject(new Error('Channel doesn\'t exist'));
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 });
const queue = await this.#channel.assertQueue('', { exclusive: true }); const queue = await this.#channel!.assertQueue('', { exclusive: true });
await this.#channel.bindQueue(queue.queue, name, options.routingKey ?? ''); await this.#channel!.bindQueue(queue.queue, name, options.routingKey ?? '');
await this.#channel.consume(queue.queue, async (msg) => await this.#channel!.consume(queue.queue, async (msg) =>
{ {
this.#logger?.debug(`Subscriber for ${name} (${listener.name}) fired`); this.#logger?.debug(`Subscriber for ${name} (${listener.name}) fired`);
if (msg.content && msg.content.toString().startsWith('{')) if (msg.content && msg.content.toString().startsWith('{'))
@ -361,32 +376,30 @@ class MessageBroker
} }
} }
assertExchange (exchange: string, props?: ExchangeDef) async assertExchange (exchange: string, props?: ExchangeDef)
{ {
if (!this.#channel) await this.awaitChannel();
throw new Error('Channel doesn\'t exist'); return this.#channel!.assertExchange(exchange, props?.type ?? 'fanout', props);
return this.#channel.assertExchange(exchange, props?.type ?? 'fanout', props);
} }
assertQueue (queue: string, opts?: QueueDef) async assertQueue (queue: string, opts?: QueueDef)
{ {
if (!this.#channel) await this.awaitChannel();
throw new Error('Channel doesn\'t exist'); 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 ()
{ {
if (!this.#channel) await this.awaitChannel();
throw new Error('Channel doesn\'t exist');
this.#logger?.status('Processing queues of unsent messages'); this.#logger?.status('Processing queues of unsent messages');
const pQ = [ ...this.#_pQueue ]; const pQ = [ ...this.#_pQueue ];
this.#_pQueue = []; this.#_pQueue = [];
for (const msg of pQ) for (const msg of pQ)
{ {
const { exchange, content, routingKey, properties } = msg; const { exchange, content, routingKey, properties } = msg;
const result = await this.#channel.publish(exchange, routingKey, Buffer.from(JSON.stringify(content)), properties) const result = await this.#channel!.publish(exchange, routingKey, Buffer.from(JSON.stringify(content)), properties)
.catch(() => null); .catch(() => null);
if (!result) if (!result)
this.#_pQueue.push(msg); this.#_pQueue.push(msg);
@ -395,7 +408,7 @@ class MessageBroker
for (const msg of qQ) for (const msg of qQ)
{ {
const { queue, content, properties } = msg; const { queue, content, properties } = msg;
const result = await this.#channel.sendToQueue(queue, Buffer.from(JSON.stringify(content)), properties).catch(() => null); const result = await this.#channel!.sendToQueue(queue, Buffer.from(JSON.stringify(content)), properties).catch(() => null);
if (!result) if (!result)
this.#_qQueue.push(msg); this.#_qQueue.push(msg);
} }