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 2019-01-04 14:38:25 +03:00
parent d92d85f4f6
commit 01a3e88e6b
5 changed files with 262 additions and 351 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
core/constants.js
View File

@ -15,6 +15,7 @@ exports.SMODULES_PATH_MODELS = "@models"; //Модели данных и схе
//Типовые коды ошибок //Типовые коды ошибок
exports.SERR_COMMON = "ERR_COMMON"; //Общая ошибка exports.SERR_COMMON = "ERR_COMMON"; //Общая ошибка
exports.SERR_UNEXPECTED = "ERR_UNEXPECTED"; //Неожиданная ошибка exports.SERR_UNEXPECTED = "ERR_UNEXPECTED"; //Неожиданная ошибка
exports.SERR_UNAUTH = "ERR_UNAUTH"; //Отсутствие аутентификации
//Типовые коды ошибок подключения модулей //Типовые коды ошибок подключения модулей
exports.SERR_MODULES_NO_MODULE_SPECIFIED = "ERR_MODULES_NO_MODULE_SPECIFIED"; //Не указан подключаемый модуль exports.SERR_MODULES_NO_MODULE_SPECIFIED = "ERR_MODULES_NO_MODULE_SPECIFIED"; //Не указан подключаемый модуль

49
core/out_queue.js
View File

