config.js

@@ -71,6 +71,44 @@ 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
@@ -98,5 +136,7 @@ module.exports = {
     dbConnect,
     outGoing,
     inComing,
+    kafka,
+    mqtt,
     mail
 };

config_default.js

@@ -71,6 +71,44 @@ 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
@@ -98,5 +136,7 @@ module.exports = {
     dbConnect,
     outGoing,
     inComing,
+    kafka,
+    mqtt,
     mail
 };

core/app.js

@@ -96,7 +96,7 @@ class ParusAppServer {
         //Запускаем обслуживание очереди входящих
         await this.logger.info("Запуск обработчика очереди входящих сообщений...");
         try {
-            this.inQ.startProcessing({ services: this.services });
+            await this.inQ.startProcessing({ services: this.services });
         } catch (e) {
             await this.logger.error(`Ошибка запуска обработчика очереди входящих сообщений: ${makeErrorText(e)}`);
             await this.stop();
@@ -196,9 +196,7 @@ class ParusAppServer {
         //Если настройки верны - будем стартовать
         if (!sCheckResult) {
             //Протоколируем версию и релиз
-            await this.logger.info(
+            await this.logger.info(`Версия сервера приложений: ${prms.config.common.sVersion}, релиз: ${prms.config.common.sRelease}`);
-                `Версия сервера приложений: ${prms.config.common.sVersion}, релиз: ${prms.config.common.sRelease}`
-            );
             //Создаём подключение к БД
             this.dbConn = new db.DBConnector({
                 connectSettings: {
@@ -219,7 +217,9 @@ 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,7 +227,9 @@ 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,7 +13,18 @@ const express = require("express"); //WEB-сервер Express
 const cors = require("cors"); //Управление заголовками безопасности для WEB-сервера Express
 const bodyParser = require("body-parser"); //Модуль для Express (разбор тела входящего запроса)
 const { ServerError } = require("./server_errors"); //Типовая ошибка
-const { makeErrorText, validateObject, buildURL, getAppSrvFunction, buildOptionsXML, parseOptionsXML, deepMerge } = require("./utils"); //Вспомогательные функции
+const {
+    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"); //Схемы валидации сервиса
@@ -28,6 +39,8 @@ const {
     SERR_DB_SERVER,
     SERR_UNAUTH
 } = require("./constants"); //Общесистемные константы
+const { subscribeMQTT } = require("./mqtt_connector"); //Модуль для работы с MQTT
+const { subscribeKafka } = require("./kafka_connector"); //Модуль для работы с Kafka
 
 //--------------------------
 // Глобальные идентификаторы
@@ -71,6 +84,13 @@ 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);
         }
@@ -360,8 +380,326 @@ 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
+                });
+            }
+        }
+    }
+
     //Запуск обработки очереди входящих сообщений
-    startProcessing(prms) {
+    async startProcessing(prms) {
         //Проверяем структуру переданного объекта для старта
         let sCheckResult = validateObject(prms, prmsInQueueSchema.startProcessing, "Параметры функции запуска обработки очереди входящих сообщений");
         //Если структура объекта в норме
@@ -379,55 +717,136 @@ class InQueue extends EventEmitter {
             //Конфигурируем сервер - обработка тела сообщения
             this.webApp.use(bodyParser.raw({ limit: `${this.inComing.nMsgMaxSize}mb`, type: "*/*" }));
             //Конфигурируем сервер - обходим все сервисы, работающие на приём сообщений
-            _.forEach(_.filter(this.services, { nSrvType: objServiceSchema.NSRV_TYPE_RECIVE }), srvs => {
+            _.forEach(
-                //Для любых запросов к корневому адресу сервиса - ответ о том, что это за сервис, и что он работает
+                _.filter(this.services, srv => {
-                this.webApp.all(srvs.sSrvRoot, (req, res) => {
+                    return (
-                    res.status(200).send(
+                        srv.nSrvType === objServiceSchema.NSRV_TYPE_RECIVE &&
-                        `<html><body><center><br><h1>Сервер приложений ПП Парус 8<br>(${this.common.sVersion} релиз ${this.common.sRelease})</h1><h3>Сервис: ${srvs.sName}</h3></center></body></html>`
+                        [objServiceSchema.SPROTOCOL_HTTP, objServiceSchema.SPROTOCOL_HTTPS].includes(getURLProtocol(srv.sSrvRoot))
                     );
-                });
+                }),
-                //Для всех статических функций сервиса...
+                srvs => {
-                _.forEach(
+                    //Для любых запросов к корневому адресу сервиса - ответ о том, что это за сервис, и что он работает
-                    _.filter(srvs.functions, fn => fn.sFnURL.startsWith("@")),
+                    this.webApp.all(srvs.sSrvRoot, (req, res) => {
-                    fn => {
+                        res.status(200).send(
-                        this.webApp.use(
+                            `<html><body><center><br><h1>Сервер приложений ПП Парус 8<br>(${this.common.sVersion} релиз ${this.common.sRelease})</h1><h3>Сервис: ${srvs.sName}</h3></center></body></html>`
-                            buildURL({ sSrvRoot: srvs.sSrvRoot, sFnURL: fn.sFnURL.substr(1) }),
-                            express.static(`${this.inComing.sStaticDir}/${fn.sFnURL.substr(1)}`)
                         );
-                    }
+                    });
-                );
+                    //Для всех статических функций сервиса...
-                //Для всех функций сервиса (кроме статических)...
+                    _.forEach(
-                _.forEach(
+                        _.filter(srvs.functions, fn => fn.sFnURL.startsWith("@")),
-                    _.filter(srvs.functions, fn => !fn.sFnURL.startsWith("@")),
+                        fn => {
-                    fn => {
+                            this.webApp.use(
-                        //...собственный обработчик, в зависимости от указанного способа передачи параметров
+                                buildURL({ sSrvRoot: srvs.sSrvRoot, sFnURL: fn.sFnURL.substr(1) }),
-                        this.webApp[fn.sFnPrmsType.toLowerCase()](buildURL({ sSrvRoot: srvs.sSrvRoot, sFnURL: fn.sFnURL }), async (req, res) => {
+                                express.static(`${this.inComing.sStaticDir}/${fn.sFnURL.substr(1)}`)
-                            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(e), {
+                                await this.logger.error(makeErrorText(new ServerError(SERR_WEB_SERVER, err.message)), {
                                     nServiceId: srvs.nId,
                                     nServiceFnId: fn.nId
                                 });
                                 //Отправим ошибку клиенту
-                                res.status(500).send(makeErrorText(e));
+                                res.status(500).send(makeErrorText(new ServerError(SERR_WEB_SERVER, err.message)));
-                            }
-                        });
-                        //...и собственный обработчик ошибок
-                        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)));
+                    );
-                        });
+                }
+            );
+            //Инициализируем настройки подключения
+            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").`
+                            );
+                        }
                     }
+                }
+            }
+            //Считываем прием сообщений по 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(
@@ -453,10 +872,39 @@ 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

@@ -0,0 +1,119 @@
+/*
+  Сервис интеграции ПП Парус 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

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

core/out_queue_processor.js

@@ -12,7 +12,20 @@ const _ = require("lodash"); //Работа с массивами и объек
 const rqp = require("request-promise"); //Работа с HTTP/HTTPS запросами
 const lg = require("./logger"); //Протоколирование работы
 const db = require("./db_connector"); //Взаимодействие с БД
-const { makeErrorText, validateObject, getAppSrvFunction, buildURL, parseOptionsXML, buildOptionsXML, deepMerge } = require("./utils"); //Вспомогательные функции
+const {
+    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"); //Схема валидации параметров функций модуля
@@ -20,6 +33,7 @@ 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,
@@ -28,6 +42,8 @@ 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 запросами
 
 //--------------------------
 // Глобальные идентификаторы
@@ -129,21 +145,71 @@ const appProcess = async prms => {
                 let optionsResp = {};
                 //Флаг прекращения обработки сообщения
                 let bStopPropagation = false;
-                //Определимся с URL и телом сообщения в зависимости от способа передачи параметров (для POST, PATCH и PUT - данные в теле, для остальных - в URI)
+                //Флаг выполнения обработчика "До"
-                if (
+                let bExecuteBefore = true;
-                    [objServiceFnSchema.NFN_PRMS_TYPE_POST, objServiceFnSchema.NFN_PRMS_TYPE_PATCH, objServiceFnSchema.NFN_PRMS_TYPE_PUT].includes(
+                //Флаг выполнения обработчика "После"
-                        prms.function.nFnPrmsType
+                let bExecuteAfter = true;
-                    )
+                //Считываем протокол работы
-                ) {
+                let sProtocol = getURLProtocol(prms.service.sSrvRoot);
-                    options.url = buildURL({ sSrvRoot: prms.service.sSrvRoot, sFnURL: prms.function.sFnURL });
+                //Исходя из протокола собираем параметры
-                    options.body = prms.queue.blMsg;
+                switch (true) {
-                    options.headers = { "content-type": "application/octet-stream" };
+                    //Kafka
-                } else {
+                    case sProtocol === objServiceSchema.SPROTOCOL_KAFKA:
-                    options.url = buildURL({
+                        options.url = getKafkaBroker(prms.service.sSrvRoot);
-                        sSrvRoot: prms.service.sSrvRoot,
+                        options.body = prms.queue.blMsg;
-                        sFnURL: prms.function.sFnURL,
+                        options.topic = prms.function.sFnURL;
-                        sQuery: prms.queue.blMsg === null ? "" : prms.queue.blMsg.toString()
+                        options.auth = getKafkaAuth(prms.service.sSrvUser, prms.service.sSrvPass);
-                    });
+                        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) {
@@ -167,7 +233,7 @@ const appProcess = async prms => {
                     }
                 }
                 //Выполняем обработчик "До" (если он есть)
-                if (prms.function.sAppSrvBefore) {
+                if (prms.function.sAppSrvBefore && bExecuteBefore) {
                     const fnBefore = getAppSrvFunction(prms.function.sAppSrvBefore);
                     let resBefore = null;
                     try {
@@ -199,7 +265,10 @@ 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 {
@@ -261,9 +330,39 @@ const appProcess = async prms => {
                                 nQueueId: prms.queue.nId
                             });
                         }
-                        //Ждем ответ от удалённого сервера
+                        //Инициализируем ответ от сервера
-                        options.resolveWithFullResponse = true;
+                        let serverResp = null;
-                        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({
@@ -293,7 +392,7 @@ const appProcess = async prms => {
                         throw new ServerError(SERR_WEB_SERVER, sError);
                     }
                     //Выполняем обработчик "После" (если он есть)
-                    if (prms.function.sAppSrvAfter) {
+                    if (prms.function.sAppSrvAfter && bExecuteAfter) {
                         const fnAfter = getAppSrvFunction(prms.function.sAppSrvAfter);
                         let resAfter = null;
                         try {
@@ -515,7 +614,9 @@ 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,6 +22,7 @@ const {
 } = require("./constants"); //Глобавльные константы системы
 const { ServerError } = require("./server_errors"); //Ошибка сервера
 const prmsUtilsSchema = require("../models/prms_utils"); //Схемы валидации параметров функций
+const { SPROTOCOL_HTTP, SPROTOCOL_KAFKA } = require("../models/obj_service"); //Схемы валидации сервиса
 
 //------------
 // Тело модуля
@@ -45,9 +46,7 @@ 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 {
                 //Валидатор вернул не то, что мы ожидали
@@ -142,8 +141,7 @@ const getAppSrvFunction = sAppSrv => {
     //Объявим формат (для сообщений об ошибках)
     const sFormat = "(ожидаемый формат: <МОДУЛЬ>/<ФУНКЦИЯ>)";
     //Проверим, что есть что разбирать
-    if (!sAppSrv)
+    if (!sAppSrv) throw new ServerError(SERR_MODULES_NO_MODULE_SPECIFIED, `Не указаны модуль и функция обработчика ${sFormat}`);
-        throw new ServerError(SERR_MODULES_NO_MODULE_SPECIFIED, `Не указаны модуль и функция обработчика ${sFormat}`);
     //Разбираем
     try {
         //Разбираем на модуль и функцию
@@ -175,11 +173,7 @@ const getAppSrvFunction = sAppSrv => {
 const sendMail = prms => {
     return new Promise((resolve, reject) => {
         //Проверяем структуру переданного объекта для отправки E-Mail уведомления
-        let sCheckResult = validateObject(
+        let sCheckResult = validateObject(prms, prmsUtilsSchema.sendMail, "Параметры функции отправки E-Mail уведомления");
-            prms,
-            prmsUtilsSchema.sendMail,
-            "Параметры функции отправки E-Mail уведомления"
-        );
         //Если структура объекта в норме
         if (!sCheckResult) {
             //Формируем параметры для подключения к SMTP
@@ -196,7 +190,7 @@ const sendMail = prms => {
                 transpOptions.auth = {
                     user: prms.mail.sUser,
                     pass: prms.mail.sPass
-                }
+                };
             }
             //Настраиваем подключение к SMTP-серверу
             let transporter = nodemailer.createTransport(transpOptions);
@@ -215,12 +209,7 @@ 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(
+                        reject(new ServerError(SERR_MAIL_FAILED, `Сообщение не доствлено адресатам: ${info.rejected.join(", ")}`));
-                            new ServerError(
-                                SERR_MAIL_FAILED,
-                                `Сообщение не доствлено адресатам: ${info.rejected.join(", ")}`
-                            )
-                        );
                     } else {
                         resolve(info);
                     }
@@ -239,7 +228,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);
     }
@@ -283,11 +272,7 @@ const parseXML = prms => {
 //Разбор параметров сообщения/ответа (XML > JSON)
 const parseOptionsXML = async prms => {
     //Проверяем структуру переданных параметров
-    let sCheckResult = validateObject(
+    let sCheckResult = validateObject(prms, prmsUtilsSchema.parseOptionsXML, "Параметры функции разбора XML параметров сообщения/ответа");
-        prms,
-        prmsUtilsSchema.parseOptionsXML,
-        "Параметры функции разбора XML параметров сообщения/ответа"
-    );
     //Если структура объекта в норме
     if (!sCheckResult) {
         try {
@@ -311,11 +296,7 @@ const parseOptionsXML = async prms => {
 //Сборка параметров сообщения/ответа (JSON > XML)
 const buildOptionsXML = prms => {
     //Проверяем структуру переданных параметров
-    let sCheckResult = validateObject(
+    let sCheckResult = validateObject(prms, prmsUtilsSchema.buildOptionsXML, "Параметры функции сборки XML параметров сообщения/ответа");
-        prms,
-        prmsUtilsSchema.buildOptionsXML,
-        "Параметры функции сборки XML параметров сообщения/ответа"
-    );
     //Если структура объекта в норме
     if (!sCheckResult) {
         try {
@@ -346,6 +327,62 @@ 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);
+};
+
 //-----------------
 // Интерфейс модуля
 //-----------------
@@ -364,3 +401,8 @@ 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,6 +33,13 @@ 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
+
 //-------------
 //  Тело модуля
 //-------------
@@ -59,6 +66,11 @@ 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({
@@ -144,10 +156,8 @@ exports.Service = new Schema({
         enum: [NUNAVLBL_NTF_SIGN_NO, NUNAVLBL_NTF_SIGN_YES],
         required: true,
         message: {
-            type: path =>
+            type: path => `Признак необходимости оповещения о простое внешнего сервиса (${path}) имеет некорректный тип данных (ожидалось - Number)`,
-                `Признак необходимости оповещения о простое внешнего сервиса (${path}) имеет некорректный тип данных (ожидалось - Number)`,
+            enum: path => `Значение признака необходимости оповещения о простое внешнего сервиса (${path}) не поддерживается`,
-            enum: path =>
-                `Значение признака необходимости оповещения о простое внешнего сервиса (${path}) не поддерживается`,
             required: path => `Не указан признак необходимости оповещения о простое внешнего сервиса (${path})`
         }
     },
@@ -159,10 +169,8 @@ exports.Service = new Schema({
         message: {
             type: path =>
                 `Строковый код признака необходимости оповещения о простое внешнего сервиса (${path}) имеет некорректный тип данных (ожидалось - String)`,
-            enum: path =>
+            enum: path => `Значение строкового кода признака необходимости оповещения о простое внешнего сервиса (${path}) не поддерживается`,
-                `Значение строкового кода признака необходимости оповещения о простое внешнего сервиса (${path}) не поддерживается`,
+            required: path => `Не указан строковый код признака необходимости оповещения о простое внешнего сервиса (${path})`
-            required: path =>
-                `Не указан строковый код признака необходимости оповещения о простое внешнего сервиса (${path})`
         }
     },
     //Максимальное время простоя (мин) удалённого сервиса для генерации оповещения
@@ -172,8 +180,7 @@ exports.Service = new Schema({
         message: {
             type: path =>
                 `Максимальное время простоя (мин) удалённого сервиса для генерации оповещения (${path}) имеет некорректный тип данных (ожидалось - Number)`,
-            required: path =>
+            required: path => `Не указано максимальное время простоя (мин) удалённого сервиса для генерации оповещения (${path})`
-                `Не указано максимальное время простоя (мин) удалённого сервиса для генерации оповещения (${path})`
         }
     },
     //Список адресов E-Mail для оповещения о простое внешнего сервиса
@@ -194,8 +201,7 @@ exports.Service = new Schema({
         type: String,
         required: false,
         message: {
-            type: path =>
+            type: path => `Адрес прокси-сервера в очереди обмена (${path}) имеет некорректный тип данных (ожидалось - String)`,
-                `Адрес прокси-сервера в очереди обмена (${path}) имеет некорректный тип данных (ожидалось - String)`,
             required: path => `Не указан адрес прокси-сервера в очереди обмена (${path})`
         }
     },
@@ -237,8 +243,7 @@ exports.ServiceCtx = new Schema({
         type: String,
         required: false,
         message: {
-            type: path =>
+            type: path => `Строковое представление даты истечения контекста (${path}) имеет некорректный тип данных (ожидалось - String)`,
-                `Строковое представление даты истечения контекста (${path}) имеет некорректный тип данных (ожидалось - String)`,
             required: path => `Не указано строковое представление даты истечения контекста (${path})`
         }
     },
@@ -248,8 +253,7 @@ exports.ServiceCtx = new Schema({
         enum: [NIS_AUTH_YES, NIS_AUTH_NO],
         required: true,
         message: {
-            type: path =>
+            type: path => `Признака аутентицированности сервиса (${path}) имеет некорректный тип данных (ожидалось - Number)`,
-                `Признака аутентицированности сервиса (${path}) имеет некорректный тип данных (ожидалось - Number)`,
             enum: path => `Значение признака аутентицированности сервиса (${path}) не поддерживается`,
             required: path => `Не указан признак аутентицированности сервиса (${path})`
         }
@@ -260,8 +264,7 @@ exports.ServiceCtx = new Schema({
         enum: [SIS_AUTH_YES, SIS_AUTH_NO],
         required: true,
         message: {
-            type: path =>
+            type: path => `Строковый код признака аутентицированности сервиса (${path}) имеет некорректный тип данных (ожидалось - String)`,
-                `Строковый код признака аутентицированности сервиса (${path}) имеет некорректный тип данных (ожидалось - String)`,
             enum: path => `Значение строкового кода признака аутентицированности сервиса (${path}) не поддерживается`,
             required: path => `Не указан строковый код признака аутентицированности сервиса (${path})`
         }
@@ -284,8 +287,7 @@ exports.ServiceExpiredQueueInfo = new Schema({
         type: Number,
         required: true,
         message: {
-            type: path =>
+            type: path => `Количество просроченных сообщений обмена (${path}) имеет некорректный тип данных (ожидалось - Number)`,
-                `Количество просроченных сообщений обмена (${path}) имеет некорректный тип данных (ожидалось - Number)`,
             required: path => `Не указано количество просроченных сообщений обмена (${path})`
         }
     },
@@ -294,8 +296,7 @@ exports.ServiceExpiredQueueInfo = new Schema({
         type: String,
         required: true,
         message: {
-            type: path =>
+            type: path => `Информация о просроченных сообщениях обмена (${path}) имеет некорректный тип данных (ожидалось - String)`,
-                `Информация о просроченных сообщениях обмена (${path}) имеет некорректный тип данных (ожидалось - String)`,
             required: path => `Не указана информация о просроченных сообщениях обмена (${path})`
         }
     }

models/obj_service_function.js

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

package.json

@@ -21,8 +21,10 @@
         "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",