NeoAnd/evolution-api
1
0
Fork 0
mirror of https://github.com/EvolutionAPI/evolution-api.git synced 2025-12-18 19:32:21 -06:00
Code Issues Actions Packages Projects Releases Wiki Activity

New global mode for rabbitmq events

Browse Source
This commit is contained in:
Davidson Gomes
2024-05-24 18:48:54 -03:00
parent 68dcc1c86a
commit f0d40eea15
10 changed files with 298 additions and 51 deletions
Download Patch File Download Diff File Expand all files Collapse all files

35
src/api/integrations/rabbitmq/libs/amqp.server.ts
View File

@@ -42,6 +42,41 @@ export const getAMQP = (): amqp.Channel | null => {
return amqpChannel;
};
export const initGlobalQueues = () => {
logger.info('Initializing global queues');
const events = configService.get<Rabbitmq>('RABBITMQ').EVENTS;
if (!events) {
logger.warn('No events to initialize on AMQP');
return;
}
const eventKeys = Object.keys(events);
eventKeys.forEach((event) => {
if (events[event] === false) return;
const queueName = `${event.replace(/_/g, '.').toLowerCase()}`;
const amqp = getAMQP();
const exchangeName = 'evolution_exchange';
amqp.assertExchange(exchangeName, 'topic', {
durable: true,
autoDelete: false,
});
amqp.assertQueue(queueName, {
durable: true,
autoDelete: false,
arguments: {
'x-queue-type': 'quorum',
},
});
amqp.bindQueue(queueName, exchangeName, event);
});
};
export const initQueues = (instanceName: string, events: string[]) => {
if (!events || !events.length) return;

145
src/api/services/channel.service.ts
View File

@@ -13,6 +13,7 @@ import {
Database,
HttpServer,
Log,
Rabbitmq,
Sqs,
Webhook,
Websocket,
@@ -688,6 +689,9 @@ export class ChannelStartupService {
const rabbitmqLocal = this.localRabbitmq.events;
const sqsLocal = this.localSqs.events;
const serverUrl = this.configService.get<HttpServer>('SERVER').URL;
const rabbitmqEnabled = this.configService.get<Rabbitmq>('RABBITMQ').ENABLED;
const rabbitmqGlobal = this.configService.get<Rabbitmq>('RABBITMQ').GLOBAL_ENABLED;
const rabbitmqEvents = this.configService.get<Rabbitmq>('RABBITMQ').EVENTS;
const we = event.replace(/[.-]/gm, '_').toUpperCase();
const transformedWe = we.replace(/_/gm, '-').toLowerCase();
const tzoffset = new Date().getTimezoneOffset() * 60000; //offset in milliseconds
@@ -698,67 +702,134 @@ export class ChannelStartupService {
const tokenStore = await this.repository.auth.find(this.instanceName);
const instanceApikey = tokenStore?.apikey || 'Apikey not found';
if (this.localRabbitmq.enabled) {
if (rabbitmqEnabled) {
const amqp = getAMQP();
if (amqp) {
if (this.localRabbitmq.enabled && amqp) {
if (Array.isArray(rabbitmqLocal) && rabbitmqLocal.includes(we)) {
const exchangeName = this.instanceName ?? 'evolution_exchange';
// await amqp.assertExchange(exchangeName, 'topic', {
// durable: true,
// autoDelete: false,
// });
let retry = 0;
await this.assertExchangeAsync(amqp, exchangeName, 'topic', {
durable: true,
autoDelete: false,
});
while (retry < 3) {
try {
await amqp.assertExchange(exchangeName, 'topic', {
durable: true,
autoDelete: false,
});
const queueName = `${this.instanceName}.${event}`;
const queueName = `${this.instanceName}.${event}`;
await amqp.assertQueue(queueName, {
durable: true,
autoDelete: false,
arguments: {
'x-queue-type': 'quorum',
},
});
await amqp.assertQueue(queueName, {
durable: true,
autoDelete: false,
arguments: {
'x-queue-type': 'quorum',
},
});
await amqp.bindQueue(queueName, exchangeName, event);
await amqp.bindQueue(queueName, exchangeName, event);
const message = {
event,
instance: this.instance.name,
data,
server_url: serverUrl,
date_time: now,
sender: this.wuid,
};
const message = {
event,
instance: this.instance.name,
data,
server_url: serverUrl,
date_time: now,
sender: this.wuid,
};
if (expose && instanceApikey) {
message['apikey'] = instanceApikey;
if (expose && instanceApikey) {
message['apikey'] = instanceApikey;
}
await amqp.publish(exchangeName, event, Buffer.from(JSON.stringify(message)));
if (this.configService.get<Log>('LOG').LEVEL.includes('WEBHOOKS')) {
const logData = {
local: ChannelStartupService.name + '.sendData-RabbitMQ',
event,
instance: this.instance.name,
data,
server_url: serverUrl,
apikey: (expose && instanceApikey) || null,
date_time: now,
sender: this.wuid,
};
if (expose && instanceApikey) {
logData['apikey'] = instanceApikey;
}
this.logger.log(logData);
}
break;
} catch (error) {
retry++;
}
}
}
}
await amqp.publish(exchangeName, event, Buffer.from(JSON.stringify(message)));
if (rabbitmqGlobal && rabbitmqEvents[we] && amqp) {
const exchangeName = 'evolution_exchange';
if (this.configService.get<Log>('LOG').LEVEL.includes('WEBHOOKS')) {
const logData = {
local: ChannelStartupService.name + '.sendData-RabbitMQ',
let retry = 0;
while (retry < 3) {
try {
await amqp.assertExchange(exchangeName, 'topic', {
durable: true,
autoDelete: false,
});
const queueName = transformedWe;
await amqp.assertQueue(queueName, {
durable: true,
autoDelete: false,
arguments: {
'x-queue-type': 'quorum',
},
});
await amqp.bindQueue(queueName, exchangeName, event);
const message = {
event,
instance: this.instance.name,
data,
server_url: serverUrl,
apikey: (expose && instanceApikey) || null,
date_time: now,
sender: this.wuid,
};
if (expose && instanceApikey) {
logData['apikey'] = instanceApikey;
message['apikey'] = instanceApikey;
}
await amqp.publish(exchangeName, event, Buffer.from(JSON.stringify(message)));
if (this.configService.get<Log>('LOG').LEVEL.includes('WEBHOOKS')) {
const logData = {
local: ChannelStartupService.name + '.sendData-RabbitMQ-Global',
event,
instance: this.instance.name,
data,
server_url: serverUrl,
apikey: (expose && instanceApikey) || null,
date_time: now,
sender: this.wuid,
};
if (expose && instanceApikey) {
logData['apikey'] = instanceApikey;
}
this.logger.log(logData);
}
this.logger.log(logData);
break;
} catch (error) {
retry++;
}
}
}