Compare commits

..

2 Commits

4 changed files with 67 additions and 26 deletions

View File

@ -14,7 +14,7 @@ let common = {
//Версия сервера приложений //Версия сервера приложений
sVersion: "8.5.6.1", sVersion: "8.5.6.1",
//Релиз сервера приложений //Релиз сервера приложений
sRelease: "2025.07.29", sRelease: "2025.07.01",
//Таймаут останова сервера (мс) //Таймаут останова сервера (мс)
nTerminateTimeout: 60000, nTerminateTimeout: 60000,
//Контролировать версию Системы //Контролировать версию Системы
@ -70,7 +70,9 @@ let inComing = {
//Максимальный размер пула подключений к БД для обработчика входящих сообщений //Максимальный размер пула подключений к БД для обработчика входящих сообщений
nPoolMax: 10, nPoolMax: 10,
//Шаг инкремента подключений к БД в пуле обработчика входящих сообщений //Шаг инкремента подключений к БД в пуле обработчика входящих сообщений
nPoolIncrement: 0 nPoolIncrement: 0,
//Время ожидания отработки входящего сообщения (мс, 0 - не применять (не отменяет таймаут по умолчанию, который может быть установлен платформой))
nTimeout: 0
}; };
//Параметры подключения к Kafka //Параметры подключения к Kafka

View File

@ -329,19 +329,28 @@ class InQueue extends EventEmitter {
} }
} }
} }
//Всё успешно - отдаём результат клиенту //Если мы еще не отдали ответ от сервера
if (bStopPropagation === false) { if (!prms.res.writableFinished) {
if (optionsResp.headers) prms.res.set(optionsResp.headers); //Всё успешно - отдаём результат клиенту
prms.res.status(optionsResp.statusCode || 200).send(blResp); if (bStopPropagation === false) {
if (optionsResp.headers) prms.res.set(optionsResp.headers);
prms.res.status(optionsResp.statusCode || 200).send(blResp);
}
//Фиксируем успех обработки - в протоколе работы сервиса
await this.logger.info(`Входящее сообщение ${q.nId} успешно отработано`, { nQueueId: q.nId });
//Фиксируем успех обработки - в статусе сообщения
q = await this.dbConn.setQueueState({
nQueueId: q.nId,
nIncExecCnt: NINC_EXEC_CNT_YES,
nExecState: objQueueSchema.NQUEUE_EXEC_STATE_OK
});
} else {
//Или расскажем об ошибке
throw new ServerError(
SERR_WEB_SERVER,
"Истекло время ожидания обработки входящего запроса. Канал закрыт. Клиенту был отправлен ответ."
);
} }
//Фиксируем успех обработки - в протоколе работы сервиса
await this.logger.info(`Входящее сообщение ${q.nId} успешно отработано`, { nQueueId: q.nId });
//Фиксируем успех обработки - в статусе сообщения
q = await this.dbConn.setQueueState({
nQueueId: q.nId,
nIncExecCnt: NINC_EXEC_CNT_YES,
nExecState: objQueueSchema.NQUEUE_EXEC_STATE_OK
});
} catch (e) { } catch (e) {
//Тема и текст уведомления об ошибке //Тема и текст уведомления об ошибке
let sSubject = `Ошибка обработки входящего сообщения сервером приложений для функции "${prms.function.sCode}" сервиса "${prms.service.sCode}"`; let sSubject = `Ошибка обработки входящего сообщения сервером приложений для функции "${prms.function.sCode}" сервиса "${prms.service.sCode}"`;
@ -375,8 +384,11 @@ class InQueue extends EventEmitter {
sMessage sMessage
}); });
} }
//Отправим ошибку клиенту //Если мы еще не отдали ответ от сервера
prms.res.status(500).send(makeErrorText(e)); if (!prms.res.writableFinished) {
//Отправим ошибку клиенту
prms.res.status(500).send(makeErrorText(e));
}
} }
} else { } else {
throw new ServerError(SERR_OBJECT_BAD_INTERFACE, sCheckResult); throw new ServerError(SERR_OBJECT_BAD_INTERFACE, sCheckResult);
@ -543,6 +555,21 @@ class InQueue extends EventEmitter {
if (req.headers["content-type"] === "false") req.headers["content-type"] = "application/octet-stream"; if (req.headers["content-type"] === "false") req.headers["content-type"] = "application/octet-stream";
next(); next();
}); });
//Если требуется установить таймаут на обработку сообщений
if (this.inComing.nTimeout !== 0) {
//Конфигурируем сервер - устанавливаем таймаут обработки сообщений
this.webApp.use((req, res, next) => {
//Устанавливаем таймаут на ответ от сервера
res.setTimeout(this.inComing.nTimeout, () => {
//Формируем ошибку
let err = new Error("Истекло время ожидания формирования ответа для завершения текущего запроса.");
err.status = 504;
//Отправляем ошибку
next(err);
});
next();
});
}
//Конфигурируем сервер - обработка тела сообщения //Конфигурируем сервер - обработка тела сообщения
this.webApp.use(bodyParser.raw({ limit: `${this.inComing.nMsgMaxSize}mb`, type: "*/*" })); this.webApp.use(bodyParser.raw({ limit: `${this.inComing.nMsgMaxSize}mb`, type: "*/*" }));
//Конфигурируем сервер - обходим все сервисы, работающие на приём сообщений //Конфигурируем сервер - обходим все сервисы, работающие на приём сообщений
@ -585,8 +612,11 @@ class InQueue extends EventEmitter {
nServiceId: srvs.nId, nServiceId: srvs.nId,
nServiceFnId: fn.nId nServiceFnId: fn.nId
}); });
//Отправим ошибку клиенту //Если мы еще не отдали ответ от сервера
res.status(500).send(makeErrorText(e)); if (!res.writableFinished) {
//Отправим ошибку клиенту
res.status(500).send(makeErrorText(e));
}
} }
}); });
//...и собственный обработчик ошибок //...и собственный обработчик ошибок
@ -597,7 +627,7 @@ class InQueue extends EventEmitter {
nServiceFnId: fn.nId nServiceFnId: fn.nId
}); });
//Отправим ошибку клиенту //Отправим ошибку клиенту
res.status(500).send(makeErrorText(new ServerError(SERR_WEB_SERVER, err.message))); res.status(err.status || 500).send(makeErrorText(new ServerError(SERR_WEB_SERVER, err.message)));
}); });
} }
); );

