/* * Copyright (c) 2022. * Author Peter Placzek (tada5hi) * For the full copyright and license information, * view the LICENSE file that was distributed with this source code. */ import RabbitMQ, {Channel, Connection, ConsumeMessage} from "amqplib"; import {v4} from "uuid"; import env from "../../env"; import {QueueMessage} from "./type"; let connection : Connection | undefined; export async function useMessageQueue() { if(typeof connection !== 'undefined') { return connection; } connection = await RabbitMQ.connect(env.rabbitMqConnectionString); return connection; } export async function publishMessageQueue(routingKey: string, data: QueueMessage) { const text = JSON.stringify(data); const connection: Connection = await useMessageQueue(); const channel : Channel = await connection.createChannel(); await channel.assertExchange('app', 'topic', { durable: true }); channel.publish('app', routingKey, Buffer.from(text)); } export async function consumeMessageQueue(routingKey: string | string[], cb: (channel: Channel, msg: ConsumeMessage) => void) { const connection : Connection = await useMessageQueue(); const channel : Channel = await connection.createChannel(); await channel.assertExchange('app', 'topic', { durable: true, }); const assertionQueue = await channel.assertQueue('', { durable: false, autoDelete: true }); if(Array.isArray(routingKey)) { for(let i=0; i cb(channel, msg)); } export type QueChannelHandler = (message: QueueMessage) => Promise; export async function handleMessageQueueChannel(channel: Channel, handlers: Record, msg: ConsumeMessage) { const json : any = JSON.parse(msg.content.toString('utf-8')); const queueMessage : QueueMessage | undefined = !!json ? json : undefined; const handler = handlers[queueMessage.type] || handlers.$any; return handler ? handler(queueMessage) : Promise.resolve(true); } export function createQueueMessageTemplate() : QueueMessage { return { id: v4(), type: undefined, metadata: { }, data: { } } }