650 lines
24 KiB
JavaScript
650 lines
24 KiB
JavaScript
/*
|
|
* (c) Copyright Ascensio System SIA 2010-2024
|
|
*
|
|
* This program is a free software product. You can redistribute it and/or
|
|
* modify it under the terms of the GNU Affero General Public License (AGPL)
|
|
* version 3 as published by the Free Software Foundation. In accordance with
|
|
* Section 7(a) of the GNU AGPL its Section 15 shall be amended to the effect
|
|
* that Ascensio System SIA expressly excludes the warranty of non-infringement
|
|
* of any third-party rights.
|
|
*
|
|
* This program is distributed WITHOUT ANY WARRANTY; without even the implied
|
|
* warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. For
|
|
* details, see the GNU AGPL at: http://www.gnu.org/licenses/agpl-3.0.html
|
|
*
|
|
* You can contact Ascensio System SIA at 20A-6 Ernesta Birznieka-Upish
|
|
* street, Riga, Latvia, EU, LV-1050.
|
|
*
|
|
* The interactive user interfaces in modified source and object code versions
|
|
* of the Program must display Appropriate Legal Notices, as required under
|
|
* Section 5 of the GNU AGPL version 3.
|
|
*
|
|
* Pursuant to Section 7(b) of the License you must retain the original Product
|
|
* logo when distributing the program. Pursuant to Section 7(e) we decline to
|
|
* grant you any rights under trademark law for use of our trademarks.
|
|
*
|
|
* All the Product's GUI elements, including illustrations and icon sets, as
|
|
* well as technical writing content are licensed under the terms of the
|
|
* Creative Commons Attribution-ShareAlike 4.0 International. See the License
|
|
* terms at http://creativecommons.org/licenses/by-sa/4.0/legalcode
|
|
*
|
|
*/
|
|
|
|
'use strict';
|
|
const config = require('config');
|
|
const events = require('events');
|
|
const util = require('util');
|
|
const co = require('co');
|
|
const utils = require('./utils');
|
|
const constants = require('./constants');
|
|
const rabbitMQCore = require('./rabbitMQCore');
|
|
const activeMQCore = require('./activeMQCore');
|
|
const commonDefines = require('./commondefines');
|
|
const operationContext = require('./operationContext');
|
|
|
|
const cfgMaxRedeliveredCount = config.get('FileConverter.converter.maxRedeliveredCount');
|
|
const cfgQueueType = config.get('queue.type');
|
|
const cfgVisibilityTimeout = config.get('queue.visibilityTimeout');
|
|
const cfgQueueRetentionPeriod = config.get('queue.retentionPeriod');
|
|
const cfgRabbitQueueConvertTask = config.util.cloneDeep(config.get('rabbitmq.queueconverttask'));
|
|
const cfgRabbitQueueConvertResponse = config.util.cloneDeep(config.get('rabbitmq.queueconvertresponse'));
|
|
const cfgRabbitExchangeConvertDead = config.util.cloneDeep(config.get('rabbitmq.exchangeconvertdead'));
|
|
const cfgRabbitQueueConvertDead = config.util.cloneDeep(config.get('rabbitmq.queueconvertdead'));
|
|
const cfgRabbitQueueDelayed = config.util.cloneDeep(config.get('rabbitmq.queuedelayed'));
|
|
const cfgActiveQueueConvertTask = constants.ACTIVEMQ_QUEUE_PREFIX + config.get('activemq.queueconverttask');
|
|
const cfgActiveQueueConvertResponse = constants.ACTIVEMQ_QUEUE_PREFIX + config.get('activemq.queueconvertresponse');
|
|
const cfgActiveQueueConvertDead = constants.ACTIVEMQ_QUEUE_PREFIX + config.get('activemq.queueconvertdead');
|
|
const cfgActiveQueueDelayed = constants.ACTIVEMQ_QUEUE_PREFIX + config.get('activemq.queuedelayed');
|
|
|
|
function initRabbit(taskqueue, isAddTask, isAddResponse, isAddTaskReceive, isAddResponseReceive, isEmitDead, isAddDelayed, callback) {
|
|
return co(function* () {
|
|
let e = null;
|
|
try {
|
|
const conn = yield rabbitMQCore.connetPromise(() => {
|
|
clear(taskqueue);
|
|
if (!taskqueue.isClose) {
|
|
setTimeout(() => {
|
|
init(taskqueue, isAddTask, isAddResponse, isAddTaskReceive, isAddResponseReceive, isEmitDead, isAddDelayed, null);
|
|
}, rabbitMQCore.RECONNECT_TIMEOUT);
|
|
}
|
|
});
|
|
taskqueue.connection = conn;
|
|
let bAssertTaskQueue = false;
|
|
const optionsTaskQueueDefault = {
|
|
messageTtl: cfgQueueRetentionPeriod * 1000,
|
|
deadLetterExchange: cfgRabbitExchangeConvertDead.name
|
|
};
|
|
const optionsTaskQueue = {...optionsTaskQueueDefault, ...cfgRabbitQueueConvertTask.options};
|
|
if (isAddTask) {
|
|
taskqueue.channelConvertTask = yield rabbitMQCore.createConfirmChannelPromise(conn);
|
|
yield rabbitMQCore.assertQueuePromise(taskqueue.channelConvertTask, cfgRabbitQueueConvertTask.name, optionsTaskQueue);
|
|
bAssertTaskQueue = true;
|
|
}
|
|
let bAssertResponseQueue = false;
|
|
if (isAddResponse) {
|
|
taskqueue.channelConvertResponse = yield rabbitMQCore.createConfirmChannelPromise(conn);
|
|
yield rabbitMQCore.assertQueuePromise(
|
|
taskqueue.channelConvertResponse,
|
|
cfgRabbitQueueConvertResponse.name,
|
|
cfgRabbitQueueConvertResponse.options
|
|
);
|
|
bAssertResponseQueue = true;
|
|
}
|
|
const optionsReceive = {noAck: false};
|
|
if (isAddTaskReceive) {
|
|
taskqueue.channelConvertTaskReceive = yield rabbitMQCore.createChannelPromise(conn);
|
|
taskqueue.channelConvertTaskReceive.prefetch(1);
|
|
if (!bAssertTaskQueue) {
|
|
yield rabbitMQCore.assertQueuePromise(taskqueue.channelConvertTaskReceive, cfgRabbitQueueConvertTask.name, optionsTaskQueue);
|
|
}
|
|
yield rabbitMQCore.consumePromise(
|
|
taskqueue.channelConvertTaskReceive,
|
|
cfgRabbitQueueConvertTask.name,
|
|
message => {
|
|
co(function* () {
|
|
const ack = function () {
|
|
taskqueue.channelConvertTaskReceive && taskqueue.channelConvertTaskReceive.ack(message);
|
|
};
|
|
const redelivered = yield* pushBackRedeliveredRabbit(taskqueue, message, ack);
|
|
if (!redelivered) {
|
|
if (message) {
|
|
taskqueue.emit('task', message.content.toString(), ack);
|
|
}
|
|
}
|
|
});
|
|
},
|
|
optionsReceive
|
|
);
|
|
}
|
|
if (isAddResponseReceive) {
|
|
taskqueue.channelConvertResponseReceive = yield rabbitMQCore.createChannelPromise(conn);
|
|
if (!bAssertResponseQueue) {
|
|
yield rabbitMQCore.assertQueuePromise(
|
|
taskqueue.channelConvertResponseReceive,
|
|
cfgRabbitQueueConvertResponse.name,
|
|
cfgRabbitQueueConvertResponse.options
|
|
);
|
|
}
|
|
yield rabbitMQCore.consumePromise(
|
|
taskqueue.channelConvertResponseReceive,
|
|
cfgRabbitQueueConvertResponse.name,
|
|
message => {
|
|
if (message) {
|
|
taskqueue.emit('response', message.content.toString(), () => {
|
|
taskqueue.channelConvertResponseReceive && taskqueue.channelConvertResponseReceive.ack(message);
|
|
});
|
|
}
|
|
},
|
|
optionsReceive
|
|
);
|
|
}
|
|
if (isAddDelayed) {
|
|
const optionsDelayedQueueDefault = {
|
|
deadLetterExchange: cfgRabbitExchangeConvertDead.name
|
|
};
|
|
const optionsDelayedQueue = {...optionsDelayedQueueDefault, ...cfgRabbitQueueDelayed.options};
|
|
taskqueue.channelDelayed = yield rabbitMQCore.createConfirmChannelPromise(conn);
|
|
yield rabbitMQCore.assertQueuePromise(taskqueue.channelDelayed, cfgRabbitQueueDelayed.name, optionsDelayedQueue);
|
|
}
|
|
if (isEmitDead) {
|
|
taskqueue.channelConvertDead = yield rabbitMQCore.createChannelPromise(conn);
|
|
yield rabbitMQCore.assertExchangePromise(
|
|
taskqueue.channelConvertDead,
|
|
cfgRabbitExchangeConvertDead.name,
|
|
'fanout',
|
|
cfgRabbitExchangeConvertDead.options
|
|
);
|
|
const queue = yield rabbitMQCore.assertQueuePromise(
|
|
taskqueue.channelConvertDead,
|
|
cfgRabbitQueueConvertDead.name,
|
|
cfgRabbitQueueConvertDead.options
|
|
);
|
|
|
|
taskqueue.channelConvertDead.bindQueue(queue, cfgRabbitExchangeConvertDead.name, '');
|
|
yield rabbitMQCore.consumePromise(
|
|
taskqueue.channelConvertDead,
|
|
queue,
|
|
message => {
|
|
if (null != taskqueue.channelConvertDead) {
|
|
if (message) {
|
|
taskqueue.emit('dead', message.content.toString(), () => {
|
|
taskqueue.channelConvertDead.ack(message);
|
|
});
|
|
}
|
|
}
|
|
},
|
|
{noAck: false}
|
|
);
|
|
}
|
|
|
|
//process messages received while reconnection time
|
|
repeat(taskqueue);
|
|
} catch (err) {
|
|
e = err;
|
|
}
|
|
if (callback) {
|
|
callback(e);
|
|
}
|
|
});
|
|
}
|
|
function initActive(taskqueue, isAddTask, isAddResponse, isAddTaskReceive, isAddResponseReceive, isEmitDead, isAddDelayed, callback) {
|
|
return co(function* () {
|
|
let e = null;
|
|
try {
|
|
const conn = yield activeMQCore.connetPromise(() => {
|
|
clear(taskqueue);
|
|
if (!taskqueue.isClose) {
|
|
setTimeout(() => {
|
|
init(taskqueue, isAddTask, isAddResponse, isAddTaskReceive, isAddResponseReceive, isEmitDead, isAddDelayed, null);
|
|
}, activeMQCore.RECONNECT_TIMEOUT);
|
|
}
|
|
});
|
|
taskqueue.connection = conn;
|
|
if (isAddTask) {
|
|
//https://github.com/amqp/rhea/issues/251#issuecomment-535076570
|
|
const optionsConvertTask = {
|
|
target: {
|
|
address: cfgActiveQueueConvertTask,
|
|
capabilities: ['queue']
|
|
}
|
|
};
|
|
taskqueue.channelConvertTask = yield activeMQCore.openSenderPromise(conn, optionsConvertTask);
|
|
initSenderActive(taskqueue.channelConvertTask, taskqueue.channelConvertTaskData);
|
|
}
|
|
if (isAddResponse) {
|
|
const optionsConvertResponse = {
|
|
target: {
|
|
address: cfgActiveQueueConvertResponse,
|
|
capabilities: ['queue']
|
|
}
|
|
};
|
|
taskqueue.channelConvertResponse = yield activeMQCore.openSenderPromise(conn, optionsConvertResponse);
|
|
initSenderActive(taskqueue.channelConvertResponse, taskqueue.channelConvertResponseData);
|
|
}
|
|
if (isAddTaskReceive) {
|
|
const optionsConvertTask = {
|
|
source: {
|
|
address: cfgActiveQueueConvertTask,
|
|
capabilities: ['queue']
|
|
},
|
|
credit_window: 0,
|
|
autoaccept: false
|
|
};
|
|
const receiver = yield activeMQCore.openReceiverPromise(conn, optionsConvertTask);
|
|
//todo ?consumer.dispatchAsync=false&consumer.prefetchSize=1
|
|
receiver.add_credit(1);
|
|
receiver.on('message', context => {
|
|
co(function* () {
|
|
const ack = function () {
|
|
context.delivery.accept();
|
|
receiver.add_credit(1);
|
|
};
|
|
const redelivered = yield* pushBackRedeliveredActive(taskqueue, context, ack);
|
|
if (!redelivered) {
|
|
if (context) {
|
|
taskqueue.emit('task', context.message.body, ack);
|
|
}
|
|
}
|
|
});
|
|
});
|
|
taskqueue.channelConvertTaskReceive = receiver;
|
|
}
|
|
if (isAddResponseReceive) {
|
|
const optionsConvertResponse = {
|
|
source: {
|
|
address: cfgActiveQueueConvertResponse,
|
|
capabilities: ['queue']
|
|
},
|
|
credit_window: 0,
|
|
autoaccept: false
|
|
};
|
|
const receiver = yield activeMQCore.openReceiverPromise(conn, optionsConvertResponse);
|
|
//todo ?consumer.dispatchAsync=false&consumer.prefetchSize=1
|
|
receiver.add_credit(1);
|
|
receiver.on('message', context => {
|
|
if (context) {
|
|
taskqueue.emit('response', context.message.body, () => {
|
|
context.delivery.accept();
|
|
receiver.add_credit(1);
|
|
});
|
|
}
|
|
});
|
|
taskqueue.channelConvertResponseReceive = receiver;
|
|
}
|
|
if (isAddDelayed) {
|
|
const optionsDelayed = {
|
|
target: {
|
|
address: cfgActiveQueueDelayed,
|
|
capabilities: ['queue']
|
|
}
|
|
};
|
|
taskqueue.channelDelayed = yield activeMQCore.openSenderPromise(conn, optionsDelayed);
|
|
initSenderActive(taskqueue.channelDelayed, taskqueue.channelDelayedData);
|
|
}
|
|
if (isEmitDead) {
|
|
const optionsConvertDead = {
|
|
source: {
|
|
address: cfgActiveQueueConvertDead,
|
|
capabilities: ['queue']
|
|
},
|
|
credit_window: 0,
|
|
autoaccept: false
|
|
};
|
|
const receiver = yield activeMQCore.openReceiverPromise(conn, optionsConvertDead);
|
|
//todo ?consumer.dispatchAsync=false&consumer.prefetchSize=1
|
|
receiver.add_credit(1);
|
|
receiver.on('message', context => {
|
|
if (context) {
|
|
taskqueue.emit('dead', context.message.body, () => {
|
|
context.delivery.accept();
|
|
receiver.add_credit(1);
|
|
});
|
|
}
|
|
});
|
|
taskqueue.channelConvertDead = receiver;
|
|
}
|
|
|
|
//process messages received while reconnection time
|
|
repeat(taskqueue);
|
|
} catch (err) {
|
|
e = err;
|
|
}
|
|
if (callback) {
|
|
callback(e);
|
|
}
|
|
});
|
|
}
|
|
function clear(taskqueue) {
|
|
taskqueue.channelConvertTask = null;
|
|
taskqueue.channelConvertTaskReceive = null;
|
|
taskqueue.channelConvertDead = null;
|
|
taskqueue.channelConvertResponse = null;
|
|
taskqueue.channelConvertResponseReceive = null;
|
|
taskqueue.channelDelayed = null;
|
|
//todo clear all listeners
|
|
taskqueue.channelConvertTaskData = {};
|
|
taskqueue.channelConvertResponseData = {};
|
|
taskqueue.channelDelayedData = {};
|
|
}
|
|
function* pushBackRedeliveredRabbit(taskqueue, message, ack) {
|
|
if (message?.fields?.redelivered) {
|
|
try {
|
|
operationContext.global.logger.warn('checkRedelivered redelivered data=%j', message);
|
|
//remove current task and add new into tail of queue to remove redelivered flag
|
|
const data = message.content.toString();
|
|
const redeliveredCount = message.properties.headers['x-redelivered-count'];
|
|
if (!redeliveredCount || redeliveredCount < cfgMaxRedeliveredCount) {
|
|
message.properties.headers['x-redelivered-count'] = redeliveredCount ? redeliveredCount + 1 : 1;
|
|
yield addTaskString(taskqueue, data, message.properties.priority, undefined, message.properties.headers);
|
|
} else if (taskqueue.simulateErrorResponse) {
|
|
yield taskqueue.addResponse(taskqueue.simulateErrorResponse(data));
|
|
}
|
|
} catch (err) {
|
|
operationContext.global.logger.error('checkRedelivered error: %s', err.stack);
|
|
} finally {
|
|
ack();
|
|
}
|
|
return true;
|
|
}
|
|
return false;
|
|
}
|
|
function* pushBackRedeliveredActive(taskqueue, context, ack) {
|
|
if (undefined !== context.message.delivery_count) {
|
|
operationContext.global.logger.warn('checkRedelivered redelivered data=%j', context.message);
|
|
if (context.message.delivery_count > cfgMaxRedeliveredCount) {
|
|
try {
|
|
if (taskqueue.simulateErrorResponse) {
|
|
yield taskqueue.addResponse(taskqueue.simulateErrorResponse(context.message.body));
|
|
}
|
|
} catch (err) {
|
|
operationContext.global.logger.error('checkRedelivered error: %s', err.stack);
|
|
} finally {
|
|
ack();
|
|
}
|
|
return true;
|
|
}
|
|
}
|
|
return false;
|
|
}
|
|
function repeat(taskqueue) {
|
|
//repeat addTask because they are lost after the reconnection
|
|
//unlike unconfirmed task will come again
|
|
//acknowledge data after reconnect raises an exception 'PRECONDITION_FAILED - unknown delivery tag'
|
|
for (let i = 0; i < taskqueue.addTaskStore.length; ++i) {
|
|
const elem = taskqueue.addTaskStore[i];
|
|
addTask(taskqueue, elem.task, elem.priority, () => {}, elem.expiration, elem.headers);
|
|
}
|
|
taskqueue.addTaskStore.length = 0;
|
|
for (let i = 0; i < taskqueue.addDelayedStore.length; ++i) {
|
|
const elem = taskqueue.addDelayedStore[i];
|
|
addDelayed(taskqueue, elem.task, elem.ttl, () => {});
|
|
}
|
|
taskqueue.addDelayedStore.length = 0;
|
|
}
|
|
function addTaskRabbit(taskqueue, content, priority, callback, opt_expiration, opt_headers) {
|
|
const options = {persistent: true, priority};
|
|
if (undefined !== opt_expiration) {
|
|
options.expiration = opt_expiration.toString();
|
|
}
|
|
if (undefined !== opt_headers) {
|
|
options.headers = opt_headers;
|
|
}
|
|
taskqueue.channelConvertTask.sendToQueue(cfgRabbitQueueConvertTask.name, content, options, callback);
|
|
}
|
|
function addTaskActive(taskqueue, content, priority, callback, opt_expiration, _opt_headers) {
|
|
const msg = {durable: true, priority, body: content, ttl: cfgQueueRetentionPeriod * 1000};
|
|
if (undefined !== opt_expiration) {
|
|
msg.ttl = opt_expiration;
|
|
}
|
|
const delivery = taskqueue.channelConvertTask.send(msg);
|
|
if (delivery) {
|
|
taskqueue.channelConvertTaskData[delivery.id] = callback;
|
|
}
|
|
}
|
|
function addTaskString(taskqueue, task, priority, opt_expiration, opt_headers) {
|
|
//todo confirmation mode
|
|
return new Promise((resolve, reject) => {
|
|
const content = Buffer.from(task);
|
|
if (null != taskqueue.channelConvertTask) {
|
|
addTask(
|
|
taskqueue,
|
|
content,
|
|
priority,
|
|
(err, _ok) => {
|
|
if (null != err) {
|
|
reject(err);
|
|
} else {
|
|
resolve();
|
|
}
|
|
},
|
|
opt_expiration,
|
|
opt_headers
|
|
);
|
|
} else {
|
|
taskqueue.addTaskStore.push({task: content, priority, expiration: opt_expiration, headers: opt_headers});
|
|
resolve();
|
|
}
|
|
});
|
|
}
|
|
function addResponseRabbit(taskqueue, content, callback) {
|
|
const options = {persistent: true};
|
|
taskqueue.channelConvertResponse.sendToQueue(cfgRabbitQueueConvertResponse.name, content, options, callback);
|
|
}
|
|
function addResponseActive(taskqueue, content, callback) {
|
|
const msg = {durable: true, body: content};
|
|
const delivery = taskqueue.channelConvertResponse.send(msg);
|
|
if (delivery) {
|
|
taskqueue.channelConvertResponseData[delivery.id] = callback;
|
|
}
|
|
}
|
|
function closeRabbit(conn) {
|
|
return rabbitMQCore.closePromise(conn);
|
|
}
|
|
function closeActive(conn) {
|
|
return activeMQCore.closePromise(conn);
|
|
}
|
|
function addDelayedRabbit(taskqueue, content, ttl, callback) {
|
|
const options = {persistent: true, expiration: ttl.toString()};
|
|
taskqueue.channelDelayed.sendToQueue(cfgRabbitQueueDelayed.name, content, options, callback);
|
|
}
|
|
function addDelayedActive(taskqueue, content, ttl, callback) {
|
|
const msg = {durable: true, body: content, ttl};
|
|
const delivery = taskqueue.channelDelayed.send(msg);
|
|
if (delivery) {
|
|
taskqueue.channelDelayedData[delivery.id] = callback;
|
|
}
|
|
}
|
|
|
|
function healthCheckRabbit(taskqueue) {
|
|
return co(function* () {
|
|
if (!taskqueue.channelConvertDead) {
|
|
return false;
|
|
}
|
|
const exchange = yield rabbitMQCore.assertExchangePromise(
|
|
taskqueue.channelConvertDead,
|
|
cfgRabbitExchangeConvertDead.name,
|
|
'fanout',
|
|
cfgRabbitExchangeConvertDead.options
|
|
);
|
|
return !!exchange;
|
|
});
|
|
}
|
|
/**
|
|
* Check ActiveMQ connection health without using a generator.
|
|
* @param {TaskQueueRabbitMQ} taskqueue - Queue instance holding the connection
|
|
* @returns {Promise<boolean>} true if connection is open
|
|
*/
|
|
function healthCheckActive(taskqueue) {
|
|
// todo better check
|
|
if (!taskqueue.connection) {
|
|
return Promise.resolve(false);
|
|
}
|
|
return Promise.resolve(taskqueue.connection.is_open());
|
|
}
|
|
|
|
function initSenderActive(sender, senderData) {
|
|
const processEvent = function (context, res) {
|
|
const id = context?.delivery?.id;
|
|
const callback = senderData[id];
|
|
if (callback) {
|
|
delete senderData[id];
|
|
callback(res);
|
|
}
|
|
};
|
|
|
|
sender.on('accepted', context => {
|
|
processEvent(context, null);
|
|
});
|
|
sender.on('rejected ', context => {
|
|
const error = context.delivery?.remote_state?.error;
|
|
processEvent(context, new Error('[AMQP] message is rejected (error=' + error + ')'));
|
|
});
|
|
sender.on('released', context => {
|
|
const delivery_failed = context.delivery?.remote_state?.delivery_failed;
|
|
const undeliverable_here = context.delivery?.remote_state?.undeliverable_here;
|
|
const err = new Error('[AMQP] message is released (delivery_failed=' + delivery_failed + ', undeliverable_here=' + undeliverable_here + ')');
|
|
processEvent(context, err);
|
|
});
|
|
sender.on('modified ', context => {
|
|
const delivery_failed = context.delivery?.remote_state?.delivery_failed;
|
|
const undeliverable_here = context.delivery?.remote_state?.undeliverable_here;
|
|
const err = new Error('[AMQP] message is modified (delivery_failed=' + delivery_failed + ', undeliverable_here=' + undeliverable_here + ')');
|
|
processEvent(context, err);
|
|
});
|
|
}
|
|
|
|
let init;
|
|
let addTask;
|
|
let addResponse;
|
|
let close;
|
|
let addDelayed;
|
|
let healthCheck;
|
|
if (commonDefines.c_oAscQueueType.rabbitmq === cfgQueueType) {
|
|
init = initRabbit;
|
|
addTask = addTaskRabbit;
|
|
addResponse = addResponseRabbit;
|
|
close = closeRabbit;
|
|
addDelayed = addDelayedRabbit;
|
|
healthCheck = healthCheckRabbit;
|
|
} else {
|
|
init = initActive;
|
|
addTask = addTaskActive;
|
|
addResponse = addResponseActive;
|
|
close = closeActive;
|
|
addDelayed = addDelayedActive;
|
|
healthCheck = healthCheckActive;
|
|
}
|
|
|
|
function TaskQueueRabbitMQ(simulateErrorResponse) {
|
|
this.isClose = false;
|
|
this.connection = null;
|
|
this.channelConvertTask = null;
|
|
this.channelConvertTaskReceive = null;
|
|
this.channelConvertDead = null;
|
|
this.channelConvertResponse = null;
|
|
this.channelConvertResponseReceive = null;
|
|
this.channelDelayed = null;
|
|
this.addTaskStore = [];
|
|
this.addDelayedStore = [];
|
|
this.channelConvertTaskData = {};
|
|
this.channelConvertResponseData = {};
|
|
this.channelDelayedData = {};
|
|
this.simulateErrorResponse = simulateErrorResponse;
|
|
}
|
|
util.inherits(TaskQueueRabbitMQ, events.EventEmitter);
|
|
TaskQueueRabbitMQ.prototype.init = function (isAddTask, isAddResponse, isAddTaskReceive, isAddResponseReceive, isEmitDead, isAddDelayed, callback) {
|
|
init(this, isAddTask, isAddResponse, isAddTaskReceive, isAddResponseReceive, isEmitDead, isAddDelayed, callback);
|
|
};
|
|
TaskQueueRabbitMQ.prototype.initPromise = function (isAddTask, isAddResponse, isAddTaskReceive, isAddResponseReceive, isEmitDead, isAddDelayed) {
|
|
const t = this;
|
|
return new Promise((resolve, reject) => {
|
|
init(t, isAddTask, isAddResponse, isAddTaskReceive, isAddResponseReceive, isEmitDead, isAddDelayed, err => {
|
|
if (err) {
|
|
reject(err);
|
|
} else {
|
|
resolve();
|
|
}
|
|
});
|
|
});
|
|
};
|
|
TaskQueueRabbitMQ.prototype.addTask = function (task, priority, opt_expiration, _opt_headers) {
|
|
task.setVisibilityTimeout(cfgVisibilityTimeout);
|
|
return addTaskString(this, JSON.stringify(task), priority, opt_expiration);
|
|
};
|
|
TaskQueueRabbitMQ.prototype.addResponse = function (task) {
|
|
const t = this;
|
|
return new Promise((resolve, reject) => {
|
|
const content = Buffer.from(JSON.stringify(task));
|
|
if (null != t.channelConvertResponse) {
|
|
addResponse(t, content, (err, _ok) => {
|
|
if (null != err) {
|
|
reject(err);
|
|
} else {
|
|
resolve();
|
|
}
|
|
});
|
|
} else {
|
|
resolve();
|
|
}
|
|
});
|
|
};
|
|
TaskQueueRabbitMQ.prototype.addDelayed = function (task, ttl) {
|
|
const t = this;
|
|
return new Promise((resolve, reject) => {
|
|
const content = new Buffer(JSON.stringify(task));
|
|
if (null != t.channelDelayed) {
|
|
addDelayed(t, content, ttl, (err, _ok) => {
|
|
if (null != err) {
|
|
reject(err);
|
|
} else {
|
|
resolve();
|
|
}
|
|
});
|
|
} else {
|
|
t.addDelayedStore.push({task: content, ttl});
|
|
resolve();
|
|
}
|
|
});
|
|
};
|
|
TaskQueueRabbitMQ.prototype.close = function () {
|
|
const t = this;
|
|
return co(function* () {
|
|
t.isClose = true;
|
|
if (t.channelConvertTask) {
|
|
yield close(t.channelConvertTask);
|
|
}
|
|
if (t.channelConvertTaskReceive) {
|
|
yield close(t.channelConvertTaskReceive);
|
|
}
|
|
if (t.channelConvertDead) {
|
|
yield close(t.channelConvertDead);
|
|
}
|
|
if (t.channelConvertResponse) {
|
|
yield close(t.channelConvertResponse);
|
|
}
|
|
if (t.channelConvertResponseReceive) {
|
|
yield close(t.channelConvertResponseReceive);
|
|
}
|
|
if (t.channelDelayed) {
|
|
yield close(t.channelDelayed);
|
|
}
|
|
yield close(t.connection);
|
|
});
|
|
};
|
|
TaskQueueRabbitMQ.prototype.closeOrWait = function () {
|
|
if (commonDefines.c_oAscQueueType.rabbitmq === cfgQueueType) {
|
|
return this.close();
|
|
} else {
|
|
//todo remove sleep
|
|
//sleep to wait acknowledge
|
|
return this.close().then(() => {
|
|
return utils.sleep(1000);
|
|
});
|
|
}
|
|
};
|
|
TaskQueueRabbitMQ.prototype.healthCheck = function () {
|
|
return healthCheck(this);
|
|
};
|
|
|
|
module.exports = TaskQueueRabbitMQ;
|