Добавлен параметр "promiseCancellable" для "wrapPromiseTimeout" - обеспечение корректной обработки промисов, не имеющих методов "isPending" и "cancel"

ЦИТК-255 - исправлена ошибка обработки таймаута для сообщений, отработанных с флагом "bStopPropagation = true"
Сервер ошибочно считал такие сообщения "нарушением таймаута" и писал в журнал несуществующие ошибки, в то время как сообщение нормально отрабатывалось.
2025-12-01 19:23:46 +03:00 · 2025-12-01 19:22:00 +03:00 · 2025-11-26 12:50:32 +03:00 · 2025-11-26 12:48:48 +03:00 · 2025-11-12 12:11:40 +03:00 · 2025-11-10 16:17:58 +03:00
6 changed files with 60 additions and 38 deletions
--- a/config.js
+++ b/config.js
@ -14,7 +14,7 @@ let common = {
    //Версия сервера приложений
    sVersion: "8.5.6.1",
    //Релиз сервера приложений
-    sRelease: "2025.07.01",
+    sRelease: "2025.11.06",
    //Таймаут останова сервера (мс)
    nTerminateTimeout: 60000,
    //Контролировать версию Системы
--- a/core/in_queue.js
+++ b/core/in_queue.js
@ -152,6 +152,8 @@ class InQueue extends EventEmitter {
                    sOptions: buildOptionsXML({ options }),
                    blMsg
                });
