hrms-bangkok/hrms-api-org
7
0
Fork 0
Code Issues Pull requests Projects Releases Packages 1 Wiki Activity Actions

ออกคำสั่ง ต้องกดออกคำสั่ง 2 รอบติดกันระบบถึงจะออกคำสั่งให้#97

Browse source
This commit is contained in:
AdisakKanthawilang 2024-11-25 14:16:08 +07:00
parent 2ebb3e9b2b
commit 16d0e5545e
1 changed files with 9 additions and 15 deletions
Download patch file Download diff file Expand all files Collapse all files

24
src/services/rabbitmq.ts
View file

@ -25,11 +25,9 @@ export async function init() {
console.log(channel ? "[AMQ] Create channel success" : "[AMQ] Create channel failed"); console.log(channel ? "[AMQ] Create channel success" : "[AMQ] Create channel failed");
await Promise.all([ channel.assertQueue(queue, { durable: true }), //----> (1.5) assert queue and set durable (if "true" save to disk on RabbitMQ)
channel.assertQueue(queue, { durable: true }), //----> (1.5) assert queue and set durable (if "true" save to disk on RabbitMQ) channel.assertQueue(queue_org, { durable: true }),
channel.assertQueue(queue_org, { durable: true }), channel.prefetch(1)
channel.prefetch(1)
]);
sendToQueue = (payload: any, persistent = true) => { sendToQueue = (payload: any, persistent = true) => {
//----> (2) sendQueue To RabbitMQ and set persistent (if "true" redo the failed queue when server run again) //----> (2) sendQueue To RabbitMQ and set persistent (if "true" redo the failed queue when server run again)
@ -43,16 +41,14 @@ export async function init() {
}; };
console.log("[AMQ] Listening for message..."); console.log("[AMQ] Listening for message...");
await Promise.all([ createConsumer(queue, channel, handler), //----> (3) Process Consumer
createConsumer(queue, channel, handler), //----> (3) Process Consumer createConsumer(queue_org, channel, handler_org)
createConsumer(queue_org, channel, handler_org) // createConsumer(queue2, channel, handler2);
// createConsumer(queue2, channel, handler2);
]);
} }
let retries = 0; let retries = 0;
async function createConsumer( //----> consumer function createConsumer( //----> consumer
queue: string, queue: string,
channel: amqp.Channel, channel: amqp.Channel,
handler: (msg: amqp.ConsumeMessage) => Promise<boolean> | boolean, handler: (msg: amqp.ConsumeMessage) => Promise<boolean> | boolean,
@ -60,14 +56,13 @@ async function createConsumer( //----> consumer
channel.consume( channel.consume(
queue, queue,
async (msg) => { async (msg) => {
console.log("[AMQ] MSQ",msg);
if (!msg) return; if (!msg) return;
if ((await handler(msg)) || retries++ >= 3) { if ((await handler(msg)) || retries++ >= 3) {
retries = 0; retries = 0;
console.log("[AMQ] Process consumer success"); console.log("[AMQ] Process Consumer success");
return channel.ack(msg); return channel.ack(msg);
} }
console.log("[AMQ] Process consumer fail"); console.log("[AMQ] Process Consumer fail");
return await new Promise((resolve) => setTimeout(() => resolve(channel.nack(msg)), 3000)); return await new Promise((resolve) => setTimeout(() => resolve(channel.nack(msg)), 3000));
}, },
{ noAck: false }, { noAck: false },
@ -75,7 +70,6 @@ async function createConsumer( //----> consumer
} }
async function handler(msg: amqp.ConsumeMessage): Promise<boolean> { async function handler(msg: amqp.ConsumeMessage): Promise<boolean> {
console.log(msg ? "[AMQ] On handler" : "[AMQ] Not handler");
//----> condition before process consumer //----> condition before process consumer
const repo = AppDataSource.getRepository(Command); const repo = AppDataSource.getRepository(Command);
const { data, token, user } = JSON.parse(msg.content.toString()); const { data, token, user } = JSON.parse(msg.content.toString());