diff --git a/config.js b/config.js index 1207399..9b6511a 100644 --- a/config.js +++ b/config.js @@ -70,7 +70,9 @@ let inComing = { //Максимальный размер пула подключений к БД для обработчика входящих сообщений nPoolMax: 10, //Шаг инкремента подключений к БД в пуле обработчика входящих сообщений - nPoolIncrement: 0 + nPoolIncrement: 0, + //Время ожидания отработки входящего сообщения (мс, 0 - не применять (не отменяет таймаут по умолчанию, который может быть установлен платформой)) + nTimeout: 0 }; //Параметры подключения к Kafka diff --git a/core/in_queue.js b/core/in_queue.js index a6ab9c9..cee141a 100644 --- a/core/in_queue.js +++ b/core/in_queue.js @@ -329,19 +329,28 @@ class InQueue extends EventEmitter { } } } - //Всё успешно - отдаём результат клиенту - if (bStopPropagation === false) { - if (optionsResp.headers) prms.res.set(optionsResp.headers); - prms.res.status(optionsResp.statusCode || 200).send(blResp); + //Если мы еще не отдали ответ от сервера + if (!prms.res.writableFinished) { + //Всё успешно - отдаём результат клиенту + 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) { //Тема и текст уведомления об ошибке let sSubject = `Ошибка обработки входящего сообщения сервером приложений для функции "${prms.function.sCode}" сервиса "${prms.service.sCode}"`; @@ -375,8 +384,11 @@ class InQueue extends EventEmitter { sMessage }); } - //Отправим ошибку клиенту - prms.res.status(500).send(makeErrorText(e)); + //Если мы еще не отдали ответ от сервера + if (!prms.res.writableFinished) { + //Отправим ошибку клиенту + prms.res.status(500).send(makeErrorText(e)); + } } } else { 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"; 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: "*/*" })); //Конфигурируем сервер - обходим все сервисы, работающие на приём сообщений @@ -585,8 +612,11 @@ class InQueue extends EventEmitter { nServiceId: srvs.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 }); //Отправим ошибку клиенту - 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))); }); } ); diff --git a/models/obj_config.js b/models/obj_config.js index 563dd5a..f17d7d4 100644 --- a/models/obj_config.js +++ b/models/obj_config.js @@ -49,6 +49,9 @@ const validatePoolMaxInComing = val => val >= 1 && val <= 1000; //Функция проверки значения шага инкремента подключений к БД в пуле обработчика входящих сообщений const validatePoolIncrementInComing = val => val >= 0 && val <= 1000; +//Функция проверки значения времени ожидания отработки входящего сообщения для обработчика входящих сообщений +const validateTimeoutInComing = val => val >= 0; + //Схема валидации общих параметров сервера приложений const common = new Schema({ //Наименование сервера приложений @@ -323,6 +326,17 @@ const inComing = new Schema({ validatePoolIncrementInComing: path => `Значение шага инкремента подключений к БД в пуле обработчика входящих сообщений (${path}) должно быть целым числом в диапазоне от 0 до 1000` } + }, + //Время ожидания отработки входящего сообщения (мс) + nTimeout: { + type: Number, + required: true, + use: { validateTimeoutInComing }, + message: { + type: path => `Время ожидания отработки входящего сообщения (${path}) имеет некорректный тип данных (ожидалось - Number)`, + required: path => `Не указано время ожидания отработки входящего сообщения (${path})`, + validateTimeoutInComing: path => `Время ожидания отработки входящего сообщения (${path}) должно быть неотрицательным целым числом` + } } });