+                //Запомним идентификатор записи очереди в запросе
+                prms.req.nQId = q.nId;
                //Скажем что пришло новое входящее сообщение
                await this.logger.info(
                    `Новое входящее сообщение от ${prms.req.connection.address().address} для функции ${prms.function.sCode} (${buildURL({
@ -329,13 +331,19 @@ class InQueue extends EventEmitter {
                        }
                    }
                }
-                //Если мы еще не отдали ответ от сервера
-                if (!prms.res.writableFinished) {
-                    //Всё успешно - отдаём результат клиенту
-                    if (bStopPropagation === false) {
-                        if (optionsResp.headers) prms.res.set(optionsResp.headers);
-                        prms.res.status(optionsResp.statusCode || 200).send(blResp);
-                    }
+                //Всё успешно - отдаём результат клиенту, если ещё не отдали
+                if (bStopPropagation === false && !prms.res.writableFinished) {
+                    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 });
                    //Фиксируем успех обработки - в статусе сообщения
@ -344,12 +352,6 @@ class InQueue extends EventEmitter {
                        nIncExecCnt: NINC_EXEC_CNT_YES,
                        nExecState: objQueueSchema.NQUEUE_EXEC_STATE_OK
                    });
-                } else {
-                    //Или расскажем об ошибке
-                    throw new ServerError(
-                        SERR_WEB_SERVER,
-                        "Истекло время ожидания обработки входящего запроса. Канал закрыт. Клиенту был отправлен ответ."
-                    );
                }
            } catch (e) {
                //Тема и текст уведомления об ошибке
@ -555,21 +557,24 @@ 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) => {
+            //Конфигурируем сервер - устанавливаем таймаут обработки сообщений
+            this.webApp.use((req, res, next) => {
+                //Поднимем флаг истечения таймаута обработки
+                req.bIsTimedOut = false;
+                //Если требуется установить таймаут на обработку сообщений
+                if (this.inComing.nTimeout !== 0)
                    //Устанавливаем таймаут на ответ от сервера
                    res.setTimeout(this.inComing.nTimeout, () => {
+                        //Поднимем флаг исчетечение таймаута обработки
+                        req.bIsTimedOut = true;
                        //Формируем ошибку
                        let err = new Error("Истекло время ожидания формирования ответа для завершения текущего запроса.");
                        err.status = 504;
                        //Отправляем ошибку
                        next(err);
                    });
-                    next();
-                });
-            }
+                next();
+            });
            //Конфигурируем сервер - обработка тела сообщения
            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)), {
                                    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)));
--- a/core/kafka_connector.js
+++ b/core/kafka_connector.js
@ -90,7 +90,7 @@ const subscribeKafka = async ({ settings, service, processMessage, logger }) =>
                        })
                    });
                } catch (e) {
-                    await logger.error(`Ошибка обработки исходящего сообщения Kafka: ${makeErrorText(e)}`);
+                    await logger.error(`Ошибка обработки входящего сообщения Kafka: ${makeErrorText(e)}`);
                }
            }
        });
@ -107,7 +107,7 @@ const subscribeKafka = async ({ settings, service, processMessage, logger }) =>
        //Возвращаем соединение
        return consumer;
    } catch (e) {
-        await logger.error(`Ошибка запуска обработчика очереди исходящих сообщений Kafka: ${makeErrorText(e)}`);
+        await logger.error(`Ошибка запуска обработчика очереди входящих сообщений Kafka: ${makeErrorText(e)}`);
    }
 };

--- a/core/utils.js
+++ b/core/utils.js
@ -404,7 +404,7 @@ const getURLProtocol = sURL => {
 };

 //Обёртывание промиса в таймаут исполнения
-const wrapPromiseTimeout = (timeout, promise) => {
+const wrapPromiseTimeout = (timeout, promise, promiseCancellable = true) => {
    if (!timeout) return promise;
    let timeoutPid;
    const timeoutPromise = new Promise((resolve, reject) => {
@ -414,7 +414,7 @@ const wrapPromiseTimeout = (timeout, promise) => {
        timeoutPid = setTimeout(() => reject(e), timeout);
    });
    return Promise.race([promise, timeoutPromise]).finally(() => {
-        if (promise.promise().isPending()) promise.cancel();
+        if (promiseCancellable && promise.promise().isPending()) promise.cancel();
        if (timeoutPid) clearTimeout(timeoutPid);
    });
 };
--- a/modules/diadoc.js
+++ b/modules/diadoc.js
@ -27,7 +27,8 @@ const tag = [
    "Resolutions",
    "XmlSignatureRejections",
    "RecipientTitles",
-    "Requests"
+    "Requests",
+    "UniversalMessages"
 ];

 //------------
@ -144,6 +145,11 @@ const getOrganizations = organizations => {
    //Найдем активную организацию не в роуминге
    organization.Organizations[0] = organizations.Organizations.find(org => (org.IsRoaming === isRoaming) && (org.IsActive === isActive));
    //Если не удалось получить организацию не в роуминге
+    if (!organization.Organizations[0]) {
+        //Найдем активную организацию
+        organization.Organizations[0] = organizations.Organizations.find(org => (org.IsActive === isActive));
+    };
+    //Если не удалось получить активную организацию
    if (!organization.Organizations[0]) {
        //Если нет организации не в роуминге и найдено более одной организации
        if (organizations.Organizations.length > 1) {
@ -290,6 +296,8 @@ const beforeMessagePost = async prms => {
        }
        //Если не достали из контекста токен доступа - значит нет аутентификации на сервере
        if (!sToken) return { bUnAuth: true };
+        //Получим параметры запроса
+        const optionsData = await toJSON(prms.queue.sOptions);
        //Конвертируем XML из "Парус 8" в JSON
        let obj = await toJSON(prms.queue.blMsg.toString());
        //Формируем запрос для получения FromBoxId
@ -307,7 +315,7 @@ const beforeMessagePost = async prms => {
            //Получим идентификатор организации по ИНН/КПП поставщика документа
            for (let i in serverResp.Organizations) {
                //Если найдена подходящая организация - запомним идентификатор и выходим из цикла
-                if (serverResp.Organizations[i].Inn == prms.options.inn_pr && serverResp.Organizations[i].Kpp == prms.options.kpp_pr) {
+                if (serverResp.Organizations[i].Inn == optionsData.inn_pr && serverResp.Organizations[i].Kpp == optionsData.kpp_pr) {
                    //Сохраняем полученный ответ
                    obj.FromBoxId = serverResp.Organizations[i].Boxes[0].BoxId;
                    break;
@ -315,13 +323,13 @@ const beforeMessagePost = async prms => {
            }
            //Не удалось получить ящик отправителя
            if (!obj.FromBoxId) {
-                throw new Error(`Не удалось получить ящик текущей организации с ИНН: ${prms.options.inn_pr} и КПП: ${prms.options.kpp_pr}`);
+                throw new Error(`Не удалось получить ящик текущей организации с ИНН: ${optionsData.inn_pr} и КПП: ${optionsData.kpp_pr}`);
            }
        } catch (e) {
            throw Error(`Ошибка при получении ящика текущей организации:  ${e.message}`);
        }
        //Получим ящик получателя
-        organization = await getOrganizationBoxId(prms.service.sSrvRoot, buildHeaders(sAPIClientId, sToken), prms.options.inn_cs, prms.options.kpp_cs);
+        organization = await getOrganizationBoxId(prms.service.sSrvRoot, buildHeaders(sAPIClientId, sToken), optionsData.inn_cs, optionsData.kpp_cs);
        obj.ToBoxId = organization.BoxId;
        //Если не заполнен идентификатор подразделения и при получении ящика удалось его подобрать
        if ((!obj.ToDepartmentId) && (organization.DepartmentId)) {
@ -448,6 +456,8 @@ const beforeEvent = async prms => {
        }
        //Если не достали из контекста токен доступа - значит нет аутентификации на сервере
        if (!sToken) return { bUnAuth: true };
+        //Получим параметры запроса
+        const optionsData = await toJSON(prms.queue.sOptions);
        //Формируем запрос для получения BoxId
        let rqpOptions = {
            uri: buildMyOrganizationURL(prms.service.sSrvRoot),
@ -460,7 +470,7 @@ const beforeEvent = async prms => {
            //Получим идентификатор организации по ИНН/КПП контрагента организации
            for (let i in serverResp.Organizations) {
                //Если найдена подходящая организация - запомним идентификатор и выходим из цикла
-                if (serverResp.Organizations[i].Inn == prms.options.inn && serverResp.Organizations[i].Kpp == prms.options.kpp) {
+                if (serverResp.Organizations[i].Inn == optionsData.inn && serverResp.Organizations[i].Kpp == optionsData.kpp) {
                    //Сохраняем полученный ответ
                    sBoxId = serverResp.Organizations[i].Boxes[0].BoxId;
                    //Если задано подразделение
@ -487,7 +497,7 @@ const beforeEvent = async prms => {
            }
            //Не удалось получить ящик текущей организации
            if (!sBoxId) {
-                throw new Error(`Не удалось получить ящик текущей организации с ИНН: ${prms.options.inn} и КПП: ${prms.options.kpp}`);
+                throw new Error(`Не удалось получить ящик текущей организации с ИНН: ${optionsData.inn} и КПП: ${optionsData.kpp}`);
            }
        } catch (e) {
            throw Error(`Ошибка при получении ящика текущей организации: ${e.message}`);
@ -683,12 +693,16 @@ const beforeDocLoad = async prms => {
                entId = "documentId=";
                msgId = "letterId=";
                break;
+            //Загрузка Универсального сообщения
+            case 6:
+                entId = "attachmentId=";
+                break;
            default:
        }
        surl = `${surl}?${msgId}${prms.options.smsgid}&${entId}${prms.options.sentid}`;
        let obj;
        let rblMsg;
-        if (prms.queue.blMsg && prms.options.type != 5) {
+        if (prms.queue.blMsg && (prms.options.type != 5) && (prms.options.type != 6)) {
            //Конвертируем XML из "Парус 8" в понятный "ДИАДОК" JSON
            obj = await toJSON(prms.queue.blMsg.toString());
            rblMsg = Buffer.from(JSON.stringify(obj));
@ -823,6 +837,8 @@ const beforeDepartmentIdGet = async prms => {
 const afterDepartmentIdGet = async prms => {
    let resu = null;
    let organization = {};
+    //Получим параметры запроса
+    const optionsData = await toJSON(prms.queue.sOptions);
    //Действие выполнено успешно
    if (prms.optionsResp.statusCode == 200) {
        try {
@ -830,7 +846,7 @@ const afterDepartmentIdGet = async prms => {
                //Получим организацию не в роуминге (или единственную организацию в роуминге)
                organization = getOrganizations(JSON.parse(prms.queue.blResp.toString()));
                if (!organization) {
-                    throw Error(`Не удалось получить ящик для контрагента с ИНН: ${prms.options.nINN} и КПП: ${prms.options.nKPP}`);
+                    throw Error(`Не удалось получить ящик для контрагента с ИНН: ${optionsData.nINN} и КПП: ${optionsData.nKPP}`);
                }
            } catch (e) {
                //Получим ключ разработчика
@ -838,7 +854,7 @@ const afterDepartmentIdGet = async prms => {
                //Считаем токен доступа из контекста сервиса
                let sToken = prms.service.sCtx;
                //Получим головную организацию по ИНН/КПП
-                organization = await getOrganization(prms.service.sSrvRoot, buildHeaders(sAPIClientId, sToken), prms.options.nINN, prms.options.nKPP);
+                organization = await getOrganization(prms.service.sSrvRoot, buildHeaders(sAPIClientId, sToken), optionsData.nINN, optionsData.nKPP);
            };
            //Преобразуем JSON ответ сервиса "ДИАДОК" в XML, понятный "Парус 8"
            resu = toXML({ root: organization });
--- a/package.json
+++ b/package.json
@ -9,14 +9,14 @@
    },
    "repository": {
        "type": "git",
-        "url": "git+https://github.com/CITKParus/ExchangeService.git"
+        "url": "git+https://git.citpb.ru/CITKParus/P8-ExchangeService.git"
    },
    "author": "CITK Parus",
    "license": "ISC",
    "bugs": {
-        "url": "https://github.com/CITKParus/ExchangeService/issues"
+        "url": "https://git.citpb.ru/CITKParus/P8-ExchangeService/issues"
    },
-    "homepage": "https://github.com/CITKParus/ExchangeService#readme",
+    "homepage": "https://git.citpb.ru/CITKParus/P8-ExchangeService/",
    "dependencies": {
        "body-parser": "^1.19.0",
        "cors": "^2.8.5",
Author	SHA1	Message	Date
Mikhail Chechnev	62786dcc5a	Добавлен параметр "promiseCancellable" для "wrapPromiseTimeout" - обеспечение корректной обработки промисов, не имеющих методов "isPending" и "cancel"	2025-12-01 19:23:46 +03:00
Mikhail Chechnev	9cd779dd64	ЦИТК-255 - исправлена ошибка обработки таймаута для сообщений, отработанных с флагом "bStopPropagation = true" Сервер ошибочно считал такие сообщения "нарушением таймаута" и писал в журнал несуществующие ошибки, в то время как сообщение нормально отрабатывалось.	2025-12-01 19:22:00 +03:00
Mikhail Chechnev	80f50d43a3	ЦИТК-999 - ДИАДОК - исправлена проблема конвертации ИНН в число с последующим исчезновением лидирующего нуля	2025-11-26 12:50:32 +03:00
Mikhail Chechnev	54988da32c	В "package.json" удалены ссылки на GitHub (заменены на git.citpb.ru)	2025-11-26 12:48:48 +03:00
Mikhail Chechnev	f2bb9e1dee	Релиз 2025.11.06	2025-11-12 12:11:40 +03:00
Mikhail Chechnev	a4d26956ab	П8-61674 - исправлена ошибка сохранения ID контрагента при роуминге	2025-11-10 16:17:58 +03:00
Mikhail Chechnev	3721c8b4be	Коннектор "Kafka" - исправлены опечатки в консольной выдаче	2025-11-07 18:24:22 +03:00
Mim	e559b534bf	ЦИТК-255 - Добавление параметра времени ожидания отработки входящего сообщения Reviewed-on: CITKParus/P8-ExchangeService#11	2025-11-05 12:18:11 +03:00
Mikhail Chechnev	5e767ca3eb	Релиз 2025.10.03	2025-10-06 18:39:09 +03:00
Mikhail Chechnev	b94afb06f9	Релиз 2025.08.28	2025-09-01 14:35:35 +03:00
Mikhail Chechnev	b1bc06edc9	Релиз 2025.07.29	2025-08-04 14:56:27 +03:00
Mikhail Chechnev	61dd9e8f1a	ЦИТК-977 - поддержка обмена "Универсальными сообщениями"	2025-08-04 14:35:11 +03:00