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

ЦИТК-255 - исправлена ошибка обработки таймаута для сообщений, отработанных с флагом "bStopPropagation = true"

Browse Source 
Сервер ошибочно считал такие сообщения "нарушением таймаута" и писал в журнал несуществующие ошибки, в то время как сообщение нормально отрабатывалось.
This commit is contained in:
Mikhail Chechnev 2025-12-01 19:22:00 +03:00
parent 80f50d43a3
commit 9cd779dd64
1 changed files with 27 additions and 21 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

48
core/in_queue.js
View File

@ -152,6 +152,8 @@ class InQueue extends EventEmitter {
sOptions: buildOptionsXML({ options }), sOptions: buildOptionsXML({ options }),
blMsg blMsg
}); });
//Запомним идентификатор записи очереди в запросе
prms.req.nQId = q.nId;
//Скажем что пришло новое входящее сообщение //Скажем что пришло новое входящее сообщение
await this.logger.info( await this.logger.info(
`Новое входящее сообщение от ${prms.req.connection.address().address} для функции ${prms.function.sCode} (${buildURL({ `Новое входящее сообщение от ${prms.req.connection.address().address} для функции ${prms.function.sCode} (${buildURL({
@ -329,13 +331,19 @@ class InQueue extends EventEmitter {
} }
} }
} }
//Если мы еще не отдали ответ от сервера //Всё успешно - отдаём результат клиенту, если ещё не отдали
if (!prms.res.writableFinished) { if (bStopPropagation === false && !prms.res.writableFinished) {
//Всё успешно - отдаём результат клиенту if (optionsResp.headers) prms.res.set(optionsResp.headers);
if (bStopPropagation === false) { prms.res.status(optionsResp.statusCode || 200).send(blResp);
if (optionsResp.headers) prms.res.set(optionsResp.headers); }
prms.res.status(optionsResp.statusCode || 200).send(blResp); //Если отправка ответа была прервана по таймауту
} if (prms.req.bIsTimedOut === true) {
//Вернем ошибку обработчика с информацией об этом
throw new ServerError(
SERR_WEB_SERVER,
"Истекло время ожидания обработки входящего запроса. Канал закрыт. Клиенту был отправлен ответ с ошибкой истечения таймаута (504)."
);
} else {
//Фиксируем успех обработки - в протоколе работы сервиса //Фиксируем успех обработки - в протоколе работы сервиса
await this.logger.info(`Входящее сообщение ${q.nId} успешно отработано`, { nQueueId: q.nId }); await this.logger.info(`Входящее сообщение ${q.nId} успешно отработано`, { nQueueId: q.nId });
//Фиксируем успех обработки - в статусе сообщения //Фиксируем успех обработки - в статусе сообщения
@ -344,12 +352,6 @@ class InQueue extends EventEmitter {
nIncExecCnt: NINC_EXEC_CNT_YES, nIncExecCnt: NINC_EXEC_CNT_YES,
nExecState: objQueueSchema.NQUEUE_EXEC_STATE_OK nExecState: objQueueSchema.NQUEUE_EXEC_STATE_OK
}); });
} else {
//Или расскажем об ошибке
throw new ServerError(
SERR_WEB_SERVER,
"Истекло время ожидания обработки входящего запроса. Канал закрыт. Клиенту был отправлен ответ."
);
} }
} catch (e) { } catch (e) {
//Тема и текст уведомления об ошибке //Тема и текст уведомления об ошибке
@ -555,21 +557,24 @@ 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) => {
//Конфигурируем сервер - устанавливаем таймаут обработки сообщений //Поднимем флаг истечения таймаута обработки
this.webApp.use((req, res, next) => { req.bIsTimedOut = false;
//Если требуется установить таймаут на обработку сообщений
if (this.inComing.nTimeout !== 0)
//Устанавливаем таймаут на ответ от сервера //Устанавливаем таймаут на ответ от сервера
res.setTimeout(this.inComing.nTimeout, () => { res.setTimeout(this.inComing.nTimeout, () => {
//Поднимем флаг исчетечение таймаута обработки
req.bIsTimedOut = true;
//Формируем ошибку //Формируем ошибку
let err = new Error("Истекло время ожидания формирования ответа для завершения текущего запроса."); let err = new Error("Истекло время ожидания формирования ответа для завершения текущего запроса.");
err.status = 504; err.status = 504;
//Отправляем ошибку //Отправляем ошибку
next(err); next(err);
}); });
next(); next();
}); });
}
//Конфигурируем сервер - обработка тела сообщения //Конфигурируем сервер - обработка тела сообщения
this.webApp.use(bodyParser.raw({ limit: `${this.inComing.nMsgMaxSize}mb`, type: "*/*" })); this.webApp.use(bodyParser.raw({ limit: `${this.inComing.nMsgMaxSize}mb`, type: "*/*" }));
//Конфигурируем сервер - обходим все сервисы, работающие на приём сообщений //Конфигурируем сервер - обходим все сервисы, работающие на приём сообщений
@ -624,7 +629,8 @@ class InQueue extends EventEmitter {
//Протоколируем в журнал работы сервера //Протоколируем в журнал работы сервера
await this.logger.error(makeErrorText(new ServerError(SERR_WEB_SERVER, err.message)), { await this.logger.error(makeErrorText(new ServerError(SERR_WEB_SERVER, err.message)), {
nServiceId: srvs.nId, nServiceId: srvs.nId,
nServiceFnId: fn.nId nServiceFnId: fn.nId,
nQueueId: req.nQId || null
}); });
//Отправим ошибку клиенту //Отправим ошибку клиенту
res.status(err.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)));