config.js

@@ -71,44 +71,6 @@ let inComing = {
     nPoolIncrement: 0
 };
 
-//Параметры подключения к Kafka
-const kafka = [
-    {
-        //Мнемокод сервиса обмена (пусто - использовать по умолчанию)
-        sService: "",
-        //ID клиента-отправителя
-        sClientIdSender: "Parus",
-        //ID клиента-получателя
-        sClientIdRecipient: "Parus",
-        //Группа получателя
-        sGroupId: "Parus",
-        //Время ожидания успешного подключения (мс)
-        nConnectionTimeout: 5000,
-        //Необходимость попытки переподключения при потере соединения
-        bRestartOnFailure: true,
-        //Время максимального ожидания между попытками переподключения (мс)
-        nMaxRetryTime: 20000,
-        //Время ожидания между попытками переподключения (мс)
-        nInitialRetryTime: 10000
-    }
-];
-
-//Параметры подключения к MQTT
-const mqtt = [
-    {
-        //Мнемокод сервиса обмена (пусто - использовать по умолчанию)
-        sService: "",
-        //ID клиента-отправителя
-        sClientIdSender: "Parus",
-        //ID клиента-получателя (если равен sClientIdSender, то отправленные сообщения будут игнорироваться)
-        sClientIdRecipient: "ParusRecipient",
-        //Время ожидания успешного подключения (мс)
-        nConnectTimeout: 5000,
-        //Время ожидания между попытками переподключения (мс)
-        nReconnectPeriod: 10000
-    }
-];
-
 //Параметры отправки E-Mail уведомлений
 let mail = {
     //Адреc сервера SMTP
@@ -136,7 +98,5 @@ module.exports = {
     dbConnect,
     outGoing,
     inComing,
-    kafka,
-    mqtt,
     mail
 };

config_default.js

@@ -71,44 +71,6 @@ let inComing = {
     nPoolIncrement: 0
 };
 
-//Параметры подключения к Kafka
-const kafka = [
-    {
-        //Мнемокод сервиса обмена (пусто - использовать по умолчанию)
-        sService: "",
-        //ID клиента-отправителя
-        sClientIdSender: "Parus",
-        //ID клиента-получателя
-        sClientIdRecipient: "Parus",
-        //Группа получателя
-        sGroupId: "Parus",
-        //Время ожидания успешного подключения (мс)
-        nConnectionTimeout: 5000,
-        //Необходимость попытки переподключения при потере соединения
-        bRestartOnFailure: true,
-        //Время максимального ожидания между попытками переподключения (мс)
-        nMaxRetryTime: 20000,
-        //Время ожидания между попытками переподключения (мс)
-        nInitialRetryTime: 10000
-    }
-];
-
-//Параметры подключения к MQTT
-const mqtt = [
-    {
-        //Мнемокод сервиса обмена (пусто - использовать по умолчанию)
-        sService: "",
-        //ID клиента-отправителя
-        sClientIdSender: "Parus",
-        //ID клиента-получателя (если равен sClientIdSender, то отправленные сообщения будут игнорироваться)
-        sClientIdRecipient: "ParusRecipient",
-        //Время ожидания успешного подключения (мс)
-        nConnectTimeout: 5000,
-        //Время ожидания между попытками переподключения (мс)
-        nReconnectPeriod: 10000
-    }
-];
-
 //Параметры отправки E-Mail уведомлений
 let mail = {
     //Адреc сервера SMTP
@@ -136,7 +98,5 @@ module.exports = {
     dbConnect,
     outGoing,
     inComing,
-    kafka,
-    mqtt,
     mail
 };

core/app.js

@@ -96,7 +96,7 @@ class ParusAppServer {
         //Запускаем обслуживание очереди входящих
         await this.logger.info("Запуск обработчика очереди входящих сообщений...");
         try {
-            await this.inQ.startProcessing({ services: this.services });
+            this.inQ.startProcessing({ services: this.services });
         } catch (e) {
             await this.logger.error(`Ошибка запуска обработчика очереди входящих сообщений: ${makeErrorText(e)}`);
             await this.stop();
@@ -196,7 +196,9 @@ class ParusAppServer {
         //Если настройки верны - будем стартовать
         if (!sCheckResult) {
             //Протоколируем версию и релиз
-            await this.logger.info(`Версия сервера приложений: ${prms.config.common.sVersion}, релиз: ${prms.config.common.sRelease}`);
+            await this.logger.info(
+                `Версия сервера приложений: ${prms.config.common.sVersion}, релиз: ${prms.config.common.sRelease}`
+            );
             //Создаём подключение к БД
             this.dbConn = new db.DBConnector({
                 connectSettings: {
@@ -217,9 +219,7 @@ class ParusAppServer {
                 dbConn: this.dbConn,
                 logger: this.logger,
                 notifier: this.notifier,
-                sProxy: prms.config.outGoing.sProxy,
+                sProxy: prms.config.outGoing.sProxy
-                kafka: prms.config.kafka,
-                mqtt: prms.config.mqtt
             });
             //Создаём обработчик очереди входящих
             this.inQ = new iq.InQueue({
@@ -227,9 +227,7 @@ class ParusAppServer {
                 inComing: prms.config.inComing,
                 dbConn: this.dbConn,
                 logger: this.logger,
-                notifier: this.notifier,
+                notifier: this.notifier
-                kafka: prms.config.kafka,
-                mqtt: prms.config.mqtt
             });
             //Создаём контроллер доступности удалённых сервисов
             this.srvAvlCtrl = new sac.ServiceAvailableController({

core/in_queue.js

@@ -13,18 +13,7 @@ const express = require("express"); //WEB-сервер Express
 const cors = require("cors"); //Управление заголовками безопасности для WEB-сервера Express
 const bodyParser = require("body-parser"); //Модуль для Express (разбор тела входящего запроса)
 const { ServerError } = require("./server_errors"); //Типовая ошибка
-const {
+const { makeErrorText, validateObject, buildURL, getAppSrvFunction, buildOptionsXML, parseOptionsXML, deepMerge } = require("./utils"); //Вспомогательные функции
-    makeErrorText,
-    validateObject,
-    buildURL,
-    getAppSrvFunction,
-    buildOptionsXML,
-    parseOptionsXML,
-    deepMerge,
-    getKafkaConnectionSettings,
-    getMQTTConnectionSettings,
-    getURLProtocol
-} = require("./utils"); //Вспомогательные функции
 const { NINC_EXEC_CNT_YES, NIS_ORIGINAL_NO, NIS_ORIGINAL_YES } = require("../models/prms_db_connector"); //Схемы валидации параметров функций модуля взаимодействия с БД
 const objInQueueSchema = require("../models/obj_in_queue"); //Схема валидации сообщений обмена с бработчиком очереди входящих сообщений
 const objServiceSchema = require("../models/obj_service"); //Схемы валидации сервиса
@@ -39,8 +28,6 @@ const {
     SERR_DB_SERVER,
     SERR_UNAUTH
 } = require("./constants"); //Общесистемные константы
-const { subscribeMQTT } = require("./mqtt_connector"); //Модуль для работы с MQTT
-const { subscribeKafka } = require("./kafka_connector"); //Модуль для работы с Kafka
 
 //--------------------------
 // Глобальные идентификаторы
@@ -84,13 +71,6 @@ class InQueue extends EventEmitter {
             this.webApp.options("*", cors());
             //WEB-сервер
             this.srv = null;
-            //Параметры подключения к Kafka
-            this.kafka = prms.kafka;
-            //Параметры подключения к MQTT
-            this.mqtt = prms.mqtt;
-            //Внешние подключения
-            this.kafkaConnections = [];
-            this.mqttConnections = [];
         } else {
             throw new ServerError(SERR_OBJECT_BAD_INTERFACE, sCheckResult);
         }
@@ -380,326 +360,8 @@ class InQueue extends EventEmitter {
             throw new ServerError(SERR_OBJECT_BAD_INTERFACE, sCheckResult);
         }
     }
-
-    //Обработка сообщения kafka
-    async processKafkaMessage({ message, service, fn }) {
-        //Буфер для сообщения очереди
-        let q = null;
-        try {
-            //Тело сообщения и ответ на него
-            let blMsg = null;
-            let blResp = null;
-            //Параметры сообщения и ответа на него
-            let options = {};
-            let optionsResp = {};
-            //Флаг прекращения обработки сообщения
-            let bStopPropagation = false;
-            //Получим тело сообщения
-            blMsg = message.value ? message.value : null;
-            //Определимся с параметрами сообщения полученными от внешней системы
-            options = {
-                method: fn.sFnPrmsType,
-                headers: _.cloneDeep(message.headers)
-            };
-            //Кладём сообщение в очередь
-            q = await this.dbConn.putQueue({
-                nServiceFnId: fn.nId,
-                sOptions: buildOptionsXML({ options }),
-                blMsg
-            });
-            //Скажем что пришло новое входящее сообщение
-            await this.logger.info(
-                `Новое входящее Kafka-сообщение для функции ${fn.sCode} (${buildURL({
-                    sSrvRoot: service.sSrvRoot,
-                    sFnURL: fn.sFnURL
-                })})`,
-                { nQueueId: q.nId }
-            );
-            //Вызываем обработчик со стороны БД (если он есть)
-            if (fn.sPrcResp) {
-                //Фиксируем начало исполнения сервером БД - в статусе сообщения
-                q = await this.dbConn.setQueueState({
-                    nQueueId: q.nId,
-                    nExecState: objQueueSchema.NQUEUE_EXEC_STATE_DB
-                });
-                //Вызов обработчика БД
-                let prcRes = await this.dbConn.execQueueDBPrc({ nQueueId: q.nId });
-                //Если результат - ошибка пробрасываем её
-                if (prcRes.sResult == objQueueSchema.SPRC_RESP_RESULT_ERR) throw new ServerError(SERR_DB_SERVER, prcRes.sMsg);
-                //Если результат - ошибка аутентификации, то и её пробрасываем, но с правильным кодом
-                if (prcRes.sResult == objQueueSchema.SPRC_RESP_RESULT_UNAUTH) throw new ServerError(SERR_UNAUTH, prcRes.sMsg || "Нет аутентификации");
-                //Выставим статус сообщению очереди - исполнено обработчиком БД
-                q = await this.dbConn.setQueueState({
-                    nQueueId: q.nId,
-                    nExecState: objQueueSchema.NQUEUE_EXEC_STATE_DB_OK
-                });
-                //Считаем ответ полученный от системы
-                let qData = await this.dbConn.getQueueResp({ nQueueId: q.nId });
-                blResp = qData.blResp;
-            }
-            //Выполняем обработчик "После" (если он есть)
-            if (bStopPropagation === false && fn.sAppSrvAfter) {
-                //Выставим статус сообщению очереди - исполняется сервером приложений
-                q = await this.dbConn.setQueueState({
-                    nQueueId: q.nId,
-                    nExecState: objQueueSchema.NQUEUE_EXEC_STATE_APP
-                });
-                //Выполняем
-                const fnAfter = getAppSrvFunction(fn.sAppSrvAfter);
-                let resAfter = null;
-                try {
-                    let resAfterPrms = { res: { body: blMsg }, service: service, function: fn };
-                    resAfterPrms.queue = _.cloneDeep(q);
-                    resAfterPrms.queue.blMsg = blMsg;
-                    resAfterPrms.queue.blResp = blResp;
-                    resAfterPrms.options = _.cloneDeep(options);
-                    resAfterPrms.optionsResp = _.cloneDeep(optionsResp);
-                    resAfterPrms.dbConn = this.dbConn;
-                    resAfterPrms.notifier = this.notifier;
-                    resAfter = await fnAfter(resAfterPrms);
-                } catch (e) {
-                    throw new ServerError(SERR_APP_SERVER_AFTER, e.message);
-                }
-                //Проверяем структуру ответа функции предобработки
-                if (resAfter) {
-                    let sCheckResult = validateObject(
-                        resAfter,
-                        objInQueueSchema.InQueueProcessorFnAfter,
-                        "Результат функции постобработки входящего сообщения"
-                    );
-                    //Если структура ответа в норме
-                    if (!sCheckResult) {
-                        //Выставим статус сообщению очереди - исполнено сервером приложений
-                        q = await this.dbConn.setQueueState({
-                            nQueueId: q.nId,
-                            nExecState: objQueueSchema.NQUEUE_EXEC_STATE_APP_OK
-                        });
-                        //Фиксируем результат исполнения "После" - ответ системы
-                        if (!_.isUndefined(resAfter.blResp)) {
-                            blResp = resAfter.blResp;
-                            q = await this.dbConn.setQueueResp({
-                                nQueueId: q.nId,
-                                blResp,
-                                nIsOriginal: NIS_ORIGINAL_NO
-                            });
-                        }
-                        //Фиксируем результат исполнения "После" - параметры ответа на запрос
-                        if (!_.isUndefined(resAfter.optionsResp)) {
-                            optionsResp = deepMerge(optionsResp, resAfter.optionsResp);
-                            let sOptionsResp = buildOptionsXML({ options: optionsResp });
-                            q = await this.dbConn.setQueueOptionsResp({ nQueueId: q.nId, sOptionsResp });
-                        }
-                        //Если пришел флаг ошибочной аутентификации и он положительный - то это ошибка, дальше ничего не делаем
-                        if (!_.isUndefined(resAfter.bUnAuth)) if (resAfter.bUnAuth === true) throw new ServerError(SERR_UNAUTH, "Нет аутентификации");
-                    } else {
-                        //Или расскажем об ошибке
-                        throw new ServerError(SERR_OBJECT_BAD_INTERFACE, sCheckResult);
-                    }
-                }
-            }
-            //Фиксируем успех обработки - в протоколе работы сервиса
-            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 = `Ошибка обработки входящего сообщения сервером приложений для функции "${fn.sCode}" сервиса "${service.sCode}"`;
-            let sMessage = makeErrorText(e);
-            //Если сообщение очереди успели создать
-            if (q) {
-                //Фиксируем ошибку обработки сервером приложений - в статусе сообщения
-                q = await this.dbConn.setQueueState({
-                    nQueueId: q.nId,
-                    sExecMsg: sMessage,
-                    nIncExecCnt: NINC_EXEC_CNT_YES,
-                    nExecState: objQueueSchema.NQUEUE_EXEC_STATE_ERR
-                });
-                //Фиксируем ошибку обработки сервером приложений - в протоколе работы сервиса
-                await this.logger.error(`Ошибка обработки входящего сообщения ${q.nId} сервером приложений: ${sMessage}`, { nQueueId: q.nId });
-                //Добавим чуть больше информации в тему сообщения
-                sSubject = `Ошибка обработки входящего сообщения ${q.nId} сервером приложений для функции "${fn.sCode}" сервиса "${service.sCode}"`;
-            } else {
-                //Ограничимся общей ошибкой
-                await this.logger.error(sMessage, {
-                    nServiceId: service.nId,
-                    nServiceFnId: fn.nId
-                });
-            }
-            //Если для функции-обработчика указан признак необходимости оповещения об ошибках
-            if (fn.nErrNtfSign == objServiceFnSchema.NERR_NTF_SIGN_YES) {
-                //Отправим уведомление об ошибке отработки в почту
-                await this.notifier.addMessage({
-                    sTo: fn.sErrNtfMail,
-                    sSubject,
-                    sMessage
-                });
-            }
-        }
-    }
-
-    //Обработка сообщения
-    async processMQTTMessage({ message, service, fn }) {
-        //Буфер для сообщения очереди
-        let q = null;
-        try {
-            //Тело сообщения и ответ на него
-            let blMsg = null;
-            let blResp = null;
-            //Параметры сообщения и ответа на него
-            let options = {};
-            let optionsResp = {};
-            //Получим тело сообщения
-            blMsg = message ? message : null;
-            //Определимся с параметрами сообщения полученными от внешней системы
-            options = {
-                method: fn.sFnPrmsType
-            };
-            //Кладём сообщение в очередь
-            q = await this.dbConn.putQueue({
-                nServiceFnId: fn.nId,
-                sOptions: buildOptionsXML({ options }),
-                blMsg
-            });
-            //Скажем что пришло новое входящее сообщение
-            await this.logger.info(
-                `Новое входящее MQTT-сообщение для функции ${fn.sCode} (${buildURL({
-                    sSrvRoot: service.sSrvRoot,
-                    sFnURL: fn.sFnURL
-                })})`,
-                { nQueueId: q.nId }
-            );
-            //Вызываем обработчик со стороны БД (если он есть)
-            if (fn.sPrcResp) {
-                //Фиксируем начало исполнения сервером БД - в статусе сообщения
-                q = await this.dbConn.setQueueState({
-                    nQueueId: q.nId,
-                    nExecState: objQueueSchema.NQUEUE_EXEC_STATE_DB
-                });
-                //Вызов обработчика БД
-                let prcRes = await this.dbConn.execQueueDBPrc({ nQueueId: q.nId });
-                //Если результат - ошибка пробрасываем её
-                if (prcRes.sResult == objQueueSchema.SPRC_RESP_RESULT_ERR) throw new ServerError(SERR_DB_SERVER, prcRes.sMsg);
-                //Если результат - ошибка аутентификации, то и её пробрасываем, но с правильным кодом
-                if (prcRes.sResult == objQueueSchema.SPRC_RESP_RESULT_UNAUTH) throw new ServerError(SERR_UNAUTH, prcRes.sMsg || "Нет аутентификации");
-                //Выставим статус сообщению очереди - исполнено обработчиком БД
-                q = await this.dbConn.setQueueState({
-                    nQueueId: q.nId,
-                    nExecState: objQueueSchema.NQUEUE_EXEC_STATE_DB_OK
-                });
-                //Считаем ответ полученный от системы
-                let qData = await this.dbConn.getQueueResp({ nQueueId: q.nId });
-                blResp = qData.blResp;
-            }
-            //Выполняем обработчик "После" (если он есть)
-            if (fn.sAppSrvAfter) {
-                //Выставим статус сообщению очереди - исполняется сервером приложений
-                q = await this.dbConn.setQueueState({
-                    nQueueId: q.nId,
-                    nExecState: objQueueSchema.NQUEUE_EXEC_STATE_APP
-                });
-                //Выполняем
-                const fnAfter = getAppSrvFunction(fn.sAppSrvAfter);
-                let resAfter = null;
-                try {
-                    let resAfterPrms = { res: { body: message }, service: service, function: fn };
-                    resAfterPrms.queue = _.cloneDeep(q);
-                    resAfterPrms.queue.blMsg = blMsg;
-                    resAfterPrms.queue.blResp = blResp;
-                    resAfterPrms.options = _.cloneDeep(options);
-                    resAfterPrms.optionsResp = _.cloneDeep(optionsResp);
-                    resAfterPrms.dbConn = this.dbConn;
-                    resAfterPrms.notifier = this.notifier;
-                    resAfter = await fnAfter(resAfterPrms);
-                } catch (e) {
-                    throw new ServerError(SERR_APP_SERVER_AFTER, e.message);
-                }
-                //Проверяем структуру ответа функции предобработки
-                if (resAfter) {
-                    let sCheckResult = validateObject(
-                        resAfter,
-                        objInQueueSchema.InQueueProcessorFnAfter,
-                        "Результат функции постобработки входящего сообщения"
-                    );
-                    //Если структура ответа в норме
-                    if (!sCheckResult) {
-                        //Выставим статус сообщению очереди - исполнено сервером приложений
-                        q = await this.dbConn.setQueueState({
-                            nQueueId: q.nId,
-                            nExecState: objQueueSchema.NQUEUE_EXEC_STATE_APP_OK
-                        });
-                        //Фиксируем результат исполнения "После" - ответ системы
-                        if (!_.isUndefined(resAfter.blResp)) {
-                            blResp = resAfter.blResp;
-                            q = await this.dbConn.setQueueResp({
-                                nQueueId: q.nId,
-                                blResp,
-                                nIsOriginal: NIS_ORIGINAL_NO
-                            });
-                        }
-                        //Фиксируем результат исполнения "После" - параметры ответа на запрос
-                        if (!_.isUndefined(resAfter.optionsResp)) {
-                            optionsResp = deepMerge(optionsResp, resAfter.optionsResp);
-                            let sOptionsResp = buildOptionsXML({ options: optionsResp });
-                            q = await this.dbConn.setQueueOptionsResp({ nQueueId: q.nId, sOptionsResp });
-                        }
-                        //Если пришел флаг ошибочной аутентификации и он положительный - то это ошибка, дальше ничего не делаем
-                        if (!_.isUndefined(resAfter.bUnAuth)) if (resAfter.bUnAuth === true) throw new ServerError(SERR_UNAUTH, "Нет аутентификации");
-                    } else {
-                        //Или расскажем об ошибке
-                        throw new ServerError(SERR_OBJECT_BAD_INTERFACE, sCheckResult);
-                    }
-                }
-            }
-            //Фиксируем успех обработки - в протоколе работы сервиса
-            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 = `Ошибка обработки входящего сообщения сервером приложений для функции "${fn.sCode}" сервиса "${service.sCode}"`;
-            let sMessage = makeErrorText(e);
-            //Если сообщение очереди успели создать
-            if (q) {
-                //Фиксируем ошибку обработки сервером приложений - в статусе сообщения
-                q = await this.dbConn.setQueueState({
-                    nQueueId: q.nId,
-                    sExecMsg: sMessage,
-                    nIncExecCnt: NINC_EXEC_CNT_YES,
-                    nExecState: objQueueSchema.NQUEUE_EXEC_STATE_ERR
-                });
-                //Фиксируем ошибку обработки сервером приложений - в протоколе работы сервиса
-                await this.logger.error(`Ошибка обработки входящего сообщения ${q.nId} сервером приложений: ${sMessage}`, { nQueueId: q.nId });
-                //Добавим чуть больше информации в тему сообщения
-                sSubject = `Ошибка обработки входящего сообщения ${q.nId} сервером приложений для функции "${fn.sCode}" сервиса "${service.sCode}"`;
-            } else {
-                //Ограничимся общей ошибкой
-                await this.logger.error(sMessage, {
-                    nServiceId: service.nId,
-                    nServiceFnId: fn.nId
-                });
-            }
-            //Если для функции-обработчика указан признак необходимости оповещения об ошибках
-            if (fn.nErrNtfSign == objServiceFnSchema.NERR_NTF_SIGN_YES) {
-                //Отправим уведомление об ошибке отработки в почту
-                await this.notifier.addMessage({
-                    sTo: fn.sErrNtfMail,
-                    sSubject,
-                    sMessage
-                });
-            }
-        }
-    }
-
     //Запуск обработки очереди входящих сообщений
-    async startProcessing(prms) {
+    startProcessing(prms) {
         //Проверяем структуру переданного объекта для старта
         let sCheckResult = validateObject(prms, prmsInQueueSchema.startProcessing, "Параметры функции запуска обработки очереди входящих сообщений");
         //Если структура объекта в норме
@@ -717,136 +379,55 @@ class InQueue extends EventEmitter {
             //Конфигурируем сервер - обработка тела сообщения
             this.webApp.use(bodyParser.raw({ limit: `${this.inComing.nMsgMaxSize}mb`, type: "*/*" }));
             //Конфигурируем сервер - обходим все сервисы, работающие на приём сообщений
-            _.forEach(
+            _.forEach(_.filter(this.services, { nSrvType: objServiceSchema.NSRV_TYPE_RECIVE }), srvs => {
-                _.filter(this.services, srv => {
+                //Для любых запросов к корневому адресу сервиса - ответ о том, что это за сервис, и что он работает
-                    return (
+                this.webApp.all(srvs.sSrvRoot, (req, res) => {
-                        srv.nSrvType === objServiceSchema.NSRV_TYPE_RECIVE &&
+                    res.status(200).send(
-                        [objServiceSchema.SPROTOCOL_HTTP, objServiceSchema.SPROTOCOL_HTTPS].includes(getURLProtocol(srv.sSrvRoot))
+                        `<html><body><center><br><h1>Сервер приложений ПП Парус 8<br>(${this.common.sVersion} релиз ${this.common.sRelease})</h1><h3>Сервис: ${srvs.sName}</h3></center></body></html>`
                     );
-                }),
+                });
-                srvs => {
+                //Для всех статических функций сервиса...
-                    //Для любых запросов к корневому адресу сервиса - ответ о том, что это за сервис, и что он работает
+                _.forEach(
-                    this.webApp.all(srvs.sSrvRoot, (req, res) => {
+                    _.filter(srvs.functions, fn => fn.sFnURL.startsWith("@")),
-                        res.status(200).send(
+                    fn => {
-                            `<html><body><center><br><h1>Сервер приложений ПП Парус 8<br>(${this.common.sVersion} релиз ${this.common.sRelease})</h1><h3>Сервис: ${srvs.sName}</h3></center></body></html>`
+                        this.webApp.use(
+                            buildURL({ sSrvRoot: srvs.sSrvRoot, sFnURL: fn.sFnURL.substr(1) }),
+                            express.static(`${this.inComing.sStaticDir}/${fn.sFnURL.substr(1)}`)
                         );
-                    });
+                    }
-                    //Для всех статических функций сервиса...
+                );
-                    _.forEach(
+                //Для всех функций сервиса (кроме статических)...
-                        _.filter(srvs.functions, fn => fn.sFnURL.startsWith("@")),
+                _.forEach(
-                        fn => {
+                    _.filter(srvs.functions, fn => !fn.sFnURL.startsWith("@")),
-                            this.webApp.use(
+                    fn => {
-                                buildURL({ sSrvRoot: srvs.sSrvRoot, sFnURL: fn.sFnURL.substr(1) }),
+                        //...собственный обработчик, в зависимости от указанного способа передачи параметров
-                                express.static(`${this.inComing.sStaticDir}/${fn.sFnURL.substr(1)}`)
+                        this.webApp[fn.sFnPrmsType.toLowerCase()](buildURL({ sSrvRoot: srvs.sSrvRoot, sFnURL: fn.sFnURL }), async (req, res) => {
-                            );
+                            try {
-                        }
+                                //Вызываем обработчик
-                    );
+                                await this.processMessage({ req, res, service: srvs, function: fn });
-                    //Для всех функций сервиса (кроме статических)...
+                            } catch (e) {
-                    _.forEach(
-                        _.filter(srvs.functions, fn => !fn.sFnURL.startsWith("@")),
-                        fn => {
-                            //...собственный обработчик, в зависимости от указанного способа передачи параметров
-                            this.webApp[fn.sFnPrmsType.toLowerCase()](buildURL({ sSrvRoot: srvs.sSrvRoot, sFnURL: fn.sFnURL }), async (req, res) => {
-                                try {
-                                    //Вызываем обработчик
-                                    await this.processMessage({ req, res, service: srvs, function: fn });
-                                } catch (e) {
-                                    //Протоколируем в журнал работы сервера
-                                    await this.logger.error(makeErrorText(e), {
-                                        nServiceId: srvs.nId,
-                                        nServiceFnId: fn.nId
-                                    });
-                                    //Отправим ошибку клиенту
-                                    res.status(500).send(makeErrorText(e));
-                                }
-                            });
-                            //...и собственный обработчик ошибок
-                            this.webApp.use(buildURL({ sSrvRoot: srvs.sSrvRoot, sFnURL: fn.sFnURL }), async (err, req, res, next) => {
                                 //Протоколируем в журнал работы сервера
-                                await this.logger.error(makeErrorText(new ServerError(SERR_WEB_SERVER, err.message)), {
+                                await this.logger.error(makeErrorText(e), {
                                     nServiceId: srvs.nId,
                                     nServiceFnId: fn.nId
                                 });
                                 //Отправим ошибку клиенту
-                                res.status(500).send(makeErrorText(new ServerError(SERR_WEB_SERVER, err.message)));
+                                res.status(500).send(makeErrorText(e));
-                            });
-                        }
-                    );
-                }
-            );
-            //Инициализируем настройки подключения
-            let connectionSettings = null;
-            //Считываем прием сообщений по Kafka
-            let kafkaSrvs = this.services.filter(srv => {
-                return srv.nSrvType === objServiceSchema.NSRV_TYPE_RECIVE && getURLProtocol(srv.sSrvRoot) === objServiceSchema.SPROTOCOL_KAFKA;
-            });
-            //Если есть сервисы с приемом сообщений по Kafka
-            if (kafkaSrvs.length !== 0) {
-                //Обходим данные сервисы
-                for (let srv of kafkaSrvs) {
-                    //Если у сервиса обмена есть функции
-                    if (srv.functions.length !== 0) {
-                        //Считываем настройки подключения к Kafka
-                        connectionSettings = getKafkaConnectionSettings(srv.sCode, this.kafka);
-                        //Если настройки подключения считаны
-                        if (connectionSettings) {
-                            //Подключаемся и подписываемся на соответствующий брокер
-                            let connectionKafka = await subscribeKafka({
-                                settings: connectionSettings,
-                                service: srv,
-                                processKafkaMessage: prms => this.processKafkaMessage(prms),
-                                logger: this.logger
-                            });
-                            //Если подключение было создано
-                            if (connectionKafka) {
-                                //Добавляем в общий список подключений kafka
-                                this.kafkaConnections.push(connectionKafka);
                             }
-                        } else {
+                        });
-                            await this.logger.error(
+                        //...и собственный обработчик ошибок
-                                `Ошибка получения настроек подключения к Kafka для сервиса "${srv.sCode}". Необходимо проверить соответствующий параметр ("kafka") файла конфигурации сервиса приложений ("config.js").`
+                        this.webApp.use(buildURL({ sSrvRoot: srvs.sSrvRoot, sFnURL: fn.sFnURL }), async (err, req, res, next) => {
-                            );
+                            //Протоколируем в журнал работы сервера
-                        }
+                            await this.logger.error(makeErrorText(new ServerError(SERR_WEB_SERVER, err.message)), {
+                                nServiceId: srvs.nId,
+                                nServiceFnId: fn.nId
+                            });
+                            //Отправим ошибку клиенту
+                            res.status(500).send(makeErrorText(new ServerError(SERR_WEB_SERVER, err.message)));
+                        });
                     }
-                }
-            }
-            //Считываем прием сообщений по MQTT
-            let mqttSrvs = this.services.filter(srv => {
-                return (
-                    srv.nSrvType === objServiceSchema.NSRV_TYPE_RECIVE &&
-                    [objServiceSchema.SPROTOCOL_MQTT, objServiceSchema.SPROTOCOL_MQTTS].includes(getURLProtocol(srv.sSrvRoot))
                 );
             });
-            //Если есть сервисы с приемом сообщений по MQTT
-            if (mqttSrvs.length !== 0) {
-                //Обходим данные сервисы
-                for (let srv of mqttSrvs) {
-                    //Если у сервиса обмена есть функции
-                    if (srv.functions.length !== 0) {
-                        //Считываем настройки подключения к MQTT
-                        connectionSettings = getMQTTConnectionSettings(srv.sCode, this.mqtt);
-                        //Если настройки подключения считаны
-                        if (connectionSettings) {
-                            //Подключаемся и подписываемся на соответствующий брокер
-                            let connectionMQTT = await subscribeMQTT({
-                                settings: connectionSettings,
-                                service: srv,
-                                processMQTTMessage: prms => this.processMQTTMessage(prms),
-                                logger: this.logger
-                            });
-                            //Если подключение было создано
-                            if (connectionMQTT) {
-                                //Добавляем в общий список подключений kafka
-                                this.mqttConnections.push(connectionMQTT);
-                            }
-                        } else {
-                            await this.logger.error(
-                                `Ошибка получения настроек подключения к MQTT для сервиса "${srv.sCode}". Необходимо проверить соответствующий параметр ("mqtt") файла конфигурации сервиса приложений ("config.js").`
-                            );
-                        }
-                    }
-                }
-            }
             //Запросы на адреса, не входящие в состав объявленных сервисов - 404 NOT FOUND
             this.webApp.use("*", (req, res) => {
                 res.status(404).send(
@@ -872,39 +453,10 @@ class InQueue extends EventEmitter {
             throw new ServerError(SERR_OBJECT_BAD_INTERFACE, sCheckResult);
         }
     }
-
-    //Закрытие подключений, если они есть
-    stopConnections() {
-        //Если у нас есть соединения с MQTT
-        if (this.mqttConnections.length !== 0) {
-            //Закрываем их
-            _.forEach(this.mqttConnections, async connection => {
-                try {
-                    await connection.end();
-                } catch (e) {
-                    await this.logger.error(`Ошибка завершения MQTT подключения: ${makeErrorText(e)}`);
-                }
-            });
-        }
-        //Если у нас есть соединения с Kafka
-        if (this.kafkaConnections.length !== 0) {
-            //Закрываем их
-            _.forEach(this.kafkaConnections, async connection => {
-                try {
-                    await connection.disconnect();
-                } catch (e) {
-                    await this.logger.error(`Ошибка завершения Kafka подключения: ${makeErrorText(e)}`);
-                }
-            });
-        }
-    }
-
     //Остановка обработки очереди исходящих сообщений
     stopProcessing() {
         //Выставляем флаг неработы
         this.bWorking = false;
-        //Закрываем подключения, если они есть
-        this.stopConnections();
         //Останавливаем WEB-сервер (если создавался)
         if (this.srv) {
             this.srv.close(() => {

core/kafka_connector.js

@@ -1,119 +0,0 @@
-/*
-  Сервис интеграции ПП Парус 8 с WEB API    
-  Модуль ядра: обработчик kafka сообщений
-*/
-
-//----------------------
-// Подключение библиотек
-//----------------------
-
-const { makeErrorText, getKafkaBroker, getKafkaAuth } = require("./utils"); //Вспомогательные функции
-const { Kafka, logLevel } = require("kafkajs"); //Работа с Kafka
-
-//------------
-// Тело модуля
-//------------
-
-//Отправка сообщения Kafka
-const publishKafka = async ({ settings, url, auth, topic, message }) => {
-    //Иницализируем подключение к Kafka
-    let kafka = new Kafka({
-        clientId: settings.sClientIdSender,
-        brokers: [url],
-        connectionTimeout: settings.nConnectionTimeout,
-        logLevel: logLevel.NOTHING,
-        ...auth
-    });
-    //Инициализируем продюсера
-    let producer = kafka.producer();
-    //Подключаемся к Kafka
-    await producer.connect();
-    //Отправляем сообщение
-    await producer.send({ topic: topic, messages: [{ value: message }] });
-    //Отключаемся
-    await producer.disconnect();
-    //Возвращаем статус успешной отправки
-    return { statusCode: 200 };
-};
-
-//Получение MQTT сообщений
-const subscribeKafka = async ({ settings, service, processKafkaMessage, logger }) => {
-    try {
-        //Признак необходимости вывода сообщения о потере соединения
-        let bLogLostConnection = true;
-        //Получаем брокера по URL сервиса
-        let sBroker = getKafkaBroker(service.sSrvRoot);
-        //Иницализируем подключение к Kafka
-        let client = new Kafka({
-            clientId: settings.sClientIdRecipient,
-            brokers: [sBroker],
-            connectionTimeout: settings.nConnectionTimeout,
-            ...getKafkaAuth(service.sSrvUser, service.sSrvPass),
-            logLevel: logLevel.NOTHING,
-            retry: {
-                retries: 0,
-                maxRetryTime: settings.nMaxRetryTime,
-                initialRetryTime: settings.nInitialRetryTime,
-                restartOnFailure: error => {
-                    return new Promise(resolve => {
-                        //Если требуется вывести ошибку
-                        if (bLogLostConnection) {
-                            //Выводим ошибку
-                            logger.error(`Соединение с Kafka потеряно (${sBroker}): ${makeErrorText(error)}`);
-                            //Сбрасываем признак необходимости вывода ошибки
-                            bLogLostConnection = false;
-                        }
-                        resolve(settings.bRestartOnFailure);
-                    });
-                }
-            }
-        });
-        //Инициализируем получателя
-        let consumer = client.consumer({ groupId: settings.sGroupId });
-        //Устанавливаем прослушивание
-        await consumer.connect();
-        consumer.subscribe({
-            topics: service.functions.map(fn => {
-                return fn.sFnURL;
-            })
-        });
-        //Запускаем прослушивание необходимых топиков
-        consumer.run({
-            eachMessage: async ({ topic, message }) => {
-                try {
-                    //Вызываем обработчик
-                    processKafkaMessage({
-                        message,
-                        service,
-                        fn: service.functions.find(fn => {
-                            return fn.sFnURL === topic;
-                        })
-                    });
-                } catch (e) {
-                    await logger.error(`Ошибка обработки исходящего сообщения Kafka: ${makeErrorText(e)}`);
-                }
-            }
-        });
-        //Отслеживаем соединение
-        consumer.on(consumer.events.CONNECT, () => {
-            //Если сообщение о потере соединения уже выводилось
-            if (!bLogLostConnection) {
-                //Сообщим о восстановлении соединения
-                logger.info(`Соединение с Kafka восстановлено (${sBroker})`);
-                //Устанавливаем признак сообщения о потере соединения
-                bLogLostConnection = true;
-            }
-        });
-        //Возвращаем соединение
-        return consumer;
-    } catch (e) {
-        await logger.error(`Ошибка запуска обработчика очереди исходящих сообщений Kafka: ${makeErrorText(e)}`);
-    }
-};
-
-//-----------------
-// Интерфейс модуля
-//-----------------
-
-exports.publishKafka = publishKafka;
-exports.subscribeKafka = subscribeKafka;

core/mqtt_connector.js

@@ -1,87 +0,0 @@
-/*
-  Сервис интеграции ПП Парус 8 с WEB API    
-  Модуль ядра: обработчик mqtt сообщений
-*/
-
-//----------------------
-// Подключение библиотек
-//----------------------
-
-const { makeErrorText } = require("./utils"); //Вспомогательные функции
-const mqtt = require("mqtt"); //Работа с MQTT
-
-//------------
-// Тело модуля
-//------------
-
-//Отправка MQTT сообщения
-const publishMQTT = async ({ settings, url, auth, topic, message }) => {
-    //Инициализируем подключение
-    const client = await mqtt.connectAsync(url, {
-        clientId: settings.sClientIdSender,
-        clean: true,
-        connectTimeout: settings.nConnectTimeout,
-        username: auth.user,
-        password: auth.pass,
-        reconnectPeriod: settings.nReconnectPeriod
-    });
-    //Отправляем сообщение
-    await client.publishAsync(topic, message);
-    //Закрываем подключение
-    await client.endAsync();
-    //Возвращаем статус успешной отправки
-    return { statusCode: 200 };
-};
-
-//Получение MQTT сообщений
-const subscribeMQTT = async ({ settings, service, processMQTTMessage, logger }) => {
-    try {
-        //Инициализируем строку подключения
-        let sBroker = service.sSrvRoot;
-        //Инициализируем подключение
-        const client = await mqtt.connectAsync(sBroker, {
-            clientId: settings.sClientIdRecipient,
-            clean: true,
-            connectTimeout: settings.nConnectTimeout,
-            username: service.sSrvUser,
-            password: service.sSrvPass,
-            reconnectPeriod: settings.nReconnectPeriod
-        });
-        //Обходим функции сервиса
-        service.functions.forEach(fn => {
-            client.subscribe(fn.sFnURL);
-        });
-        //Прослушиваем сообщения
-        client.on("message", (topic, message) => {
-            //Обрабатываем сообщение
-            processMQTTMessage({
-                message,
-                service,
-                fn: service.functions.find(fn => {
-                    return fn.sFnURL === topic;
-                })
-            });
-        });
-        //Прослушиваем отключение от сервера
-        client.on("offline", () => {
-            //Выводим ошибку
-            logger.error(`Соединение с MQTT потеряно (${sBroker})`);
-        });
-        //Прослушиваем восстановление соединения
-        client.on("reconnect", () => {
-            //Сообщим о восстановлении соединения
-            logger.info(`Соединение с MQTT восстановлено (${sBroker})`);
-        });
-        //Возвращаем подключение
-        return client;
-    } catch (e) {
-        logger.error(`Ошибка запуска обработчика очереди исходящих сообщений MQTT: ${makeErrorText(e)}`);
-    }
-};
-
-//-----------------
-// Интерфейс модуля
-//-----------------
-
-exports.publishMQTT = publishMQTT;
-exports.subscribeMQTT = subscribeMQTT;

core/out_queue.js

@@ -70,10 +70,6 @@ class OutQueue extends EventEmitter {
             this.inProgress = [];
             //Привяжем методы к указателю на себя для использования в обработчиках событий
             this.outDetectingLoop = this.outDetectingLoop.bind(this);
-            //Параметры подключения к Kafka
-            this.kafka = prms.kafka;
-            //Параметры подключения к MQTT
-            this.mqtt = prms.mqtt;
         } else {
             throw new ServerError(SERR_OBJECT_BAD_INTERFACE, sCheckResult);
         }
@@ -145,7 +141,11 @@ class OutQueue extends EventEmitter {
     //Старт обработчика
     startQueueProcessor(prms) {
         //Проверяем структуру переданного объекта для старта обработчика
-        let sCheckResult = validateObject(prms, prmsOutQueueSchema.startQueueProcessor, "Параметры функции запуска обработчика сообщения очереди");
+        let sCheckResult = validateObject(
+            prms,
+            prmsOutQueueSchema.startQueueProcessor,
+            "Параметры функции запуска обработчика сообщения очереди"
+        );
         //Если структура объекта в норме
         if (!sCheckResult) {
             //Добавляем идентификатор позиции очереди в список обрабатываемых
@@ -163,9 +163,7 @@ class OutQueue extends EventEmitter {
                 function: _.find(_.find(this.services, { nId: prms.queue.nServiceId }).functions, {
                     nId: prms.queue.nServiceFnId
                 }),
-                sProxy: this.sProxy,
+                sProxy: this.sProxy
-                kafka: this.kafka,
-                mqtt: this.mqtt
             });
             //Уменьшаем количество доступных обработчиков
             this.nWorkersLeft--;
@@ -176,7 +174,11 @@ class OutQueue extends EventEmitter {
     //Останов обработчика
     stopQueueProcessor(prms) {
         //Проверяем структуру переданного объекта для останова обработчика
-        let sCheckResult = validateObject(prms, prmsOutQueueSchema.stopQueueProcessor, "Параметры функции останова обработчика сообщения очереди");
+        let sCheckResult = validateObject(
+            prms,
+            prmsOutQueueSchema.stopQueueProcessor,
+            "Параметры функции останова обработчика сообщения очереди"
+        );
         //Если структура объекта в норме
         if (!sCheckResult) {
             //Удаляем идентификатор позиции очереди из списка обрабатываемых
@@ -217,13 +219,19 @@ class OutQueue extends EventEmitter {
                 throw new ServerError(SERR_OBJECT_BAD_INTERFACE, sCheckResult);
             }
         } catch (e) {
-            await this.logger.error(`При отправке уведомления об ошибке обработки исходящего сообщения: ${makeErrorText(e)}`);
+            await this.logger.error(
+                `При отправке уведомления об ошибке обработки исходящего сообщения: ${makeErrorText(e)}`
+            );
         }
     }
     //Запуск обработки очередного сообщения
     processMessage(prms) {
         //Проверяем структуру переданного объекта
-        let sCheckResult = validateObject(prms, prmsOutQueueSchema.processMessage, "Параметры функции запуска обработки очередного сообщения");
+        let sCheckResult = validateObject(
+            prms,
+            prmsOutQueueSchema.processMessage,
+            "Параметры функции запуска обработки очередного сообщения"
+        );
         //Если структура объекта в норме
         if (!sCheckResult) {
             //Проверим, что есть доступные обработчики
@@ -268,9 +276,10 @@ class OutQueue extends EventEmitter {
                                     });
                                 } catch (e) {
                                     //Отразим в протоколе ошибку постановки задачи на аутентификацию сервиса
-                                    await self.logger.error(`Ошибка постановки задачи на аутентификацию сервиса: ${makeErrorText(e)}`, {
+                                    await self.logger.error(
-                                        nQueueId: prms.queue.nId
+                                        `Ошибка постановки задачи на аутентификацию сервиса: ${makeErrorText(e)}`,
-                                    });
+                                        { nQueueId: prms.queue.nId }
+                                    );
                                 }
                             }
                         }
@@ -289,21 +298,25 @@ class OutQueue extends EventEmitter {
                             sExecMsg: sError,
                             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
+                                (nQueueOldExecCnt == prms.queue.nExecCnt
+                                    ? prms.queue.nExecCnt + 1
+                                    : prms.queue.nExecCnt) < prms.queue.nRetryAttempts
                                     ? prms.queue.nExecState
                                     : objQueueSchema.NQUEUE_EXEC_STATE_ERR
                         });
                     }
                     //Если исполнение завершилось полностью и с ошибкой - расскажем об этом
-                    if (prms.queue.nExecState == objQueueSchema.NQUEUE_EXEC_STATE_ERR) await this.notifyMessageProcessError(prms);
+                    if (prms.queue.nExecState == objQueueSchema.NQUEUE_EXEC_STATE_ERR)
+                        await this.notifyMessageProcessError(prms);
                     //Останавливаем обработчик и инкрементируем флаг их доступного количества
                     try {
                         this.stopQueueProcessor({ nQueueId: prms.queue.nId, proc });
                     } catch (e) {
                         //Отразим в протоколе ошибку останова
-                        await self.logger.error(`Ошибка останова обработчика исходящего сообщения: ${makeErrorText(e)}`, {
+                        await self.logger.error(
-                            nQueueId: prms.queue.nId
+                            `Ошибка останова обработчика исходящего сообщения: ${makeErrorText(e)}`,
-                        });
+                            { nQueueId: prms.queue.nId }
+                        );
                     }
                 });
                 //Перехват ошибок обработчика
@@ -320,20 +333,23 @@ class OutQueue extends EventEmitter {
                         sExecMsg: makeErrorText(e),
                         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
+                            (nQueueOldExecCnt == prms.queue.nExecCnt ? prms.queue.nExecCnt + 1 : prms.queue.nExecCnt) <
+                            prms.queue.nRetryAttempts
                                 ? prms.queue.nExecState
                                 : objQueueSchema.NQUEUE_EXEC_STATE_ERR
                     });
                     //Если исполнение завершилось полностью и с ошибкой - расскажем об этом
-                    if (prms.queue.nExecState == objQueueSchema.NQUEUE_EXEC_STATE_ERR) await this.notifyMessageProcessError(prms);
+                    if (prms.queue.nExecState == objQueueSchema.NQUEUE_EXEC_STATE_ERR)
+                        await this.notifyMessageProcessError(prms);
                     //Останавливаем обработчик и инкрементируем флаг их доступного количества
                     try {
                         this.stopQueueProcessor({ nQueueId: prms.queue.nId, proc });
                     } catch (e) {
                         //Отразим в протоколе ошибку останова
-                        await self.logger.error(`Ошибка останова обработчика исходящего сообщения: ${makeErrorText(e)}`, {
+                        await self.logger.error(
-                            nQueueId: prms.queue.nId
+                            `Ошибка останова обработчика исходящего сообщения: ${makeErrorText(e)}`,
-                        });
+                            { nQueueId: prms.queue.nId }
+                        );
                     }
                 });
                 //Перехват останова обработчика
@@ -376,12 +392,16 @@ class OutQueue extends EventEmitter {
                                     nQueueId: outMsg.nId,
                                     sExecMsg: makeErrorText(e),
                                     nIncExecCnt: NINC_EXEC_CNT_YES,
-                                    nExecState: outMsg.nExecCnt + 1 < outMsg.nRetryAttempts ? outMsg.nExecState : objQueueSchema.NQUEUE_EXEC_STATE_ERR
+                                    nExecState:
+                                        outMsg.nExecCnt + 1 < outMsg.nRetryAttempts
+                                            ? outMsg.nExecState
+                                            : objQueueSchema.NQUEUE_EXEC_STATE_ERR
                                 });
                                 //Фиксируем ошибку обработки сервером приложений - запись в протокол работы сервера приложений
                                 await this.logger.error(makeErrorText(e), { nQueueId: outMsg.nId });
                                 //Если исполнение завершилось полностью и с ошибкой - расскажем об этом
-                                if (queue.nExecState == objQueueSchema.NQUEUE_EXEC_STATE_ERR) await this.notifyMessageProcessError({ queue });
+                                if (queue.nExecState == objQueueSchema.NQUEUE_EXEC_STATE_ERR)
+                                    await this.notifyMessageProcessError({ queue });
                             }
                         }
                     }

core/out_queue_processor.js

@@ -12,20 +12,7 @@ const _ = require("lodash"); //Работа с массивами и объек
 const rqp = require("request-promise"); //Работа с HTTP/HTTPS запросами
 const lg = require("./logger"); //Протоколирование работы
 const db = require("./db_connector"); //Взаимодействие с БД
-const {
+const { makeErrorText, validateObject, getAppSrvFunction, buildURL, parseOptionsXML, buildOptionsXML, deepMerge } = require("./utils"); //Вспомогательные функции
-    makeErrorText,
-    validateObject,
-    getAppSrvFunction,
-    buildURL,
-    parseOptionsXML,
-    buildOptionsXML,
-    deepMerge,
-    getKafkaConnectionSettings,
-    getMQTTConnectionSettings,
-    getKafkaBroker,
-    getKafkaAuth,
-    getURLProtocol
-} = require("./utils"); //Вспомогательные функции
 const { ServerError } = require("./server_errors"); //Типовая ошибка
 const objOutQueueProcessorSchema = require("../models/obj_out_queue_processor"); //Схема валидации сообщений обмена с бработчиком очереди исходящих сообщений
 const prmsOutQueueProcessorSchema = require("../models/prms_out_queue_processor"); //Схема валидации параметров функций модуля
@@ -33,7 +20,6 @@ const objQueueSchema = require("../models/obj_queue"); //Схемы валида
 const objServiceSchema = require("../models/obj_service"); //Схемы валидации сервиса
 const objServiceFnSchema = require("../models/obj_service_function"); //Схемы валидации функции сервиса
 const {
-    SERR_UNEXPECTED,
     SERR_OBJECT_BAD_INTERFACE,
     SERR_APP_SERVER_BEFORE,
     SERR_APP_SERVER_AFTER,
@@ -42,8 +28,6 @@ const {
     SERR_UNAUTH
 } = require("./constants"); //Глобальные константы
 const { NINC_EXEC_CNT_YES, NINC_EXEC_CNT_NO, NIS_ORIGINAL_NO, NIS_ORIGINAL_YES } = require("../models/prms_db_connector"); //Схемы валидации параметров функций модуля взаимодействия с БД
-const { publishMQTT } = require("./mqtt_connector"); //Работа с MQTT/MQTTS запросами
-const { publishKafka } = require("./kafka_connector"); //Работа с Kafka запросами
 
 //--------------------------
 // Глобальные идентификаторы
@@ -145,71 +129,21 @@ const appProcess = async prms => {
                 let optionsResp = {};
                 //Флаг прекращения обработки сообщения
                 let bStopPropagation = false;
-                //Флаг выполнения обработчика "До"
+                //Определимся с URL и телом сообщения в зависимости от способа передачи параметров (для POST, PATCH и PUT - данные в теле, для остальных - в URI)
-                let bExecuteBefore = true;
+                if (
-                //Флаг выполнения обработчика "После"
+                    [objServiceFnSchema.NFN_PRMS_TYPE_POST, objServiceFnSchema.NFN_PRMS_TYPE_PATCH, objServiceFnSchema.NFN_PRMS_TYPE_PUT].includes(
-                let bExecuteAfter = true;
+                        prms.function.nFnPrmsType
-                //Считываем протокол работы
+                    )
-                let sProtocol = getURLProtocol(prms.service.sSrvRoot);
+                ) {
-                //Исходя из протокола собираем параметры
+                    options.url = buildURL({ sSrvRoot: prms.service.sSrvRoot, sFnURL: prms.function.sFnURL });
-                switch (true) {
+                    options.body = prms.queue.blMsg;
-                    //Kafka
+                    options.headers = { "content-type": "application/octet-stream" };
-                    case sProtocol === objServiceSchema.SPROTOCOL_KAFKA:
+                } else {
-                        options.url = getKafkaBroker(prms.service.sSrvRoot);
+                    options.url = buildURL({
-                        options.body = prms.queue.blMsg;
+                        sSrvRoot: prms.service.sSrvRoot,
-                        options.topic = prms.function.sFnURL;
+                        sFnURL: prms.function.sFnURL,
-                        options.auth = getKafkaAuth(prms.service.sSrvUser, prms.service.sSrvPass);
+                        sQuery: prms.queue.blMsg === null ? "" : prms.queue.blMsg.toString()
-                        options.settings = getKafkaConnectionSettings(prms.service.sCode, prms.kafka);
+                    });
-                        //Если параметры подключения не считаны
-                        if (!options.settings) {
-                            //Расскажем об ошибке считывания
-                            throw new ServerError(
-                                SERR_UNEXPECTED,
-                                `Ошибка получения настроек подключения к Kafka для сервиса "${prms.service.sCode}". Необходимо проверить соответствующий параметр ("kafka") файла конфигурации сервиса приложений ("config.js").`
-                            );
-                        }
-                        //Указываем, что выполнение обработчика "После" невозможно
-                        bExecuteAfter = false;
-                        break;
-                    //mqtt и mqtts
-                    case [objServiceSchema.SPROTOCOL_MQTT, objServiceSchema.SPROTOCOL_MQTTS].includes(sProtocol):
-                        options.url = prms.service.sSrvRoot;
-                        options.body = prms.queue.blMsg;
-                        options.topic = prms.function.sFnURL;
-                        options.auth = { user: prms.service.sSrvUser, pass: prms.service.sSrvPass };
-                        options.settings = getMQTTConnectionSettings(prms.service.sCode, prms.mqtt);
-                        //Если параметры подключения не считаны
-                        if (!options.settings) {
-                            //Расскажем об ошибке считывания
-                            throw new ServerError(
-                                SERR_UNEXPECTED,
-                                `Ошибка получения настроек подключения к MQTT для сервиса "${prms.service.sCode}". Необходимо проверить соответствующий параметр ("mqtt") файла конфигурации сервиса приложений ("config.js").`
-                            );
-                        }
-                        //Указываем, что выполнение обработчика "После" невозможно
-                        bExecuteAfter = false;
-                        break;
-                    //Другие
-                    default:
-                        //Определимся с URL и телом сообщения в зависимости от способа передачи параметров (для POST, PATCH и PUT - данные в теле, для остальных - в URI)
-                        if (
-                            [
-                                objServiceFnSchema.NFN_PRMS_TYPE_POST,
-                                objServiceFnSchema.NFN_PRMS_TYPE_PATCH,
-                                objServiceFnSchema.NFN_PRMS_TYPE_PUT
-                            ].includes(prms.function.nFnPrmsType)
-                        ) {
-                            options.url = buildURL({ sSrvRoot: prms.service.sSrvRoot, sFnURL: prms.function.sFnURL });
-                            options.body = prms.queue.blMsg;
-                            options.headers = { "content-type": "application/octet-stream" };
-                        } else {
-                            options.url = buildURL({
-                                sSrvRoot: prms.service.sSrvRoot,
-                                sFnURL: prms.function.sFnURL,
-                                sQuery: prms.queue.blMsg === null ? "" : prms.queue.blMsg.toString()
-                            });
-                        }
-                        break;
                 }
                 // Если у сервиса указан прокси, либо у приложения установлен глобальный прокси
                 if (prms.service.sProxyURL || prms.sProxy) {
@@ -233,7 +167,7 @@ const appProcess = async prms => {
                     }
                 }
                 //Выполняем обработчик "До" (если он есть)
-                if (prms.function.sAppSrvBefore && bExecuteBefore) {
+                if (prms.function.sAppSrvBefore) {
                     const fnBefore = getAppSrvFunction(prms.function.sAppSrvBefore);
                     let resBefore = null;
                     try {
@@ -265,10 +199,7 @@ const appProcess = async prms => {
                                         objServiceFnSchema.NFN_PRMS_TYPE_POST,
                                         objServiceFnSchema.NFN_PRMS_TYPE_PATCH,
                                         objServiceFnSchema.NFN_PRMS_TYPE_PUT
-                                    ].includes(prms.function.nFnPrmsType) ||
+                                    ].includes(prms.function.nFnPrmsType)
-                                    [objServiceSchema.SPROTOCOL_KAFKA, objServiceSchema.SPROTOCOL_MQTT, objServiceSchema.SPROTOCOL_MQTTS].includes(
-                                        sProtocol
-                                    )
                                 ) {
                                     options.body = prms.queue.blMsg;
                                 } else {
@@ -330,39 +261,9 @@ const appProcess = async prms => {
                                 nQueueId: prms.queue.nId
                             });
                         }
-                        //Инициализируем ответ от сервера
+                        //Ждем ответ от удалённого сервера
-                        let serverResp = null;
+                        options.resolveWithFullResponse = true;
-                        //Выполняем отправку исходя из протокола
+                        let serverResp = await rqp(options);
-                        switch (true) {
-                            //Kafka
-                            case sProtocol === objServiceSchema.SPROTOCOL_KAFKA:
-                                serverResp = await publishKafka({
-                                    settings: options.settings,
-                                    url: options.url,
-                                    auth: options.auth,
-                                    topic: options.topic,
-                                    message: options.body
-                                });
-                                console.log(serverResp);
-                                break;
-                            //mqtt и mqtts
-                            case [objServiceSchema.SPROTOCOL_MQTT, objServiceSchema.SPROTOCOL_MQTTS].includes(sProtocol):
-                                serverResp = await publishMQTT({
-                                    settings: options.settings,
-                                    url: options.url,
-                                    auth: options.auth,
-                                    topic: options.topic,
-                                    message: options.body
-                                });
-                                break;
-                            //Другие
-                            default:
-                                //Ждем ответ от удалённого сервера
-                                options.resolveWithFullResponse = true;
-                                //Отправляем запрос
-                                serverResp = await rqp(options);
-                                break;
-                        }
                         //Сохраняем полученный ответ
                         prms.queue.blResp = Buffer.from(serverResp.body || "");
                         await dbConn.setQueueResp({
@@ -392,7 +293,7 @@ const appProcess = async prms => {
                         throw new ServerError(SERR_WEB_SERVER, sError);
                     }
                     //Выполняем обработчик "После" (если он есть)
-                    if (prms.function.sAppSrvAfter && bExecuteAfter) {
+                    if (prms.function.sAppSrvAfter) {
                         const fnAfter = getAppSrvFunction(prms.function.sAppSrvAfter);
                         let resAfter = null;
                         try {
@@ -614,9 +515,7 @@ const processTask = async prms => {
                             queue: q,
                             service: prms.task.service,
                             function: prms.task.function,
-                            sProxy: prms.task.sProxy,
+                            sProxy: prms.task.sProxy
-                            kafka: prms.task.kafka,
-                            mqtt: prms.task.mqtt
                         });
                         //Если результат обработки ошибка - пробрасываем её дальше
                         if (res instanceof ServerError) {

core/utils.js

@@ -22,7 +22,6 @@ const {
 } = require("./constants"); //Глобавльные константы системы
 const { ServerError } = require("./server_errors"); //Ошибка сервера
 const prmsUtilsSchema = require("../models/prms_utils"); //Схемы валидации параметров функций
-const { SPROTOCOL_HTTP, SPROTOCOL_KAFKA } = require("../models/obj_service"); //Схемы валидации сервиса
 
 //------------
 // Тело модуля
@@ -46,7 +45,9 @@ const validateObject = (obj, schema, sObjName) => {
                     let a = errors.map(e => {
                         return e.message;
                     });
-                    sRes = `Объект${sObjName ? ` "${sObjName}" ` : " "}имеет некорректный формат: ${_.uniq(a).join("; ")}`;
+                    sRes = `Объект${sObjName ? ` "${sObjName}" ` : " "}имеет некорректный формат: ${_.uniq(a).join(
+                        "; "
+                    )}`;
                 }
             } else {
                 //Валидатор вернул не то, что мы ожидали
@@ -141,7 +142,8 @@ const getAppSrvFunction = sAppSrv => {
     //Объявим формат (для сообщений об ошибках)
     const sFormat = "(ожидаемый формат: <МОДУЛЬ>/<ФУНКЦИЯ>)";
     //Проверим, что есть что разбирать
-    if (!sAppSrv) throw new ServerError(SERR_MODULES_NO_MODULE_SPECIFIED, `Не указаны модуль и функция обработчика ${sFormat}`);
+    if (!sAppSrv)
+        throw new ServerError(SERR_MODULES_NO_MODULE_SPECIFIED, `Не указаны модуль и функция обработчика ${sFormat}`);
     //Разбираем
     try {
         //Разбираем на модуль и функцию
@@ -173,7 +175,11 @@ const getAppSrvFunction = sAppSrv => {
 const sendMail = prms => {
     return new Promise((resolve, reject) => {
         //Проверяем структуру переданного объекта для отправки E-Mail уведомления
-        let sCheckResult = validateObject(prms, prmsUtilsSchema.sendMail, "Параметры функции отправки E-Mail уведомления");
+        let sCheckResult = validateObject(
+            prms,
+            prmsUtilsSchema.sendMail,
+            "Параметры функции отправки E-Mail уведомления"
+        );
         //Если структура объекта в норме
         if (!sCheckResult) {
             //Формируем параметры для подключения к SMTP
@@ -190,7 +196,7 @@ const sendMail = prms => {
                 transpOptions.auth = {
                     user: prms.mail.sUser,
                     pass: prms.mail.sPass
-                };
+                }
             }
             //Настраиваем подключение к SMTP-серверу
             let transporter = nodemailer.createTransport(transpOptions);
@@ -209,7 +215,12 @@ const sendMail = prms => {
                     reject(new ServerError(SERR_MAIL_FAILED, `${error.code}: ${error}`));
                 } else {
                     if (info.rejected && Array.isArray(info.rejected) && info.rejected.length > 0) {
-                        reject(new ServerError(SERR_MAIL_FAILED, `Сообщение не доствлено адресатам: ${info.rejected.join(", ")}`));
+                        reject(
+                            new ServerError(
+                                SERR_MAIL_FAILED,
+                                `Сообщение не доствлено адресатам: ${info.rejected.join(", ")}`
+                            )
+                        );
                     } else {
                         resolve(info);
                     }
@@ -228,7 +239,7 @@ const buildURL = prms => {
     //Если структура объекта в норме
     if (!sCheckResult) {
         //Формируем URL с учетом лишних "/"
-        return `${prms.sSrvRoot.replace(/\/+$/, "")}/${prms.sFnURL.replace(/^\/+/, "")}${prms.sQuery ? `?${prms.sQuery}` : ""}`;
+        return `${prms.sSrvRoot.replace(/\/+$/, '')}/${prms.sFnURL.replace(/^\/+/, '')}${prms.sQuery ? `?${prms.sQuery}` : ""}`;
     } else {
         throw new ServerError(SERR_OBJECT_BAD_INTERFACE, sCheckResult);
     }
@@ -272,7 +283,11 @@ const parseXML = prms => {
 //Разбор параметров сообщения/ответа (XML > JSON)
 const parseOptionsXML = async prms => {
     //Проверяем структуру переданных параметров
-    let sCheckResult = validateObject(prms, prmsUtilsSchema.parseOptionsXML, "Параметры функции разбора XML параметров сообщения/ответа");
+    let sCheckResult = validateObject(
+        prms,
+        prmsUtilsSchema.parseOptionsXML,
+        "Параметры функции разбора XML параметров сообщения/ответа"
+    );
     //Если структура объекта в норме
     if (!sCheckResult) {
         try {
@@ -296,7 +311,11 @@ const parseOptionsXML = async prms => {
 //Сборка параметров сообщения/ответа (JSON > XML)
 const buildOptionsXML = prms => {
     //Проверяем структуру переданных параметров
-    let sCheckResult = validateObject(prms, prmsUtilsSchema.buildOptionsXML, "Параметры функции сборки XML параметров сообщения/ответа");
+    let sCheckResult = validateObject(
+        prms,
+        prmsUtilsSchema.buildOptionsXML,
+        "Параметры функции сборки XML параметров сообщения/ответа"
+    );
     //Если структура объекта в норме
     if (!sCheckResult) {
         try {
@@ -327,62 +346,6 @@ const deepMerge = (...args) => {
     return res;
 };
 
-//Считывание параметров подключения для сервиса обмена (при service === "" считывание подключения "По умолчанию", settingsArray - массив объектов [{sService: "", ...},...])
-const getConnectionSettings = (service, settingsArray) => {
-    //Считываем параметры и возвращаем
-    return settingsArray.find(connection => {
-        return connection.sService === service;
-    });
-};
-
-//Считывание параметров подключения к Kafka для сервиса обмена (kafka - массив объектов [{sService: "", ...},...])
-const getKafkaConnectionSettings = (service, kafka) => {
-    //Считываем подключение с указанным сервисом обмена
-    let kafkaConnection = getConnectionSettings(service, kafka);
-    //Если нет подключения с указанным сервисом обмена
-    if (!kafkaConnection) {
-        //Считываем "По умолчанию"
-        kafkaConnection = getConnectionSettings("", kafka);
-    }
-    //Вернем результат
-    return kafkaConnection;
-};
-
-//Считывание параметров подключения к MQTT для сервиса обмена (mqtt - массив объектов [{sService: "", ...},...])
-const getMQTTConnectionSettings = (service, mqtt) => {
-    //Считываем подключение с указанным сервисом обмена
-    let mqttConnection = getConnectionSettings(service, mqtt);
-    //Если нет подключения с указанным сервисом обмена
-    if (!mqttConnection) {
-        //Считываем "По умолчанию"
-        mqttConnection = getConnectionSettings("", mqtt);
-    }
-    //Вернем результат
-    return mqttConnection;
-};
-
-//Получение брокера Kafka по адресу сервиса обмена (прим. kafka://server.ru -> server.ru, https://server.ru => undefined)
-const getKafkaBroker = sURL => {
-    //Если протокол URL - Kafka
-    if (getURLProtocol(sURL) === SPROTOCOL_KAFKA) {
-        //Возвращаем брокера
-        return sURL.slice(8);
-    }
-    //Возвращаем undefined
-    return;
-};
-
-//Получение авторизации для Kafka
-const getKafkaAuth = (sUser, sPass) => {
-    return sUser ? { ssl: true, sasl: { mechanism: "plain", username: sUser, password: sPass } } : null;
-};
-
-//Получение протокола адреса (прим. mqtt://server.ru -> mqtt, https://server.ru => https, ...)
-const getURLProtocol = sURL => {
-    //Если начинается с "/" - HTTP, иначе получаем из URL
-    return sURL.substring(0, 1) === "/" ? SPROTOCOL_HTTP : new URL(sURL).protocol.slice(0, -1);
-};
-
 //-----------------
 // Интерфейс модуля
 //-----------------
@@ -401,8 +364,3 @@ exports.parseOptionsXML = parseOptionsXML;
 exports.buildOptionsXML = buildOptionsXML;
 exports.getNowString = getNowString;
 exports.deepMerge = deepMerge;
-exports.getKafkaConnectionSettings = getKafkaConnectionSettings;
-exports.getMQTTConnectionSettings = getMQTTConnectionSettings;
-exports.getKafkaBroker = getKafkaBroker;
-exports.getKafkaAuth = getKafkaAuth;
-exports.getURLProtocol = getURLProtocol;

models/obj_service.js

@@ -33,13 +33,6 @@ const NIS_AUTH_NO = 0; //Неаутентифицирован
 const SIS_AUTH_YES = "IS_AUTH_YES"; //Аутентифицирован (строковый код)
 const SIS_AUTH_NO = "IS_AUTH_NO"; //Неаутентифицирован (строковый код)
 
-//Протоколы работы сервиса
-const SPROTOCOL_HTTP = "http"; //Протокол HTTP
-const SPROTOCOL_HTTPS = "https"; //Протокол HTTPS
-const SPROTOCOL_MQTT = "mqtt"; //Протокол MQTT
-const SPROTOCOL_MQTTS = "mqtts"; //Протокол MQTTS
-const SPROTOCOL_KAFKA = "kafka"; //Протокол для работы с KAFKA
-
 //-------------
 //  Тело модуля
 //-------------
@@ -66,11 +59,6 @@ exports.NIS_AUTH_YES = NIS_AUTH_YES;
 exports.NIS_AUTH_NO = NIS_AUTH_NO;
 exports.SIS_AUTH_YES = SIS_AUTH_YES;
 exports.SIS_AUTH_NO = SIS_AUTH_NO;
-exports.SPROTOCOL_HTTP = SPROTOCOL_HTTP;
-exports.SPROTOCOL_HTTPS = SPROTOCOL_HTTPS;
-exports.SPROTOCOL_MQTT = SPROTOCOL_MQTT;
-exports.SPROTOCOL_MQTTS = SPROTOCOL_MQTTS;
-exports.SPROTOCOL_KAFKA = SPROTOCOL_KAFKA;
 
 //Схема валидации сервиса
 exports.Service = new Schema({
@@ -156,8 +144,10 @@ exports.Service = new Schema({
         enum: [NUNAVLBL_NTF_SIGN_NO, NUNAVLBL_NTF_SIGN_YES],
         required: true,
         message: {
-            type: path => `Признак необходимости оповещения о простое внешнего сервиса (${path}) имеет некорректный тип данных (ожидалось - Number)`,
+            type: path =>
-            enum: path => `Значение признака необходимости оповещения о простое внешнего сервиса (${path}) не поддерживается`,
+                `Признак необходимости оповещения о простое внешнего сервиса (${path}) имеет некорректный тип данных (ожидалось - Number)`,
+            enum: path =>
+                `Значение признака необходимости оповещения о простое внешнего сервиса (${path}) не поддерживается`,
             required: path => `Не указан признак необходимости оповещения о простое внешнего сервиса (${path})`
         }
     },
@@ -169,8 +159,10 @@ exports.Service = new Schema({
         message: {
             type: path =>
                 `Строковый код признака необходимости оповещения о простое внешнего сервиса (${path}) имеет некорректный тип данных (ожидалось - String)`,
-            enum: path => `Значение строкового кода признака необходимости оповещения о простое внешнего сервиса (${path}) не поддерживается`,
+            enum: path =>
-            required: path => `Не указан строковый код признака необходимости оповещения о простое внешнего сервиса (${path})`
+                `Значение строкового кода признака необходимости оповещения о простое внешнего сервиса (${path}) не поддерживается`,
+            required: path =>
+                `Не указан строковый код признака необходимости оповещения о простое внешнего сервиса (${path})`
         }
     },
     //Максимальное время простоя (мин) удалённого сервиса для генерации оповещения
@@ -180,7 +172,8 @@ exports.Service = new Schema({
         message: {
             type: path =>
                 `Максимальное время простоя (мин) удалённого сервиса для генерации оповещения (${path}) имеет некорректный тип данных (ожидалось - Number)`,
-            required: path => `Не указано максимальное время простоя (мин) удалённого сервиса для генерации оповещения (${path})`
+            required: path =>
+                `Не указано максимальное время простоя (мин) удалённого сервиса для генерации оповещения (${path})`
         }
     },
     //Список адресов E-Mail для оповещения о простое внешнего сервиса
@@ -201,7 +194,8 @@ exports.Service = new Schema({
         type: String,
         required: false,
         message: {
-            type: path => `Адрес прокси-сервера в очереди обмена (${path}) имеет некорректный тип данных (ожидалось - String)`,
+            type: path =>
+                `Адрес прокси-сервера в очереди обмена (${path}) имеет некорректный тип данных (ожидалось - String)`,
             required: path => `Не указан адрес прокси-сервера в очереди обмена (${path})`
         }
     },
@@ -243,7 +237,8 @@ exports.ServiceCtx = new Schema({
         type: String,
         required: false,
         message: {
-            type: path => `Строковое представление даты истечения контекста (${path}) имеет некорректный тип данных (ожидалось - String)`,
+            type: path =>
+                `Строковое представление даты истечения контекста (${path}) имеет некорректный тип данных (ожидалось - String)`,
             required: path => `Не указано строковое представление даты истечения контекста (${path})`
         }
     },
@@ -253,7 +248,8 @@ exports.ServiceCtx = new Schema({
         enum: [NIS_AUTH_YES, NIS_AUTH_NO],
         required: true,
         message: {
-            type: path => `Признака аутентицированности сервиса (${path}) имеет некорректный тип данных (ожидалось - Number)`,
+            type: path =>
+                `Признака аутентицированности сервиса (${path}) имеет некорректный тип данных (ожидалось - Number)`,
             enum: path => `Значение признака аутентицированности сервиса (${path}) не поддерживается`,
             required: path => `Не указан признак аутентицированности сервиса (${path})`
         }
@@ -264,7 +260,8 @@ exports.ServiceCtx = new Schema({
         enum: [SIS_AUTH_YES, SIS_AUTH_NO],
         required: true,
         message: {
-            type: path => `Строковый код признака аутентицированности сервиса (${path}) имеет некорректный тип данных (ожидалось - String)`,
+            type: path =>
+                `Строковый код признака аутентицированности сервиса (${path}) имеет некорректный тип данных (ожидалось - String)`,
             enum: path => `Значение строкового кода признака аутентицированности сервиса (${path}) не поддерживается`,
             required: path => `Не указан строковый код признака аутентицированности сервиса (${path})`
         }
@@ -287,7 +284,8 @@ exports.ServiceExpiredQueueInfo = new Schema({
         type: Number,
         required: true,
         message: {
-            type: path => `Количество просроченных сообщений обмена (${path}) имеет некорректный тип данных (ожидалось - Number)`,
+            type: path =>
+                `Количество просроченных сообщений обмена (${path}) имеет некорректный тип данных (ожидалось - Number)`,
             required: path => `Не указано количество просроченных сообщений обмена (${path})`
         }
     },
@@ -296,7 +294,8 @@ exports.ServiceExpiredQueueInfo = new Schema({
         type: String,
         required: true,
         message: {
-            type: path => `Информация о просроченных сообщениях обмена (${path}) имеет некорректный тип данных (ожидалось - String)`,
+            type: path =>
+                `Информация о просроченных сообщениях обмена (${path}) имеет некорректный тип данных (ожидалось - String)`,
             required: path => `Не указана информация о просроченных сообщениях обмена (${path})`
         }
     }

models/obj_service_function.js

@@ -156,7 +156,8 @@ exports.ServiceFunction = new Schema({
         type: Number,
         required: true,
         message: {
-            type: path => `Идентификатор родительского сервиса функции (${path}) имеет некорректный тип данных (ожидалось - Number)`,
+            type: path =>
+                `Идентификатор родительского сервиса функции (${path}) имеет некорректный тип данных (ожидалось - Number)`,
             required: path => `Не указан идентификатор родительского сервиса функции (${path})`
         }
     },
@@ -186,7 +187,8 @@ exports.ServiceFunction = new Schema({
         enum: [SFN_TYPE_DATA, SFN_TYPE_LOGIN, SFN_TYPE_LOGOUT],
         required: true,
         message: {
-            type: path => `Строковый код типа функции сервиса (${path}) имеет некорректный тип данных (ожидалось - String)`,
+            type: path =>
+                `Строковый код типа функции сервиса (${path}) имеет некорректный тип данных (ожидалось - String)`,
             enum: path => `Значение строкового кода типа функции сервиса (${path}) не поддерживается`,
             required: path => `Не указан строковый код типа функции сервиса (${path})`
         }
@@ -216,7 +218,8 @@ exports.ServiceFunction = new Schema({
         ],
         required: true,
         message: {
-            type: path => `Способ передачи параметров функции сервиса (${path}) имеет некорректный тип данных (ожидалось - Number)`,
+            type: path =>
+                `Способ передачи параметров функции сервиса (${path}) имеет некорректный тип данных (ожидалось - Number)`,
             enum: path => `Значение способа передачи параметров функции сервиса (${path}) не поддерживается`,
             required: path => `Не указан способ передачи параметров функции сервиса (${path})`
         }
@@ -237,8 +240,10 @@ exports.ServiceFunction = new Schema({
         ],
         required: true,
         message: {
-            type: path => `Строковый код способа передачи параметров функции сервиса (${path}) имеет некорректный тип данных (ожидалось - String)`,
+            type: path =>
-            enum: path => `Значение строкового кода способа передачи параметров функции сервиса (${path}) не поддерживается`,
+                `Строковый код способа передачи параметров функции сервиса (${path}) имеет некорректный тип данных (ожидалось - String)`,
+            enum: path =>
+                `Значение строкового кода способа передачи параметров функции сервиса (${path}) не поддерживается`,
             required: path => `Не указан строковый код способа передачи параметров функции сервиса (${path})`
         }
     },
@@ -256,7 +261,8 @@ exports.ServiceFunction = new Schema({
         ],
         required: true,
         message: {
-            type: path => `График повторной отправки запроса функции сервиса (${path}) имеет некорректный тип данных (ожидалось - Number)`,
+            type: path =>
+                `График повторной отправки запроса функции сервиса (${path}) имеет некорректный тип данных (ожидалось - Number)`,
             enum: path => `Значение графика повторной отправки запроса функции сервиса (${path}) не поддерживается`,
             required: path => `Не указан график повторной отправки запроса функции сервиса (${path})`
         }
@@ -277,7 +283,8 @@ exports.ServiceFunction = new Schema({
         message: {
             type: path =>
                 `Строковый код графика повторной отправки запроса функции сервиса (${path}) имеет некорректный тип данных (ожидалось - String)`,
-            enum: path => `Значение строкового кода графика повторной отправки запроса функции сервиса (${path}) не поддерживается`,
+            enum: path =>
+                `Значение строкового кода графика повторной отправки запроса функции сервиса (${path}) не поддерживается`,
             required: path => `Не указан строковый код графика повторной отправки запроса функции сервиса (${path})`
         }
     },
@@ -288,7 +295,8 @@ exports.ServiceFunction = new Schema({
         message: {
             type: path =>
                 `Идентификатор типового сообщения обмена, обрабатываемого функцией сервиса (${path}) имеет некорректный тип данных (ожидалось - Number)`,
-            required: path => `Не указан идентификатор типового сообщения обмена, обрабатываемого функцией сервиса (${path})`
+            required: path =>
+                `Не указан идентификатор типового сообщения обмена, обрабатываемого функцией сервиса (${path})`
         }
     },
     //Код типового сообщения обмена, обрабатываемого функцией сервиса
@@ -306,7 +314,8 @@ exports.ServiceFunction = new Schema({
         type: String,
         required: false,
         message: {
-            type: path => `Обработчик сообщения со стороны БД для функции сервиса (${path}) имеет некорректный тип данных (ожидалось - String)`,
+            type: path =>
+                `Обработчик сообщения со стороны БД для функции сервиса (${path}) имеет некорректный тип данных (ожидалось - String)`,
             required: path => `Не указан обработчик сообщения со стороны БД для функции сервиса (${path})`
         }
     },
@@ -318,7 +327,8 @@ exports.ServiceFunction = new Schema({
         message: {
             type: path =>
                 `Обработчик сообщения 'до' на строне сервера приложений для функции сервиса (${path}) имеет некорректный тип данных (ожидалось - String)`,
-            required: path => `Не указан обработчик сообщения 'до' на строне сервера приложений для функции сервиса (${path})`,
+            required: path =>
+                `Не указан обработчик сообщения 'до' на строне сервера приложений для функции сервиса (${path})`,
             validateAppSrvFn: path =>
                 `Обработчик сообщения 'до' на строне сервера приложений для функции сервиса (${path}) имеет некорректный формат, ожидалось: <МОДУЛЬ>.js/<ФУНКЦИЯ>`
         }
@@ -331,7 +341,8 @@ exports.ServiceFunction = new Schema({
         message: {
             type: path =>
                 `Обработчик сообщения 'после' на строне сервера приложений для функции сервиса (${path}) имеет некорректный тип данных (ожидалось - String)`,
-            required: path => `Не указан обработчик сообщения 'после' на строне сервера приложений для функции сервиса (${path})`,
+            required: path =>
+                `Не указан обработчик сообщения 'после' на строне сервера приложений для функции сервиса (${path})`,
             validateAppSrvFn: path =>
                 `Обработчик сообщения 'после' на строне сервера приложений для функции сервиса (${path}) имеет некорректный формат, ожидалось: <МОДУЛЬ>.js/<ФУНКЦИЯ>`
         }
@@ -344,8 +355,10 @@ exports.ServiceFunction = new Schema({
         message: {
             type: path =>
                 `Признак необходимости аутентификации для исполнения функции сервсиа обмена (${path}) имеет некорректный тип данных (ожидалось - Number)`,
-            enum: path => `Значение признака необходимости аутентификации для исполнения функции сервсиа обмена (${path}) не поддерживается`,
+            enum: path =>
-            required: path => `Не указан признак необходимости аутентификации для исполнения функции сервсиа обмена (${path})`
+                `Значение признака необходимости аутентификации для исполнения функции сервсиа обмена (${path}) не поддерживается`,
+            required: path =>
+                `Не указан признак необходимости аутентификации для исполнения функции сервсиа обмена (${path})`
         }
     },
     //Признак необходимости аутентификации для исполнения функции сервсиа обмена (строковый код)
@@ -358,7 +371,8 @@ exports.ServiceFunction = new Schema({
                 `Строковый код признака необходимости аутентификации для исполнения функции сервсиа обмена (${path}) имеет некорректный тип данных (ожидалось - String)`,
             enum: path =>
                 `Значение строкового кода признака необходимости аутентификации для исполнения функции сервсиа обмена (${path}) не поддерживается`,
-            required: path => `Не указан строковый код признака необходимости аутентификации для исполнения функции сервсиа обмена (${path})`
+            required: path =>
+                `Не указан строковый код признака необходимости аутентификации для исполнения функции сервсиа обмена (${path})`
         }
     },
     //Признак оповещения об ошибке исполнения сообщения очереди для функции обработки
@@ -369,8 +383,10 @@ exports.ServiceFunction = new Schema({
         message: {
             type: path =>
                 `Признак оповещения об ошибке исполнения сообщения очереди для функции обработки (${path}) имеет некорректный тип данных (ожидалось - Number)`,
-            enum: path => `Значение признака оповещения об ошибке исполнения сообщения очереди для функции обработки (${path}) не поддерживается`,
+            enum: path =>
-            required: path => `Не указан признак оповещения об ошибке исполнения сообщения очереди для функции обработки (${path})`
+                `Значение признака оповещения об ошибке исполнения сообщения очереди для функции обработки (${path}) не поддерживается`,
+            required: path =>
+                `Не указан признак оповещения об ошибке исполнения сообщения очереди для функции обработки (${path})`
         }
     },
     //Признак оповещения об ошибке исполнения сообщения очереди для функции обработки (строковый код)
@@ -383,7 +399,8 @@ exports.ServiceFunction = new Schema({
                 `Строковый код признака оповещения об ошибке исполнения сообщения очереди для функции обработки (${path}) имеет некорректный тип данных (ожидалось - String)`,
             enum: path =>
                 `Значение строкового кода признака оповещения об ошибке исполнения сообщения очереди для функции обработки (${path}) не поддерживается`,
-            required: path => `Не указан строковый код признака оповещения об ошибке исполнения сообщения очереди для функции обработки (${path})`
+            required: path =>
+                `Не указан строковый код признака оповещения об ошибке исполнения сообщения очереди для функции обработки (${path})`
         }
     },
     //Список адресов E-Mail для оповещения об ошибке исполнения сообщения очереди для функции обработки
@@ -394,7 +411,8 @@ exports.ServiceFunction = new Schema({
         message: {
             type: path =>
                 `Список адресов E-Mail для оповещения об ошибке исполнения сообщения очереди для функции обработки (${path}) имеет некорректный тип данных (ожидалось - String)`,
-            required: path => `Не указан список адресов E-Mail для оповещения об ошибке исполнения сообщения очереди для функции обработки (${path})`,
+            required: path =>
+                `Не указан список адресов E-Mail для оповещения об ошибке исполнения сообщения очереди для функции обработки (${path})`,
             validateErrNtfMail: path =>
                 `Неверный формат списка адресов E-Mail для оповещения об ошибке исполнения сообщения очереди для функции обработки (${path}), для указания нескольких адресов следует использовать запятую в качестве разделителя (без пробелов)`
         }

package.json

@@ -21,10 +21,8 @@
         "body-parser": "^1.19.0",
         "cors": "^2.8.5",
         "express": "^4.17.1",
-        "kafkajs": "^2.2.4",
         "lodash": "^4.17.19",
         "module-alias": "^2.2.2",
-        "mqtt": "^5.10.1",
         "nodemailer": "^6.4.11",
         "oracledb": "^4.2.0",
         "request": "^2.88.2",