Press n or j to go to the next uncovered block, b, p or k for the previous block.
| 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 | 3x 3x 3x 3x 3x 1x 3x 3x 1x 1x 3x 3x 1x 1x 1x 1x 1x 1x 1x | import amqplib, { ChannelModel, Channel, ConsumeMessage } from 'amqplib';
import { Logger } from '../utils/logger.js';
import { IJobHandler } from '../types/IJobHandler.js';
export class QueueHandler {
private readonly rmqUrl: string;
private connection: ChannelModel | undefined;
private channel: Channel | undefined;
constructor(rmqUrl: string) {
this.rmqUrl = rmqUrl;
}
private async getChannel(queue: string) {
if (!this.connection) {
Logger.log('Create RabbitMq connection ...');
this.connection = await amqplib.connect(this.rmqUrl);
this.connection.on('error', (err) => {
Logger.error(err);
});
Logger.log('RabbitMq connection created');
}
if (!this.channel) {
this.channel = await this.connection.createChannel();
this.channel.assertQueue(queue, { durable: true });
}
return this.channel;
}
async addJobToQueue<T>(queue: string, data: T): Promise<boolean> {
const channel = await this.getChannel(queue);
Logger.log(`Sending message to rabbitMQ queue: ${queue}`);
const ok = channel.sendToQueue(queue, Buffer.from(JSON.stringify(data)), { persistent: true });
Logger.log(`Sent message to rabbitMQ queue: ${queue}, ok: ${ok}`);
return ok;
}
async attachJobHandlerToQueue(queue: string, jobHandler: IJobHandler) {
const channel = await this.getChannel(queue);
await channel.consume(queue, async (msg: ConsumeMessage | null) => {
if (!msg) {
Logger.error('jobHandler retrieved empty message!');
return;
}
Logger.log(`New message from rabbitMQ queue: ${queue}`);
channel.ack(msg); // default acknowledgement timeout is 30 minutes
await jobHandler.handleMessage(msg);
Logger.log(`Handled message from rabbitMQ queue: ${queue}`);
});
}
}
|