Dollerok/P8-ExchangeService
1
0
Fork 0
forked from CITKParus/P8-ExchangeService
Code Issues Pull Requests Projects Releases Wiki Activity

Переработан опрос очереди и обработка сообщений - теперь статусы передвигаются корректно, нет задвоений при опросе очереди

Browse Source
This commit is contained in:
Mikhail Chechnev 2018-12-05 22:30:30 +03:00
parent 2a3ba01e40
commit 5a45ae6663
1 changed files with 316 additions and 92 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

384
core/out_queue.js
View File

@ -13,7 +13,7 @@ const ChildProcess = require("child_process"); //Работа с дочерни
const { ServerError } = require("./server_errors"); //Типовая ошибка const { ServerError } = require("./server_errors"); //Типовая ошибка
const { SERR_UNEXPECTED, SERR_OBJECT_BAD_INTERFACE } = require("./constants"); //Общесистемные константы const { SERR_UNEXPECTED, SERR_OBJECT_BAD_INTERFACE } = require("./constants"); //Общесистемные константы
const { validateObject } = require("./utils"); //Вспомогательные функции const { validateObject } = require("./utils"); //Вспомогательные функции
const { NINC_EXEC_CNT_YES } = require("../models/prms_db_connector"); //Схемы валидации параметров функций модуля взаимодействия с БД const { NINC_EXEC_CNT_YES, NINC_EXEC_CNT_NO } = require("../models/prms_db_connector"); //Схемы валидации параметров функций модуля взаимодействия с БД
const objOutQueueProcessorSchema = require("../models/obj_out_queue_processor"); //Схемы валидации сообщений обмена с обработчиком сообщения очереди const objOutQueueProcessorSchema = require("../models/obj_out_queue_processor"); //Схемы валидации сообщений обмена с обработчиком сообщения очереди
const objQueueSchema = require("../models/obj_queue"); //Схемы валидации сообщения очереди const objQueueSchema = require("../models/obj_queue"); //Схемы валидации сообщения очереди
const prmsOutQueueSchema = require("../models/prms_out_queue"); //Схемы валидации параметров функций класса const prmsOutQueueSchema = require("../models/prms_out_queue"); //Схемы валидации параметров функций класса
@ -76,13 +76,118 @@ class OutQueue extends EventEmitter {
//оповестим подписчиков о появлении нового отчета //оповестим подписчиков о появлении нового отчета
this.emit(SEVT_OUT_QUEUE_STOPPED); this.emit(SEVT_OUT_QUEUE_STOPPED);
} }
//Останов обработчика сообщения //Установка финальных статусов сообщения в БД
stopMessageWorker(worker) { async finalise(prms) {
worker.kill(); //Проверяем структуру переданного объекта для старта
this.nWorkersLeft++; let sCheckResult = validateObject(
prms,
prmsOutQueueSchema.finalise,
"Параметры функции установки финальных статусов сообщения в БД"
);
//Если структура объекта в норме
if (!sCheckResult) {
//Если больше нет попыток исполнения и сообщение не в статусе успешной обработки сервером БД
if (
prms.queue.nExecState != objQueueSchema.NQUEUE_EXEC_STATE_DB_OK &&
prms.queue.nExecCnt >= prms.queue.nRetryAttempts
) {
//То считаем, что оно выполнено с ошибками и больше пытаться не надо
await this.dbConn.setQueueState({
nQueueId: prms.queue.nId,
sExecMsg: prms.queue.sExecMsg,
nExecState: objQueueSchema.NQUEUE_EXEC_STATE_ERR
});
} else {
//Если сообщение успешно исполнено сервером БД - то значит оно успешно исполнено вообще
if (prms.queue.nExecState == objQueueSchema.NQUEUE_EXEC_STATE_DB_OK) {
await this.dbConn.setQueueState({
nQueueId: prms.queue.nId,
nIncExecCnt: prms.queue.nExecCnt == 0 ? NINC_EXEC_CNT_YES : NINC_EXEC_CNT_NO,
nExecState: objQueueSchema.NQUEUE_EXEC_STATE_OK
});
} else {
//Если сообщение в статусе исполнения сервером приложений (чего здесь быть не может) - это ошибка
if (prms.queue.nExecState == objQueueSchema.NQUEUE_EXEC_STATE_APP) {
//То выставим ему ошибку исполнения сервером приложений
await this.dbConn.setQueueState({
nQueueId: prms.queue.nId,
sExecMsg: prms.queue.sExecMsg,
nIncExecCnt: prms.queue.nExecCnt == 0 ? NINC_EXEC_CNT_YES : NINC_EXEC_CNT_NO,
nExecState: objQueueSchema.NQUEUE_EXEC_STATE_APP_ERR
});
} else {
//Если сообщение в статусе исполнения сервером БД (чего здесь быть не может) - это ошибка
if (prms.queue.nExecState == objQueueSchema.NQUEUE_EXEC_STATE_DB) {
//То выставим ему ошибку исполнения сервером БД
await this.dbConn.setQueueState({
nQueueId: prms.queue.nId,
sExecMsg: prms.queue.sExecMsg,
nIncExecCnt: prms.queue.nExecCnt == 0 ? NINC_EXEC_CNT_YES : NINC_EXEC_CNT_NO,
nExecState: objQueueSchema.NQUEUE_EXEC_STATE_DB_ERR
});
}
}
}
}
} else {
throw new ServerError(SERR_OBJECT_BAD_INTERFACE, sCheckResult);
}
}
//Запуск обработки сообщения сервером БД
async dbProcess(prms) {
//Проверяем структуру переданного объекта для старта
let sCheckResult = validateObject(
prms,
prmsOutQueueSchema.dbProcess,
"Параметры функции запуска обработки ообщения сервером БД"
);
//Если структура объекта в норме
if (!sCheckResult) {
//Буфер для текущего состояния сообщения
let curQueue = null;
//Обрабатываем
try {
//Фиксируем начало исполнения сервером БД - в статусе сообщения
curQueue = await this.dbConn.setQueueState({
nQueueId: prms.queue.nId,
nExecState: objQueueSchema.NQUEUE_EXEC_STATE_DB
});
//Вызов обработчика БД
curQueue = await this.dbConn.execQueueDBPrc({ nQueueId: prms.queue.nId });
//Фиксируем успешное исполнение сервером БД - в статусе сообщения
curQueue = await this.dbConn.setQueueState({
nQueueId: prms.queue.nId,
nExecState: objQueueSchema.NQUEUE_EXEC_STATE_DB_OK
});
//Фиксируем успешное исполнение сервером БД - в протоколе работы сервиса
await this.logger.info(`Исходящее сообщение ${prms.queue.nId} успешно отработано сервером БД`, {
nQueueId: prms.queue.nId
});
} catch (e) {
//Сформируем текст ошибки
let sErr = `${SERR_UNEXPECTED}: ${e.message}`;
if (e instanceof ServerError) sErr = `${e.sCode}: ${e.sMessage}`;
//Фиксируем ошибку обработки сервером БД - в статусе сообщения
curQueue = await this.dbConn.setQueueState({
nQueueId: prms.queue.nId,
sExecMsg: sErr,
nIncExecCnt: NINC_EXEC_CNT_YES,
nExecState: objQueueSchema.NQUEUE_EXEC_STATE_DB_ERR
});
//Фиксируем ошибку обработки сервером БД - в протоколе работы сервиса
await this.logger.error(
`Ошибка обработки исходящего сообщения ${prms.queue.nId} сервером БД: ${sErr}`,
{ nQueueId: prms.queue.nId }
);
}
//Вернём текущее состоянии сообщения очереди
return curQueue;
} else {
throw new ServerError(SERR_OBJECT_BAD_INTERFACE, sCheckResult);
}
} }
//Запуск обработки очередного сообщения //Запуск обработки очередного сообщения
processMessage(prms) { async processMessage(prms) {
//Проверяем структуру переданного объекта для старта //Проверяем структуру переданного объекта для старта
let sCheckResult = validateObject( let sCheckResult = validateObject(
prms, prms,
@ -97,6 +202,32 @@ class OutQueue extends EventEmitter {
const self = this; const self = this;
//Создаём новый обработчик сообщений //Создаём новый обработчик сообщений
const proc = ChildProcess.fork("core/out_queue_processor", { silent: false }); const proc = ChildProcess.fork("core/out_queue_processor", { silent: false });
//Текущее состояние сообщения
let curQueue = null;
//Скажем что начали обработку
await self.logger.info(
`Обрабатываю исходящее сообщение: ${prms.queue.nId}, ${prms.queue.sInDate}, ${
prms.queue.sServiceFnCode
}, ${prms.queue.sExecState}, попытка исполнения - ${prms.queue.nExecCnt + 1}`,
{ nQueueId: prms.queue.nId }
);
//Установим его статус в БД - обрабатывается сервером приложений (только для новых или повторно обрабатываемых сервером приложений)
if (
prms.queue.nExecState == objQueueSchema.NQUEUE_EXEC_STATE_INQUEUE ||
prms.queue.nExecState == objQueueSchema.NQUEUE_EXEC_STATE_APP_ERR
) {
curQueue = await self.dbConn.setQueueState({
nQueueId: prms.queue.nId,
nExecState: objQueueSchema.NQUEUE_EXEC_STATE_APP
});
}
//Установим его статус в БД - обрабатывается в БД (только если сюда пришло сообщение на повторную обработку сервером БД)
if (prms.queue.nExecState == objQueueSchema.NQUEUE_EXEC_STATE_DB_ERR) {
curQueue = await self.dbConn.setQueueState({
nQueueId: prms.queue.nId,
nExecState: objQueueSchema.NQUEUE_EXEC_STATE_DB
});
}
//Перехват сообщений обработчика //Перехват сообщений обработчика
proc.on("message", async result => { proc.on("message", async result => {
//Проверяем структуру полученного сообщения //Проверяем структуру полученного сообщения
@ -107,10 +238,14 @@ class OutQueue extends EventEmitter {
); );
//Если структура сообщения в норме //Если структура сообщения в норме
if (!sCheckResult) { if (!sCheckResult) {
//Если обработчик вернул ошибку //Движение события по статусам в зависимости от того в каком состоянии его вернул обработчик
if (result.nExecState == objQueueSchema.NQUEUE_EXEC_STATE_APP_ERR) { try {
//Работаем от статуса сообщения полученного от обработчика
switch (result.nExecState) {
//Ошибка обработки
case objQueueSchema.NQUEUE_EXEC_STATE_APP_ERR: {
//Установим ошибочный статус в БД для сообщений и увеличим счетчик попыток отправки //Установим ошибочный статус в БД для сообщений и увеличим счетчик попыток отправки
await self.dbConn.setQueueState({ curQueue = await self.dbConn.setQueueState({
nQueueId: prms.queue.nId, nQueueId: prms.queue.nId,
sExecMsg: result.sExecMsg, sExecMsg: result.sExecMsg,
nIncExecCnt: NINC_EXEC_CNT_YES, nIncExecCnt: NINC_EXEC_CNT_YES,
@ -118,89 +253,164 @@ class OutQueue extends EventEmitter {
}); });
//Фиксируем ошибку в протоколе работы сервиса //Фиксируем ошибку в протоколе работы сервиса
await self.logger.error( await self.logger.error(
`Ошибка обработки исходящего сообщения ${prms.queue.nId} сервером приложений: ` + `Ошибка обработки исходящего сообщения ${prms.queue.nId} сервером приложений: ${
result.sExecMsg, result.sExecMsg
{ }`,
nQueueId: prms.queue.nId { nQueueId: prms.queue.nId }
}
); );
} else { break;
//Пишем в базу успех }
await self.dbConn.setQueueState({ //Успех обработки
case objQueueSchema.NQUEUE_EXEC_STATE_APP_OK: {
//Если состояние менялось (а не просто повторная отработка)
if (result.nExecState != prms.queue.nExecState) {
//Пишем в базу успех отработки сервером приложений - результаты обработки
curQueue = await self.dbConn.setQueueAppSrvResult({
nQueueId: prms.queue.nId,
blMsg: result.blMsg ? new Buffer(result.blMsg) : null,
blResp: result.blResp ? new Buffer(result.blResp) : null
});
//Пишем в базу успех отработки сервером приложений - статус сообщения
curQueue = await self.dbConn.setQueueState({
nQueueId: prms.queue.nId, nQueueId: prms.queue.nId,
sExecMsg: result.sExecMsg,
nIncExecCnt: NINC_EXEC_CNT_YES,
nExecState: result.nExecState nExecState: result.nExecState
}); });
//Фиксируем успех в протоколе работы сервиса //Пишем в базу успех отработки сервером приложений - запись в протокол работы сервера приложений
await self.logger.info( await self.logger.info(
`Исходящее сообщение ${prms.queue.nId} успешно отработано сервером приложений`, `Исходящее сообщение ${
{ prms.queue.nId
nQueueId: prms.queue.nId } успешно отработано сервером приложений`,
} { nQueueId: prms.queue.nId }
); );
} }
//Запускаем обработку сервером БД
curQueue = await self.dbProcess(prms);
break;
}
//Обработчик ничего не делал
default: {
//Обработчик ничего не делал, но если сообщение сообщение в статусе - ошибка обработки сервером БД или обрабатывается сервером БД, то запустим обработчик БД
if (result.nExecState == objQueueSchema.NQUEUE_EXEC_STATE_DB_ERR) {
//Запускаем обработчик сервера БД
curQueue = await self.dbProcess(prms);
} else { } else {
//Пришел неожиданный ответ обработчика, установим статус в БД - ошибка обработки сервером приложений //Во всех остальных случаях - ничего не делаем вообще
await self.dbConn.setQueueState({ curQueue = prms.queue;
}
break;
}
}
} catch (e) {
//Сформируем текст ошибки
let sErr = `${SERR_UNEXPECTED}: ${e.message}`;
if (e instanceof ServerError) sErr = `${e.sCode}: ${e.sMessage}`;
//Фиксируем ошибку обработки сервером приложений - статус сообщения
curQueue = await self.dbConn.setQueueState({
nQueueId: prms.queue.nId,
sExecMsg: sErr,
nIncExecCnt: NINC_EXEC_CNT_YES
});
//Фиксируем ошибку обработки сервером приложений - запись в протокол работы сервера приложений
await self.logger.error(sErr, { nQueueId: prms.queue.nId });
}
} else {
//Пришел неожиданный ответ обработчика - статус сообщения
curQueue = await self.dbConn.setQueueState({
nQueueId: prms.queue.nId, nQueueId: prms.queue.nId,
sExecMsg: sCheckResult, sExecMsg: sCheckResult,
nIncExecCnt: NINC_EXEC_CNT_YES, nIncExecCnt: NINC_EXEC_CNT_YES
nExecState: objQueueSchema.NQUEUE_EXEC_STATE_APP_ERR
}); });
//Фиксируем ошибку в протоколе работы сервиса //Пришел неожиданный ответ обработчика - запись в протокол работы сервера приложений
await self.logger.error( await self.logger.error(
`Неожиданный ответ обработчика для сообщения ${prms.queue.nId}: ${sCheckResult}`, `Неожиданный ответ обработчика для сообщения ${prms.queue.nId}: ${sCheckResult}`,
{ { nQueueId: prms.queue.nId }
nQueueId: prms.queue.nId
}
); );
} }
self.stopMessageWorker(proc); //Выставляем финальные статусы
try {
await self.finalise({ queue: curQueue });
} catch (e) {
//Сформируем текст ошибки
let sErr = `${SERR_UNEXPECTED}: ${e.message}`;
if (e instanceof ServerError) sErr = `${e.sCode}: ${e.sMessage}`;
//Установим его статус в БД - ошибка установки финального статуса
await self.dbConn.setQueueState({
nQueueId: prms.queue.nId,
sExecMsg: sErr,
nExecState: objQueueSchema.NQUEUE_EXEC_STATE_ERR
});
//Пришел неожиданный ответ обработчика - запись в протокол работы сервера приложений
await self.logger.error(`Фатальная ошибка обработчика сообщения ${prms.queue.nId}: ${sErr}`, {
nQueueId: prms.queue.nId
});
}
//Останавливаем обработчик и инкрементируем флаг их доступного количества
proc.kill();
this.nWorkersLeft++;
}); });
//Перехват ошибок обработчика //Перехват ошибок обработчика
proc.on("error", async e => { proc.on("error", async e => {
//Сформируем текст ошибки
let sErr = `${SERR_UNEXPECTED}: ${e.message}`;
if (e instanceof ServerError) sErr = `${e.sCode}: ${e.sMessage}`;
//Установим его статус в БД - ошибка обработки сервером приложений //Установим его статус в БД - ошибка обработки сервером приложений
await self.dbConn.setQueueState({ let curQueue = await self.dbConn.setQueueState({
nQueueId: prms.queue.nId, nQueueId: prms.queue.nId,
sExecMsg: e.message, sExecMsg: sErr,
nIncExecCnt: NINC_EXEC_CNT_YES, nIncExecCnt: NINC_EXEC_CNT_YES
nExecState: objQueueSchema.NQUEUE_EXEC_STATE_APP_ERR
}); });
//Так же фиксируем ошибку в протоколе работы //Так же фиксируем ошибку в протоколе работы
await self.logger.error(`Ошибка обработки исходящего сообщения сервером приложений: ${e.message}`, { await self.logger.error(`Ошибка обработки исходящего сообщения сервером приложений: ${sErr}`, {
nQueueId: prms.queue.nId nQueueId: prms.queue.nId
}); });
//Завершим обработчик //Выставляем финальные статусы
self.stopMessageWorker(proc); try {
await self.finalise({ queue: curQueue });
} catch (e) {
//Сформируем текст ошибки
let sErr = `${SERR_UNEXPECTED}: ${e.message}`;
if (e instanceof ServerError) sErr = `${e.sCode}: ${e.sMessage}`;
//Установим его статус в БД - ошибка установки финального статуса
await self.dbConn.setQueueState({
nQueueId: prms.queue.nId,
sExecMsg: sErr,
nExecState: objQueueSchema.NQUEUE_EXEC_STATE_ERR
});
//Пришел неожиданный ответ обработчика - запись в протокол работы сервера приложений
await self.logger.error(`Фатальная ошибка обработчика сообщения ${prms.queue.nId}: ${sErr}`, {
nQueueId: prms.queue.nId
});
}
//Останавливаем обработчик и инкрементируем флаг их доступного количества
proc.kill();
this.nWorkersLeft++;
}); });
//Перехват останова обработчика //Перехват останова обработчика
proc.on("exit", code => {}); proc.on("exit", code => {});
//Запускаем обработчик //Запускаем обработчик
proc.send({ proc.send({
nQueueId: prms.queue.nId, nQueueId: prms.queue.nId,
service: {}, nExecState: prms.queue.nExecState,
function: {}, blMsg: prms.queue.blMsg,
blMsg: prms.queue.blMsg blResp: prms.queue.blResp,
service: _.find(this.services, { nId: prms.queue.nServiceId }),
function: _.find(_.find(this.services, { nId: prms.queue.nServiceId }).functions, {
nId: prms.queue.nServiceFnId
})
}); });
//Уменьшаем количество доступных обработчиков //Уменьшаем количество доступных обработчиков
this.nWorkersLeft--; this.nWorkersLeft--;
//Вернем признак того, что сообщение обрабатывается
return true;
} else {
//Вернем признак того, что сообщение не обрабатывается
return false;
} }
} else { } else {
throw new ServerError(SERR_OBJECT_BAD_INTERFACE, sCheckResult); throw new ServerError(SERR_OBJECT_BAD_INTERFACE, sCheckResult);
} }
} }
//Перезапуск опроса очереди исходящих сообщений //Перезапуск опроса очереди исходящих сообщений
restartDetectingLoop() { async restartDetectingLoop() {
//Включаем опрос очереди только если установлен флаг работы //Включаем опрос очереди только если установлен флаг работы
if (this.bWorking) { if (this.bWorking) {
this.nDetectingLoopTimeOut = setTimeout(() => { this.nDetectingLoopTimeOut = await setTimeout(async () => {
this.outDetectingLoop(); await this.outDetectingLoop();
}, this.outGoing.nCheckTimeout); }, this.outGoing.nCheckTimeout);
} }
} }
@ -214,44 +424,58 @@ class OutQueue extends EventEmitter {
let outMsgs = await this.dbConn.getOutgoing({ nPortionSize: this.nWorkersLeft }); let outMsgs = await this.dbConn.getOutgoing({ nPortionSize: this.nWorkersLeft });
//Если есть сообщения //Если есть сообщения
if (Array.isArray(outMsgs) && outMsgs.length > 0) { if (Array.isArray(outMsgs) && outMsgs.length > 0) {
//Ставим их в очередь //Обходим их
for (let i = 0; i < outMsgs.length; i++) { for (let i = 0; i < outMsgs.length; i++) {
//Если сообщение успешно взято в обработку //И запускаем обработчики
if (this.processMessage({ queue: outMsgs[i] })) { try {
//Скажем что оно у нас есть await this.processMessage({ queue: outMsgs[i] });
await this.logger.info( } catch (e) {
"Исполняю отправку исходящего сообщения: " + //Какие непредвиденные ошибки при обработке текущего сообщения - подготовим текст ошибки
outMsgs[i].nId + let sErr = `${SERR_UNEXPECTED}: ${e.message}`;
", " + if (e instanceof ServerError) sErr = `${e.sCode}: ${e.sMessage}`;
outMsgs[i].sInDate + //Фиксируем ошибку обработки сервером приложений - статус сообщения (сам статус - не меняем, здесь только фатальные ошибки, но делаем инкремент количества попыток)
", " + let curQueue = await this.dbConn.setQueueState({
outMsgs[i].sServiceFnCode + nQueueId: outMsgs[i].nId,
", " + sExecMsg: sErr,
outMsgs[i].sExecState + nIncExecCnt: NINC_EXEC_CNT_YES
", попытка исполнения - " + });
(outMsgs[i].nExecCnt + 1), //Фиксируем ошибку обработки сервером приложений - запись в протокол работы сервера приложений
await this.logger.error(sErr, { nQueueId: outMsgs[i].nId });
//Выставляем финальные статусы
try {
await this.finalise({ queue: curQueue });
} catch (e) {
//Сформируем текст ошибки
let sErr = `${SERR_UNEXPECTED}: ${e.message}`;
if (e instanceof ServerError) sErr = `${e.sCode}: ${e.sMessage}`;
//Установим его статус в БД - ошибка установки финального статуса
await self.dbConn.setQueueState({
nQueueId: outMsgs[i].nId,
sExecMsg: sErr,
nExecState: objQueueSchema.NQUEUE_EXEC_STATE_ERR
});
//Пришел неожиданный ответ обработчика - запись в протокол работы сервера приложений
await self.logger.error(
`Фатальная ошибка обработчика сообщения ${outMsgs[i].nId}: ${sErr}`,
{ nQueueId: outMsgs[i].nId } { nQueueId: outMsgs[i].nId }
); );
//Установим его статус в БД - обрабатывается сервером приложений
await this.dbConn.setQueueState({
nQueueId: outMsgs[i].nId,
nExecState: objQueueSchema.NQUEUE_EXEC_STATE_APP
});
} }
} }
} else {
await this.logger.info("Нет новых сообщений");
} }
this.restartDetectingLoop(); }
//Запустили отработку всех считанных - перезапускаем цикл опроса исходящих сообщений
await this.restartDetectingLoop();
} catch (e) { } catch (e) {
if (e instanceof ServerError) //Какие непредвиденные ошибки при получении списка сообщений - подготовим текст ошибки
await this.logger.error("При обработке исходящего сообщения: " + e.sCode + ": " + e.sMessage); let sErr = `${SERR_UNEXPECTED}: ${e.message}`;
else this.logger.error(SERR_UNEXPECTED + ": " + e.message); if (e instanceof ServerError) sErr = `${e.sCode}: ${e.sMessage}`;
this.restartDetectingLoop(); //Фиксируем ошибку в протоколе работы сервера приложений
await this.logger.error(sErr);
await this.restartDetectingLoop();
} }
} else { } else {
await this.logger.info("Нет свободных обработчиков"); //Нет свободных обработчиков - ждём и перезапускаем цикл опроса
this.restartDetectingLoop(); await this.restartDetectingLoop();
} }
} }
//Запуск обработки очереди исходящих сообщений //Запуск обработки очереди исходящих сообщений