View File

@ -49,6 +49,9 @@ const validatePoolMaxInComing = val => val >= 1 && val <= 1000;
//Функция проверки значения шага инкремента подключений к БД в пуле обработчика входящих сообщений //Функция проверки значения шага инкремента подключений к БД в пуле обработчика входящих сообщений
const validatePoolIncrementInComing = val => val >= 0 && val <= 1000; const validatePoolIncrementInComing = val => val >= 0 && val <= 1000;
//Функция проверки значения времени ожидания отработки входящего сообщения для обработчика входящих сообщений
const validateTimeoutInComing = val => val >= 0;
//Схема валидации общих параметров сервера приложений //Схема валидации общих параметров сервера приложений
const common = new Schema({ const common = new Schema({
//Наименование сервера приложений //Наименование сервера приложений
@ -323,6 +326,17 @@ const inComing = new Schema({
validatePoolIncrementInComing: path => validatePoolIncrementInComing: path =>
`Значение шага инкремента подключений к БД в пуле обработчика входящих сообщений (${path}) должно быть целым числом в диапазоне от 0 до 1000` `Значение шага инкремента подключений к БД в пуле обработчика входящих сообщений (${path}) должно быть целым числом в диапазоне от 0 до 1000`
} }
},
//Время ожидания отработки входящего сообщения (мс)
nTimeout: {
type: Number,
required: true,
use: { validateTimeoutInComing },
message: {
type: path => `Время ожидания отработки входящего сообщения (${path}) имеет некорректный тип данных (ожидалось - Number)`,
required: path => `Не указано время ожидания отработки входящего сообщения (${path})`,
validateTimeoutInComing: path => `Время ожидания отработки входящего сообщения (${path}) должно быть неотрицательным целым числом`
}
} }
}); });

View File

@ -27,8 +27,7 @@ const tag = [
"Resolutions", "Resolutions",
"XmlSignatureRejections", "XmlSignatureRejections",
"RecipientTitles", "RecipientTitles",
"Requests", "Requests"
"UniversalMessages"
]; ];
//------------ //------------
@ -684,16 +683,12 @@ const beforeDocLoad = async prms => {
entId = "documentId="; entId = "documentId=";
msgId = "letterId="; msgId = "letterId=";
break; break;
//Загрузка Универсального сообщения
case 6:
entId = "attachmentId=";
break;
default: default:
} }
surl = `${surl}?${msgId}${prms.options.smsgid}&${entId}${prms.options.sentid}`; surl = `${surl}?${msgId}${prms.options.smsgid}&${entId}${prms.options.sentid}`;
let obj; let obj;
let rblMsg; let rblMsg;
if (prms.queue.blMsg && (prms.options.type != 5) && (prms.options.type != 6)) { if (prms.queue.blMsg && prms.options.type != 5) {
//Конвертируем XML из "Парус 8" в понятный "ДИАДОК" JSON //Конвертируем XML из "Парус 8" в понятный "ДИАДОК" JSON
obj = await toJSON(prms.queue.blMsg.toString()); obj = await toJSON(prms.queue.blMsg.toString());
rblMsg = Buffer.from(JSON.stringify(obj)); rblMsg = Buffer.from(JSON.stringify(obj));