@ -195,6 +195,9 @@ class OutQueue extends EventEmitter {
const self = this; const self = this;
//Запоминаем текущее количество попыток обработки //Запоминаем текущее количество попыток обработки
const nQueueOldExecCnt = prms.queue.nExecCnt; const nQueueOldExecCnt = prms.queue.nExecCnt;
//Буфер для ошибок (для журнала работы и очереди обмена)
let sErrorLog = null;
let sError = null;
//Создаём новый обработчик сообщений //Создаём новый обработчик сообщений
const proc = ChildProcess.fork("core/out_queue_processor", { silent: false }); const proc = ChildProcess.fork("core/out_queue_processor", { silent: false });
//Перехват сообщений обработчика //Перехват сообщений обработчика
@ -211,40 +214,30 @@ class OutQueue extends EventEmitter {
if (!sCheckResult) { if (!sCheckResult) {
//Анализируем результат обработки - если ошибка - фиксируем //Анализируем результат обработки - если ошибка - фиксируем
if (result.sResult == objOutQueueProcessorSchema.STASK_RESULT_ERR) { if (result.sResult == objOutQueueProcessorSchema.STASK_RESULT_ERR) {
//Фиксируем ошибку обработки - протокол работы сервиса //Запоминаем ошибку обработчика
await self.logger.error(`Ошибка обработки исходящего сообщения: ${result.sMsg}`, { sErrorLog = `Ошибка обработки исходящего сообщения: ${result.sMsg}`;
nQueueId: prms.queue.nId sError = result.sMsg;
});
//Фиксируем ошибку обработки - статус сообщения
await this.dbConn.setQueueState({
nQueueId: prms.queue.nId,
sExecMsg: result.sMsg,
nIncExecCnt:
nQueueOldExecCnt == prms.queue.nExecCnt ? NINC_EXEC_CNT_YES : NINC_EXEC_CNT_NO,
nExecState:
(nQueueOldExecCnt == prms.queue.nExecCnt
? prms.queue.nExecCnt + 1
: prms.queue.nExecCnt) < prms.queue.nRetryAttempts
? prms.queue.nExecState
: objQueueSchema.NQUEUE_EXEC_STATE_ERR
});
} else { } else {
//Ошибки нет, но если есть контекст для сервиса - сохраним его для дальнейшего использования //Ошибки обработки нет, но может быть есть ошибка аутентификации
if (!_.isUndefined(result.context)) { if (result.sResult == objOutQueueProcessorSchema.STASK_RESULT_UNAUTH) {
let tmpSrv = _.find(this.services, { nId: prms.queue.nServiceId }); //!!!!!!!!!!!!!!
tmpSrv.context = _.cloneDeep(result.context); //??????????????
//!!!!!!!!!!!!!!
} }
} }
} else { } else {
//Пришел неожиданный ответ обработчика - запись в протокол работы сервиса //Пришел неожиданный ответ обработчика
await self.logger.error( sErrorLog = `Неожиданный ответ обработчика для сообщения ${prms.queue.nId}: ${sCheckResult}`;
`Неожиданный ответ обработчика для сообщения ${prms.queue.nId}: ${sCheckResult}`, sError = sCheckResult;
{ nQueueId: prms.queue.nId } }
); //Фиксируем ошибки, если есть
//Фиксируем ошибку обработки - статус сообщения if (sError) {
//Запись в протокол работы сервиса
await self.logger.error(sErrorLog, { nQueueId: prms.queue.nId });
//Запись в статус сообщения
await this.dbConn.setQueueState({ await this.dbConn.setQueueState({
nQueueId: prms.queue.nId, nQueueId: prms.queue.nId,
sExecMsg: `Неожиданный ответ обработчика для сообщения ${prms.queue.nId}: ${sCheckResult}`, sExecMsg: sError,
nIncExecCnt: nQueueOldExecCnt == prms.queue.nExecCnt ? NINC_EXEC_CNT_YES : NINC_EXEC_CNT_NO, nIncExecCnt: nQueueOldExecCnt == prms.queue.nExecCnt ? NINC_EXEC_CNT_YES : NINC_EXEC_CNT_NO,
nExecState: nExecState:
(nQueueOldExecCnt == prms.queue.nExecCnt (nQueueOldExecCnt == prms.queue.nExecCnt

511
core/out_queue_processor.js
View File

@ -17,8 +17,14 @@ const { ServerError } = require("./server_errors"); //Типовая ошибк
const objOutQueueProcessorSchema = require("../models/obj_out_queue_processor"); //Схема валидации сообщений обмена с бработчиком очереди исходящих сообщений const objOutQueueProcessorSchema = require("../models/obj_out_queue_processor"); //Схема валидации сообщений обмена с бработчиком очереди исходящих сообщений
const prmsOutQueueProcessorSchema = require("../models/prms_out_queue_processor"); //Схема валидации параметров функций модуля const prmsOutQueueProcessorSchema = require("../models/prms_out_queue_processor"); //Схема валидации параметров функций модуля
const objQueueSchema = require("../models/obj_queue"); //Схемы валидации сообщения очереди const objQueueSchema = require("../models/obj_queue"); //Схемы валидации сообщения очереди
const objServiceSchema = require("../models/obj_service"); //Схемы валидации сервиса
const objServiceFnSchema = require("../models/obj_service_function"); //Схемы валидации функции сервиса const objServiceFnSchema = require("../models/obj_service_function"); //Схемы валидации функции сервиса
const { SERR_OBJECT_BAD_INTERFACE, SERR_APP_SERVER_BEFORE, SERR_APP_SERVER_AFTER } = require("./constants"); //Глобальные константы const {
SERR_OBJECT_BAD_INTERFACE,
SERR_APP_SERVER_BEFORE,
SERR_APP_SERVER_AFTER,
SERR_UNAUTH
} = require("./constants"); //Глобальные константы
const { NINC_EXEC_CNT_YES, NINC_EXEC_CNT_NO } = require("../models/prms_db_connector"); //Схемы валидации параметров функций модуля взаимодействия с БД const { NINC_EXEC_CNT_YES, NINC_EXEC_CNT_NO } = require("../models/prms_db_connector"); //Схемы валидации параметров функций модуля взаимодействия с БД
//-------------------------- //--------------------------
@ -44,40 +50,36 @@ const sendErrorResult = prms => {
if (!sCheckResult) { if (!sCheckResult) {
process.send({ process.send({
sResult: objOutQueueProcessorSchema.STASK_RESULT_ERR, sResult: objOutQueueProcessorSchema.STASK_RESULT_ERR,
sMsg: prms.sMessage, sMsg: prms.sMessage
context: null
}); });
} else { } else {
process.send({ process.send({
sResult: objOutQueueProcessorSchema.STASK_RESULT_ERR, sResult: objOutQueueProcessorSchema.STASK_RESULT_ERR,
sMsg: sCheckResult, sMsg: sCheckResult
context: null
}); });
} }
}; };
//Отправка родительскому процессу успеха обработки сообщения сервером приложений //Отправка родительскому процессу успеха обработки сообщения сервером приложений
const sendOKResult = prms => { const sendOKResult = () => {
//Проверяем структуру переданного сообщения process.send({
let sCheckResult = validateObject( sResult: objOutQueueProcessorSchema.STASK_RESULT_OK,
prms, sMsg: null
prmsOutQueueProcessorSchema.sendOKResult, });
"Параметры функции отправки родительскому процессу успеха обработки сообщения" };
);
//Если структура объекта в норме //Отправка родительскому процессу успеха обработки сообщения сервером приложений
if (!sCheckResult) { const sendUnAuthResult = () => {
process.send({ process.send({
sResult: objOutQueueProcessorSchema.STASK_RESULT_OK, sResult: objOutQueueProcessorSchema.STASK_RESULT_UNAUTH,
sMsg: null, sMsg: null
context: prms.context });
});
} else {
sendErrorResult({ sMessage: sCheckResult });
}
}; };
//Запуск обработки сообщения сервером приложений //Запуск обработки сообщения сервером приложений
const appProcess = async prms => { const appProcess = async prms => {
//Результат обработки - объект Queue (обработанное сообщение) или ServerError (ошибка обработки)
let res = null;
//Проверяем структуру переданного объекта для старта //Проверяем структуру переданного объекта для старта
let sCheckResult = validateObject( let sCheckResult = validateObject(
prms, prms,
@ -86,145 +88,157 @@ const appProcess = async prms => {
); );
//Если структура объекта в норме //Если структура объекта в норме
if (!sCheckResult) { if (!sCheckResult) {
//Обработанное сообщение
let newQueue = null;
//Обрабатываем //Обрабатываем
try { try {
//Фиксируем начало исполнения сервером приложений - в статусе сообщения //Считываем статус аутентификации сервиса
newQueue = await dbConn.setQueueState({ let isServiceAuth = await dbConn.isServiceAuth({ nServiceId: prms.service.nId });
nQueueId: prms.queue.nId, //Проверяем аутентификацию
nExecState: objQueueSchema.NQUEUE_EXEC_STATE_APP if (
}); prms.function.nAuthOnly == objServiceFnSchema.NAUTH_ONLY_NO ||
//Скажем что начали обработку (prms.function.nAuthOnly == objServiceFnSchema.NAUTH_ONLY_YES &&
await logger.info( isServiceAuth == objServiceSchema.NIS_AUTH_YES)
`Обрабатываю исходящее сообщение сервером приложений: ${prms.queue.nId}, ${prms.queue.sInDate}, ${ ) {
prms.queue.sServiceFnCode //Фиксируем начало исполнения сервером приложений - в статусе сообщения
}, ${prms.queue.sExecState}, попытка исполнения - ${prms.queue.nExecCnt + 1}`, res = await dbConn.setQueueState({
{ nQueueId: prms.queue.nId } nQueueId: prms.queue.nId,
); nExecState: objQueueSchema.NQUEUE_EXEC_STATE_APP
//Считаем тело сообщения
let qData = await dbConn.getQueueMsg({ nQueueId: prms.queue.nId });
//Кладём данные тела в объект сообщения и инициализируем поле для ответа
_.extend(prms.queue, { blMsg: qData.blMsg, blResp: null });
//Собираем параметры для передачи серверу
let options = { method: prms.service.sFnPrmsType };
//Определимся с URL и телом сообщения в зависимости от способа передачи параметров
if (prms.service.sFnPrmsType == objServiceFnSchema.NFN_PRMS_TYPE_POST) {
options.url = buildURL({ sSrvRoot: prms.service.sSrvRoot, sFnURL: prms.function.sFnURL });
options.body = prms.queue.blMsg;
} else {
options.url = buildURL({
sSrvRoot: prms.service.sSrvRoot,
sFnURL: prms.function.sFnURL,
sQuery: prms.queue.blMsg.toString()
}); });
} //Фиксируем начало исполнения сервером приложений - в протоколе работы сервиса
//Выполняем обработчик "До" (если он есть) await logger.info(
if (prms.function.sAppSrvBefore) { `Обрабатываю исходящее сообщение сервером приложений: ${prms.queue.nId}, ${prms.queue.sInDate}, ${
const fnBefore = getAppSrvFunction(prms.function.sAppSrvBefore); prms.queue.sServiceFnCode
let resBefore = null; }, ${prms.queue.sExecState}, попытка исполнения - ${prms.queue.nExecCnt + 1}`,
try { { nQueueId: prms.queue.nId }
let resBeforePrms = _.cloneDeep(prms); );
resBefore = await fnBefore(resBeforePrms); //Считаем тело сообщения
} catch (e) { let qData = await dbConn.getQueueMsg({ nQueueId: prms.queue.nId });
throw new ServerError(SERR_APP_SERVER_BEFORE, e.message); //Считаем контекст сервиса
let serviceCtx = await dbConn.getServiceContext({ nServiceId: prms.service.nId });
//Кладём данные тела в объект сообщения и инициализируем поле для ответа
_.extend(prms.queue, { blMsg: qData.blMsg, blResp: null });
//Кладём данные контекста в сервис
_.extend(prms.service, serviceCtx);
//Собираем параметры для передачи серверу
let options = { method: prms.service.sFnPrmsType };
//Определимся с URL и телом сообщения в зависимости от способа передачи параметров
if (prms.service.sFnPrmsType == objServiceFnSchema.NFN_PRMS_TYPE_POST) {
options.url = buildURL({ sSrvRoot: prms.service.sSrvRoot, sFnURL: prms.function.sFnURL });
options.body = prms.queue.blMsg;
} else {
options.url = buildURL({
sSrvRoot: prms.service.sSrvRoot,
sFnURL: prms.function.sFnURL,
sQuery: prms.queue.blMsg.toString()
});
} }
//Проверяем структуру ответа функции предобработки //Выполняем обработчик "До" (если он есть)
if (resBefore) { if (prms.function.sAppSrvBefore) {
let sCheckResult = validateObject( const fnBefore = getAppSrvFunction(prms.function.sAppSrvBefore);
resBefore, let resBefore = null;
objOutQueueProcessorSchema.OutQueueProcessorFnBefore, try {
"Результат функции предобработки исходящего сообщения" let resBeforePrms = _.cloneDeep(prms);
); resBefore = await fnBefore(resBeforePrms);
//Если структура ответа в норме } catch (e) {
if (!sCheckResult) { throw new ServerError(SERR_APP_SERVER_BEFORE, e.message);
//Применим её }
if (!_.isUndefined(resBefore.options)) options = _.cloneDeep(resBefore.options); //Проверяем структуру ответа функции предобработки
if (!_.isUndefined(resBefore.blMsg)) { if (resBefore) {
prms.queue.blMsg = resBefore.blMsg; let sCheckResult = validateObject(
await dbConn.setQueueMsg({ resBefore,
nQueueId: prms.queue.nId, objOutQueueProcessorSchema.OutQueueProcessorFnBefore,
blMsg: prms.queue.blMsg "Результат функции предобработки исходящего сообщения"
}); );
if (prms.service.sFnPrmsType == objServiceFnSchema.NFN_PRMS_TYPE_POST) { //Если структура ответа в норме
options.body = prms.queue.blMsg; if (!sCheckResult) {
} else { //Применим её
options.url = buildURL({ if (!_.isUndefined(resBefore.blMsg)) {
sSrvRoot: prms.service.sSrvRoot, prms.queue.blMsg = resBefore.blMsg;
sFnURL: prms.function.sFnURL, await dbConn.setQueueMsg({
sQuery: prms.queue.blMsg.toString() nQueueId: prms.queue.nId,
blMsg: prms.queue.blMsg
});
if (prms.service.sFnPrmsType == objServiceFnSchema.NFN_PRMS_TYPE_POST) {
options.body = prms.queue.blMsg;
} else {
options.url = buildURL({
sSrvRoot: prms.service.sSrvRoot,
sFnURL: prms.function.sFnURL,
sQuery: prms.queue.blMsg.toString()
});
}
}
if (!_.isUndefined(resBefore.options)) options = _.cloneDeep(resBefore.options);
if (!_.isUndefined(resBefore.bUnAuth))
throw new ServerError(SERR_UNAUTH, "Не аутентифицирован");
} else {
//Или расскажем об ошибке в структуре ответа
throw new ServerError(SERR_OBJECT_BAD_INTERFACE, sCheckResult);
}
}
}
//Фиксируем отправку сообщения в протоколе работы сервиса
await logger.info(`Отправляю исходящее сообщение ${prms.queue.nId} на URL: ${options.url}`, {
nQueueId: prms.queue.nId
});
//Отправляем сообщение удалённому серверу
let serverResp = await rqp(options);
//Сохраняем полученный ответ
prms.queue.blResp = new Buffer(serverResp);
await dbConn.setQueueResp({
nQueueId: prms.queue.nId,
blResp: prms.queue.blResp
});
//Выполняем обработчик "После" (если он есть)
if (prms.function.sAppSrvAfter) {
const fnAfter = getAppSrvFunction(prms.function.sAppSrvAfter);
let resAfter = null;
try {
let resAfterPrms = _.cloneDeep(prms);
resAfter = await fnAfter(resAfterPrms);
} catch (e) {
throw new ServerError(SERR_APP_SERVER_AFTER, e.message);
}
//Проверяем структуру ответа функции постобработки
if (resAfter) {
let sCheckResult = validateObject(
resAfter,
objOutQueueProcessorSchema.OutQueueProcessorFnAfter,
"Результат функции постобработки исходящего сообщения"
);
//Если структура ответа в норме
if (!sCheckResult) {
//Применим её
if (!_.isUndefined(resAfter.blResp)) {
prms.queue.blResp = resAfter.blResp;
await dbConn.setQueueResp({
nQueueId: prms.queue.nId,
blResp: prms.queue.blResp
}); });
} }
if (!_.isUndefined(resAfter.bUnAuth))
throw new ServerError(SERR_UNAUTH, "Не аутентифицирован");
} else {
//Или расскажем об ошибке в структуре ответа
throw new ServerError(SERR_OBJECT_BAD_INTERFACE, sCheckResult);
} }
if (!_.isUndefined(resBefore.context)) prms.service.context = _.cloneDeep(resBefore.context);
} else {
//Или расскажем об ошибке
throw new ServerError(SERR_OBJECT_BAD_INTERFACE, sCheckResult);
} }
} }
} //Фиксируем успешное исполнение сервером приложений - в статусе сообщения
//Фиксируем отправку сообщения в протоколе работы сервиса res = await dbConn.setQueueState({
await logger.info(`Отправляю исходящее сообщение ${prms.queue.nId} на URL: ${options.url}`, { nQueueId: prms.queue.nId,
nQueueId: prms.queue.nId nExecState: objQueueSchema.NQUEUE_EXEC_STATE_APP_OK
}); });
//Отправляем сообщение удалённому серверу //Фиксируем успешное исполнение сервером приложений - в протоколе работы сервиса
let serverResp = await rqp(options); await logger.info(`Исходящее сообщение ${prms.queue.nId} успешно отработано сервером приложений`, {
//Сохраняем полученный ответ nQueueId: prms.queue.nId
_.extend(prms, { serverResp }); });
await dbConn.setQueueResp({
nQueueId: prms.queue.nId,
blResp: new Buffer(prms.serverResp)
});
//Выполняем обработчик "После" (если он есть)
if (prms.function.sAppSrvAfter) {
const fnAfter = getAppSrvFunction(prms.function.sAppSrvAfter);
let resAfter = null;
try {
let resAfterPrms = _.cloneDeep(prms);
resAfter = await fnAfter(resAfterPrms);
} catch (e) {
throw new ServerError(SERR_APP_SERVER_AFTER, e.message);
}
//Проверяем структуру ответа функции постобработки
if (resAfter) {
let sCheckResult = validateObject(
resAfter,
objOutQueueProcessorSchema.OutQueueProcessorFnAfter,
"Результат функции постобработки исходящего сообщения"
);
//Если структура ответа в норме
if (!sCheckResult) {
//Применим её
if (!_.isUndefined(resAfter.blResp)) prms.queue.blResp = resAfter.blResp;
if (!_.isUndefined(resAfter.context)) prms.service.context = _.cloneDeep(resAfter.context);
} else {
//Или расскажем об ошибке
throw new ServerError(SERR_OBJECT_BAD_INTERFACE, sCheckResult);
}
} else {
prms.queue.blResp = new Buffer(serverResp.toString());
}
} else { } else {
prms.queue.blResp = new Buffer(serverResp.toString()); //Нет атуентификации (мы ещё не меняли статус сообщения и это не считается за попытку исполнения, это будет просто сигнал, что надо аутентифицироваться а потом задача снова попадёт в очередь)
res = new ServerError(SERR_UNAUTH, "Не аутентифицирован");
} }
//Фиксируем успех исполнения
newQueue = await dbConn.setQueueAppSrvResult({
nQueueId: prms.queue.nId,
blMsg: prms.queue.blMsg,
blResp: prms.queue.blResp
});
//Фиксируем успешное исполнение сервером приложений - в статусе сообщения
newQueue = await dbConn.setQueueState({
nQueueId: prms.queue.nId,
nExecState: objQueueSchema.NQUEUE_EXEC_STATE_APP_OK
});
//Фиксируем успешное исполнение сервером приложений - в протоколе работы сервиса
await logger.info(`Исходящее сообщение ${prms.queue.nId} успешно отработано сервером приложений`, {
nQueueId: prms.queue.nId
});
} catch (e) { } catch (e) {
//Фиксируем ошибку обработки сервером приложений - в статусе сообщения //Фиксируем ошибку обработки сервером приложений - в статусе сообщения
newQueue = await dbConn.setQueueState({ res = await dbConn.setQueueState({
nQueueId: prms.queue.nId, nQueueId: prms.queue.nId,
sExecMsg: makeErrorText(e), sExecMsg: makeErrorText(e),
nIncExecCnt: NINC_EXEC_CNT_YES, nIncExecCnt: NINC_EXEC_CNT_YES,
@ -239,15 +253,18 @@ const appProcess = async prms => {
{ nQueueId: prms.queue.nId } { nQueueId: prms.queue.nId }
); );
} }
//Возвращаем результат
return newQueue;
} else { } else {
throw new ServerError(SERR_OBJECT_BAD_INTERFACE, sCheckResult); //Фатальная ошибка обработки - некорректный объект параметров
res = new ServerError(SERR_OBJECT_BAD_INTERFACE, sCheckResult);
} }
//Возвращаем результат
return res;
}; };
//Запуск обработки сообщения сервером БД //Запуск обработки сообщения сервером БД
const dbProcess = async prms => { const dbProcess = async prms => {
//Результат обработки - объект Queue (обработанное сообщение) или ServerError (ошибка обработки)
let res = null;
//Проверяем структуру переданного объекта для старта //Проверяем структуру переданного объекта для старта
let sCheckResult = validateObject( let sCheckResult = validateObject(
prms, prms,
@ -259,11 +276,11 @@ const dbProcess = async prms => {
//Обрабатываем //Обрабатываем
try { try {
//Фиксируем начало исполнения сервером БД - в статусе сообщения //Фиксируем начало исполнения сервером БД - в статусе сообщения
await dbConn.setQueueState({ res = await dbConn.setQueueState({
nQueueId: prms.queue.nId, nQueueId: prms.queue.nId,
nExecState: objQueueSchema.NQUEUE_EXEC_STATE_DB nExecState: objQueueSchema.NQUEUE_EXEC_STATE_DB
}); });
//Скажем что начали обработку //Фиксируем начало исполнения сервером БД - в протоколе работы сервиса
await logger.info( await logger.info(
`Обрабатываю исходящее сообщение сервером БД: ${prms.queue.nId}, ${prms.queue.sInDate}, ${ `Обрабатываю исходящее сообщение сервером БД: ${prms.queue.nId}, ${prms.queue.sInDate}, ${
prms.queue.sServiceFnCode prms.queue.sServiceFnCode
@ -271,9 +288,9 @@ const dbProcess = async prms => {
{ nQueueId: prms.queue.nId } { nQueueId: prms.queue.nId }
); );
//Вызов обработчика БД //Вызов обработчика БД
await dbConn.execQueueDBPrc({ nQueueId: prms.queue.nId }); res = await dbConn.execQueueDBPrc({ nQueueId: prms.queue.nId });
//Фиксируем успешное исполнение сервером БД - в статусе сообщения //Фиксируем успешное исполнение (полное - дальше обработки нет) - в статусе сообщения
await dbConn.setQueueState({ res = await dbConn.setQueueState({
nQueueId: prms.queue.nId, nQueueId: prms.queue.nId,
nIncExecCnt: prms.queue.nExecCnt == 0 ? NINC_EXEC_CNT_YES : NINC_EXEC_CNT_NO, nIncExecCnt: prms.queue.nExecCnt == 0 ? NINC_EXEC_CNT_YES : NINC_EXEC_CNT_NO,
nExecState: objQueueSchema.NQUEUE_EXEC_STATE_OK nExecState: objQueueSchema.NQUEUE_EXEC_STATE_OK
@ -300,8 +317,11 @@ const dbProcess = async prms => {
); );
} }
} else { } else {
throw new ServerError(SERR_OBJECT_BAD_INTERFACE, sCheckResult); //Фатальная ошибка обработки - некорректный объект параметров
res = new ServerError(SERR_OBJECT_BAD_INTERFACE, sCheckResult);
} }
//Возвращаем результат
return res;
}; };
//Обработка задачи //Обработка задачи
@ -331,101 +351,50 @@ const processTask = async prms => {
await dbConn.connect(); await dbConn.connect();
//Считываем запись очереди //Считываем запись очереди
q = await dbConn.getQueue({ nQueueId: prms.task.nQueueId }); q = await dbConn.getQueue({ nQueueId: prms.task.nQueueId });
//!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!! Проверяем аутентификацию //Выставим флаг - нет ошибок аутентификации на удаленном сервере
//if(prms.task.function.) let bUnAuthFlag = false;
//Далее работаем от статуса считанной записи //Далее работаем от статуса считанной записи
switch (q.nExecState) { switch (q.nExecState) {
//Поставлено в очередь //Статусы "Поставлено в очередь" или "Ошибка обработки сервером приложений"
case objQueueSchema.NQUEUE_EXEC_STATE_INQUEUE: { case objQueueSchema.NQUEUE_EXEC_STATE_INQUEUE:
//Запускаем обработку сервером приложений
try {
let res = await appProcess({
queue: q,
service: prms.task.service,
function: prms.task.function
});
//И если она успешно завершилась - обработку сервером БД
if (res.nExecState == objQueueSchema.NQUEUE_EXEC_STATE_APP_OK) {
try {
await dbProcess({ queue: res });
} catch (e) {
//Фиксируем ошибку обработки сервером БД - в статусе сообщения
await dbConn.setQueueState({
nQueueId: res.nId,
sExecMsg: makeErrorText(e),
nIncExecCnt: NINC_EXEC_CNT_YES,
nExecState:
res.nExecCnt + 1 < res.nRetryAttempts
? objQueueSchema.NQUEUE_EXEC_STATE_DB_ERR
: objQueueSchema.NQUEUE_EXEC_STATE_ERR
});
//Фиксируем ошибку обработки сервером БД - в протоколе работы сервиса
await logger.error(
`Ошибка обработки исходящего сообщения ${res.nId} сервером БД: ${makeErrorText(e)}`,
{ nQueueId: res.nId }
);
}
}
} catch (e) {
//Фиксируем ошибку обработки сервером приложений - в статусе сообщения
newQueue = await dbConn.setQueueState({
nQueueId: q.nId,
sExecMsg: makeErrorText(e),
nIncExecCnt: NINC_EXEC_CNT_YES,
nExecState:
q.nExecCnt + 1 < q.nRetryAttempts
? objQueueSchema.NQUEUE_EXEC_STATE_APP_ERR
: objQueueSchema.NQUEUE_EXEC_STATE_ERR
});
//Фиксируем ошибку обработки сервером приложений - в протоколе работы сервиса
await logger.error(
`Ошибка обработки исходящего сообщения ${q.nId} сервером приложений: ${makeErrorText(e)}`,
{ nQueueId: q.nId }
);
}
break;
}
//Обрабатывается сервером приложений
case objQueueSchema.NQUEUE_EXEC_STATE_APP: {
//Предупредим о неверном статусе сообщения (такие сюда попадать не должны)
await logger.warn(`Cообщение ${q.nId} в статусе ${q.sExecState} попало в очередь обработчика`, {
nQueueId: q.nId
});
break;
}
//Ошибка обработки сервером приложений
case objQueueSchema.NQUEUE_EXEC_STATE_APP_ERR: { case objQueueSchema.NQUEUE_EXEC_STATE_APP_ERR: {
//Если ещё есть попытки отработки //Если ещё не обрабатывали или есть ещё попытки отработки
if (q.nExecCnt < q.nRetryAttempts) { if (q.nExecCnt == 0 || q.nExecCnt < q.nRetryAttempts) {
//Снова запускаем обработку сервером приложений //Запускаем обработку сервером приложений
try { try {
let res = await appProcess({ let res = await appProcess({
queue: q, queue: q,
service: prms.task.service, service: prms.task.service,
function: prms.task.function function: prms.task.function
}); });
//И если она успешно завершилась - обработку сервоером БД //Если результат обработки - ошибка аутентификации
if (res.nExecState == objQueueSchema.NQUEUE_EXEC_STATE_APP_OK) { if (res === objOutQueueProcessorSchema.STASK_RESULT_UNAUTH) {
try { //Выставим флаг, который будет указывать на ошибку аутентификации
await dbProcess({ queue: res }); bUnAuthFlag = true;
} catch (e) { } else {
//Фиксируем ошибку обработки сервером БД - в статусе сообщения //Нет такой ошибки, посмотрим что прилетело сообщение в успешном статусе и тогда запустим обработку сервером БД
await dbConn.setQueueState({ if (res.nExecState == objQueueSchema.NQUEUE_EXEC_STATE_APP_OK) {
nQueueId: res.nId, try {
sExecMsg: makeErrorText(e), await dbProcess({ queue: res });
nIncExecCnt: NINC_EXEC_CNT_YES, } catch (e) {
nExecState: //Фиксируем ошибку обработки сервером БД - в статусе сообщения
res.nExecCnt + 1 < res.nRetryAttempts await dbConn.setQueueState({
? objQueueSchema.NQUEUE_EXEC_STATE_DB_ERR nQueueId: res.nId,
: objQueueSchema.NQUEUE_EXEC_STATE_ERR sExecMsg: makeErrorText(e),
}); nIncExecCnt: NINC_EXEC_CNT_YES,
//Фиксируем ошибку обработки сервером БД - в протоколе работы сервиса nExecState:
await logger.error( res.nExecCnt + 1 < res.nRetryAttempts
`Ошибка обработки исходящего сообщения ${res.nId} сервером БД: ${makeErrorText( ? objQueueSchema.NQUEUE_EXEC_STATE_DB_ERR
e : objQueueSchema.NQUEUE_EXEC_STATE_ERR
)}`, });
{ nQueueId: res.nId } //Фиксируем ошибку обработки сервером БД - в протоколе работы сервиса
); await logger.error(
`Ошибка обработки исходящего сообщения ${
res.nId
} сервером БД: ${makeErrorText(e)}`,
{ nQueueId: res.nId }
);
}
} }
} }
} catch (e) { } catch (e) {
@ -458,42 +427,11 @@ const processTask = async prms => {
} }
break; break;
} }
//Успешно обработано сервером приложений //Статусы "Успешно обработано сервером приложений" и "Ошибка обработки в БД"
case objQueueSchema.NQUEUE_EXEC_STATE_APP_OK: { case objQueueSchema.NQUEUE_EXEC_STATE_APP_OK:
//Запускаем обработку в БД
try {
await dbProcess({ queue: q });
} catch (e) {
//Фиксируем ошибку обработки сервером БД - в статусе сообщения
await dbConn.setQueueState({
nQueueId: q.nId,
sExecMsg: makeErrorText(e),
nIncExecCnt: NINC_EXEC_CNT_YES,
nExecState:
q.nExecCnt + 1 < q.nRetryAttempts
? objQueueSchema.NQUEUE_EXEC_STATE_DB_ERR
: objQueueSchema.NQUEUE_EXEC_STATE_ERR
});
//Фиксируем ошибку обработки сервером БД - в протоколе работы сервиса
await logger.error(
`Ошибка обработки исходящего сообщения ${q.nId} сервером БД: ${makeErrorText(e)}`,
{ nQueueId: q.nId }
);
}
break;
}
//Обрабатывается в БД
case objQueueSchema.NQUEUE_EXEC_STATE_DB: {
//Предупредим о неверном статусе сообщения (такие сюда попадать не должны)
await logger.warn(`Cообщение ${q.nId} в статусе ${q.sExecState} попало в очередь обработчика`, {
nQueueId: q.nId
});
break;
}
//Ошибка обработки в БД
case objQueueSchema.NQUEUE_EXEC_STATE_DB_ERR: { case objQueueSchema.NQUEUE_EXEC_STATE_DB_ERR: {
//Если ещё есть попытки отработки //Если ещё есть попытки отработки
if (q.nExecCnt < q.nRetryAttempts) { if (q.nExecCnt == 0 || q.nExecCnt < q.nRetryAttempts) {
//Снова запускаем обработку сервером БД //Снова запускаем обработку сервером БД
try { try {
await dbProcess({ queue: q }); await dbProcess({ queue: q });
@ -536,15 +474,10 @@ const processTask = async prms => {
}); });
break; break;
} }
//Обработано с ошибками //Статусы "Обрабатывается сервером приложений", "Обрабатывается в БД", "Обработано с ошибками", "Обработано успешно"
case objQueueSchema.NQUEUE_EXEC_STATE_ERR: { case objQueueSchema.NQUEUE_EXEC_STATE_APP:
//Предупредим о неверном статусе сообщения (такие сюда попадать не должны) case objQueueSchema.NQUEUE_EXEC_STATE_DB:
await logger.warn(`Cообщение ${q.nId} в статусе ${q.sExecState} попало в очередь обработчика`, { case objQueueSchema.NQUEUE_EXEC_STATE_ERR:
nQueueId: q.nId
});
break;
}
//Обработано успешно
case objQueueSchema.NQUEUE_EXEC_STATE_OK: { case objQueueSchema.NQUEUE_EXEC_STATE_OK: {
//Предупредим о неверном статусе сообщения (такие сюда попадать не должны) //Предупредим о неверном статусе сообщения (такие сюда попадать не должны)
await logger.warn(`Cообщение ${q.nId} в статусе ${q.sExecState} попало в очередь обработчика`, { await logger.warn(`Cообщение ${q.nId} в статусе ${q.sExecState} попало в очередь обработчика`, {
@ -552,6 +485,7 @@ const processTask = async prms => {
}); });
break; break;
} }
//Неипонятный статус
default: { default: {
//Ничего не делаем //Ничего не делаем
break; break;
@ -559,8 +493,9 @@ const processTask = async prms => {
} }
//Отключаемся от БД //Отключаемся от БД
if (dbConn) await dbConn.disconnect(); if (dbConn) await dbConn.disconnect();
//Отправляем успех //Отправляем успех или ошибку аутентификации
sendOKResult({ context: prms.task.service.context }); if (bUnAuthFlag) sendUnAuthResult();
else sendOKResult();
} catch (e) { } catch (e) {
//Отключаемся от БД //Отключаемся от БД
if (dbConn) await dbConn.disconnect(); if (dbConn) await dbConn.disconnect();

35
models/obj_out_queue_processor.js
View File

@ -19,6 +19,7 @@ const { ServiceFunction } = require("./obj_service_function"); //Схема ва
//Состояния обработки сообщений очереди обмена //Состояния обработки сообщений очереди обмена
const STASK_RESULT_OK = "OK"; //Обработано успешно const STASK_RESULT_OK = "OK"; //Обработано успешно
const STASK_RESULT_ERR = "ERR"; //Обработано с ошибками const STASK_RESULT_ERR = "ERR"; //Обработано с ошибками
const STASK_RESULT_UNAUTH = "UNAUTH"; //Не обработано из-за отсутсвия аутентификации
//------------------ //------------------
// Интерфейс модуля // Интерфейс модуля
@ -27,6 +28,7 @@ const STASK_RESULT_ERR = "ERR"; //Обработано с ошибками
//Константы //Константы
exports.STASK_RESULT_OK = STASK_RESULT_OK; exports.STASK_RESULT_OK = STASK_RESULT_OK;
exports.STASK_RESULT_ERR = STASK_RESULT_ERR; exports.STASK_RESULT_ERR = STASK_RESULT_ERR;
exports.STASK_RESULT_UNAUTH = STASK_RESULT_UNAUTH;
//Схема валидации задачи обработчику очереди исходящих сообщений //Схема валидации задачи обработчику очереди исходящих сообщений
exports.OutQueueProcessorTask = new Schema({ exports.OutQueueProcessorTask = new Schema({
@ -71,7 +73,7 @@ exports.OutQueueProcessorTaskResult = new Schema({
//Состояние обработки сообщения очереди обмена //Состояние обработки сообщения очереди обмена
sResult: { sResult: {
type: String, type: String,
enum: [STASK_RESULT_OK, STASK_RESULT_ERR], enum: [STASK_RESULT_OK, STASK_RESULT_ERR, STASK_RESULT_UNAUTH],
required: true, required: true,
message: { message: {
type: path => type: path =>
@ -89,15 +91,6 @@ exports.OutQueueProcessorTaskResult = new Schema({
`Информация от обработчика сообщения очереди обмена (${path}) имеет некорректный тип данных (ожидалось - String)`, `Информация от обработчика сообщения очереди обмена (${path}) имеет некорректный тип данных (ожидалось - String)`,
required: path => `Не указана информация от обработчика сообщения очереди обмена (${path})` required: path => `Не указана информация от обработчика сообщения очереди обмена (${path})`
} }
},
//Контекст работы сервиса
context: {
type: Object,
required: true,
message: {
type: path => `Контекст работы сервиса (${path}) имеет некорректный тип данных (ожидалось - Object)`,
required: path => `Не указан контекст работы сервиса (${path})`
}
} }
}).validator({ }).validator({
required: val => typeof val != "undefined" required: val => typeof val != "undefined"
@ -125,13 +118,14 @@ exports.OutQueueProcessorFnBefore = new Schema({
required: path => `Не указано обработанное сообщение очереди (${path})` required: path => `Не указано обработанное сообщение очереди (${path})`
} }
}, },
//Контекст работы сервиса //Флаг ошибки аутентификации на удаленном сервисе
context: { bUnAuth: {
type: Object, type: Boolean,
required: false, required: false,
message: { message: {
type: path => `Контекст работы сервиса (${path}) имеет некорректный тип данных (ожидалось - Object)`, type: path =>
required: path => `Не указан контекст работы сервиса (${path})` `Флаг ошибки аутентификации на удаленном сервисе (${path}) имеет некорректный тип данных (ожидалось - Boolean)`,
required: path => `Не указан флаг ошибки аутентификации на удаленном сервисе (${path})`
} }
} }
}); });
@ -148,13 +142,14 @@ exports.OutQueueProcessorFnAfter = new Schema({
required: path => `Не указан результат обработки ответа удалённого сервиса (${path})` required: path => `Не указан результат обработки ответа удалённого сервиса (${path})`
} }
}, },
//Контекст работы сервиса //Флаг ошибки аутентификации на удаленном сервисе
context: { bUnAuth: {
type: Object, type: Boolean,
required: false, required: false,
message: { message: {
type: path => `Контекст работы сервиса (${path}) имеет некорректный тип данных (ожидалось - Object)`, type: path =>
required: path => `Не указан контекст работы сервиса (${path})` `Флаг ошибки аутентификации на удаленном сервисе (${path}) имеет некорректный тип данных (ожидалось - Boolean)`,
required: path => `Не указан флаг ошибки аутентификации на удаленном сервисе (${path})`
} }
} }
}); });

17
models/prms_out_queue_processor.js
View File

@ -24,21 +24,8 @@ exports.sendErrorResult = new Schema({
type: String, type: String,
required: true, required: true,
message: { message: {
type: path => `Идентификатор сервиса (${path}) имеет некорректный тип данных (ожидалось - String)`, type: path => `Сообщение об ошибке (${path}) имеет некорректный тип данных (ожидалось - String)`,
required: path => `Не указан идентификатор сервиса (${path})` required: path => `Не указано сообщение об ошибке (${path})`
}
}
});
//Схема валидации параметров функции отправки успеха обработки
exports.sendOKResult = new Schema({
//Контекст работы сервиса
context: {
type: Object,
required: true,
message: {
type: path => `Контекст работы сервиса (${path}) имеет некорректный тип данных (ожидалось - Object)`,
required: path => `Не указан контекст работы сервиса (${path})`
} }
} }
}); });