core/db_connector.js

@@ -540,35 +540,6 @@ class DBConnector extends EventEmitter {
             throw new ServerError(SERR_DB_EXECUTE, "Нет подключения к БД");
         }
     }
-    //Установить признак "В работе" позиции очереди
-    async setInProgress(prms) {
-        if (this.bConnected) {
-            //Проверяем структуру переданных параметров
-            let sCheckResult = validateObject(
-                prms,
-                prmsDBConnectorSchema.setInProgress,
-                `Параметры функции установки признака "В работе" позиции очереди`
-            );
-            //Если структура объекта в норме
-            if (!sCheckResult) {
-                //Подготовим параметры
-                let setInProgressData = deepClone(prms);
-                setInProgressData.connection = this.connection;
-                try {
-                    //Исполняем действие в БД
-                    await this.connector.setInProgress(setInProgressData);
-                    //Успешно - возвращаем ничего
-                    return;
-                } catch (e) {
-                    throw new ServerError(SERR_DB_EXECUTE, e.message);
-                }
-            } else {
-                throw new ServerError(SERR_OBJECT_BAD_INTERFACE, sCheckResult);
-            }
-        } else {
-            throw new ServerError(SERR_DB_EXECUTE, "Нет подключения к БД");
-        }
-    }
     //Считать очередную порцию исходящих сообщений
     async getOutgoing(prms) {
         if (this.bConnected) {

core/out_queue.js

@@ -173,18 +173,15 @@ class OutQueue extends EventEmitter {
         }
     }
     //Останов обработчика
-    async stopQueueProcessor(prms) {
+    stopQueueProcessor(prms) {
         //Проверяем структуру переданного объекта для останова обработчика
         let sCheckResult = validateObject(prms, prmsOutQueueSchema.stopQueueProcessor, "Параметры функции останова обработчика сообщения очереди");
         //Если структура объекта в норме
         if (!sCheckResult) {
             //Удаляем идентификатор позиции очереди из списка обрабатываемых
             this.rmInProgress({ nQueueId: prms.nQueueId });
-            //Сбрасываем признак "В работе" позиции очереди
+            //Завершаем дочерний процесс обработчика
-            await this.dbConn.setInProgress({
+            prms.proc.kill();
-                nQueueId: prms.nQueueId,
-                nInProgress: objQueueSchema.NQUEUE_IN_PROGRESS_NO
-            });
             //Увеличиваем количество доступных обработчиков
             this.nWorkersLeft++;
         } else {
@@ -298,8 +295,7 @@ class OutQueue extends EventEmitter {
                     if (prms.queue.nExecState == objQueueSchema.NQUEUE_EXEC_STATE_ERR) await this.notifyMessageProcessError(prms);
                     //Останавливаем обработчик и инкрементируем флаг их доступного количества
                     try {
-                        //Завершаем дочерний процесс обработчика
+                        this.stopQueueProcessor({ nQueueId: prms.queue.nId, proc });
-                        proc.kill();
                     } catch (e) {
                         //Отразим в протоколе ошибку останова
                         await self.logger.error(`Ошибка останова обработчика исходящего сообщения: ${makeErrorText(e)}`, {
@@ -329,8 +325,7 @@ class OutQueue extends EventEmitter {
                     if (prms.queue.nExecState == objQueueSchema.NQUEUE_EXEC_STATE_ERR) await this.notifyMessageProcessError(prms);
                     //Останавливаем обработчик и инкрементируем флаг их доступного количества
                     try {
-                        //Завершаем дочерний процесс обработчика
+                        this.stopQueueProcessor({ nQueueId: prms.queue.nId, proc });
-                        proc.kill();
                     } catch (e) {
                         //Отразим в протоколе ошибку останова
                         await self.logger.error(`Ошибка останова обработчика исходящего сообщения: ${makeErrorText(e)}`, {
@@ -339,17 +334,7 @@ class OutQueue extends EventEmitter {
                     }
                 });
                 //Перехват останова обработчика
-                proc.on("exit", async code => {
+                proc.on("exit", code => {});
-                    try {
-                        //Завершаем процесс обработки сообщения
-                        this.stopQueueProcessor({ nQueueId: prms.queue.nId });
-                    } catch (e) {
-                        //Отразим в протоколе ошибку останова
-                        await self.logger.error(`Ошибка останова обработчика исходящего сообщения: ${makeErrorText(e)}`, {
-                            nQueueId: prms.queue.nId
-                        });
-                    }
-                });
                 //Запускаем обработчик
                 this.startQueueProcessor({ queue: prms.queue, proc });
             }

models/obj_queue.js

@@ -33,10 +33,6 @@ const SQUEUE_EXEC_STATE_DB_ERR = "DB_ERR"; //Ошибка обработки С
 const SQUEUE_EXEC_STATE_OK = "OK"; //Обработано успешно (строковый код)
 const SQUEUE_EXEC_STATE_ERR = "ERR"; //Обработано с ошибками (строковый код)
 
-//Значения признака "В работе" сообщения очереди обмена
-const NQUEUE_IN_PROGRESS_NO = 0; //Не в работе
-const NQUEUE_IN_PROGRESS_YES = 1; //В работе
-
 //Коды результатов исполнения обработчика сообщения
 const SPRC_RESP_RESULT_OK = "OK"; //Обработано успешно
 const SPRC_RESP_RESULT_ERR = "ERR"; //Ошибка обработки
@@ -74,8 +70,6 @@ exports.SPRC_RESP_RESULT_ERR = SPRC_RESP_RESULT_ERR;
 exports.SPRC_RESP_RESULT_UNAUTH = SPRC_RESP_RESULT_UNAUTH;
 exports.NQUEUE_RESET_DATA_NO = NQUEUE_RESET_DATA_NO;
 exports.NQUEUE_RESET_DATA_YES = NQUEUE_RESET_DATA_YES;
-exports.NQUEUE_IN_PROGRESS_NO = NQUEUE_IN_PROGRESS_NO;
-exports.NQUEUE_IN_PROGRESS_YES = NQUEUE_IN_PROGRESS_YES;
 
 //Схема валидации сообщения очереди обмена
 exports.Queue = new Schema({
@@ -84,7 +78,8 @@ exports.Queue = new Schema({
         type: Number,
         required: true,
         message: {
-            type: path => `Идентификатор сообщения очереди обмена (${path}) имеет некорректный тип данных (ожидалось - Number)`,
+            type: path =>
+                `Идентификатор сообщения очереди обмена (${path}) имеет некорректный тип данных (ожидалось - Number)`,
             required: path => `Не указан идентификатор сообщения очереди обмена (${path})`
         }
     },
@@ -93,7 +88,8 @@ exports.Queue = new Schema({
         type: Date,
         required: true,
         message: {
-            type: path => `Дата постановки сообщения в очередь обмена (${path}) имеет некорректный тип данных (ожидалось - Date)`,
+            type: path =>
+                `Дата постановки сообщения в очередь обмена (${path}) имеет некорректный тип данных (ожидалось - Date)`,
             required: path => `Не указана дата постановки сообщения в очередь обмена (${path})`
         }
     },
@@ -112,7 +108,8 @@ exports.Queue = new Schema({
         type: String,
         required: true,
         message: {
-            type: path => `Пользователь, поставивший сообщение в очередь обмена (${path}) имеет некорректный тип данных (ожидалось - String)`,
+            type: path =>
+                `Пользователь, поставивший сообщение в очередь обмена (${path}) имеет некорректный тип данных (ожидалось - String)`,
             required: path => `Не указан пользователь, поставивший сообщение в очередь обмена (${path})`
         }
     },
@@ -121,7 +118,8 @@ exports.Queue = new Schema({
         type: Number,
         required: true,
         message: {
-            type: path => `Идентификатор сервиса-обработчика сообщения очереди обмена (${path}) имеет некорректный тип данных (ожидалось - Number)`,
+            type: path =>
+                `Идентификатор сервиса-обработчика сообщения очереди обмена (${path}) имеет некорректный тип данных (ожидалось - Number)`,
             required: path => `Не указан идентификатор сервиса-обработчика сообщения очереди обмена (${path})`
         }
     },
@@ -130,7 +128,8 @@ exports.Queue = new Schema({
         type: String,
         required: true,
         message: {
-            type: path => `Код сервиса-обработчика сообщения очереди обмена (${path}) имеет некорректный тип данных (ожидалось - String)`,
+            type: path =>
+                `Код сервиса-обработчика сообщения очереди обмена (${path}) имеет некорректный тип данных (ожидалось - String)`,
             required: path => `Не указан код сервиса-обработчика сообщения очереди обмена (${path})`
         }
     },
@@ -149,7 +148,8 @@ exports.Queue = new Schema({
         type: String,
         required: true,
         message: {
-            type: path => `Код функции сервиса-обработчика сообщения очереди обмена (${path}) имеет некорректный тип данных (ожидалось - String)`,
+            type: path =>
+                `Код функции сервиса-обработчика сообщения очереди обмена (${path}) имеет некорректный тип данных (ожидалось - String)`,
             required: path => `Не указан код функции сервиса-обработчика сообщения очереди обмена (${path})`
         }
     },
@@ -158,7 +158,8 @@ exports.Queue = new Schema({
         type: Date,
         required: false,
         message: {
-            type: path => `Дата обработки сообщения очереди обмена (${path}) имеет некорректный тип данных (ожидалось - Date)`,
+            type: path =>
+                `Дата обработки сообщения очереди обмена (${path}) имеет некорректный тип данных (ожидалось - Date)`,
             required: path => `Не указана дата обработки сообщения очереди обмена (${path})`
         }
     },
@@ -177,7 +178,8 @@ exports.Queue = new Schema({
         type: Number,
         required: true,
         message: {
-            type: path => `Количество попыток обработки сообщения очереди обмена (${path}) имеет некорректный тип данных (ожидалось - Number)`,
+            type: path =>
+                `Количество попыток обработки сообщения очереди обмена (${path}) имеет некорректный тип данных (ожидалось - Number)`,
             required: path => `Не указано количество попыток обработки сообщения очереди обмена (${path})`
         }
     },
@@ -207,7 +209,8 @@ exports.Queue = new Schema({
         ],
         required: true,
         message: {
-            type: path => `Состояние обработки сообщения очереди обмена (${path}) имеет некорректный тип данных (ожидалось - Number)`,
+            type: path =>
+                `Состояние обработки сообщения очереди обмена (${path}) имеет некорректный тип данных (ожидалось - Number)`,
             enum: path => `Значение состояния обработки сообщения очереди обмена (${path}) не поддерживается`,
             required: path => `Не указано состояние обработки сообщения очереди обмена (${path})`
         }
@@ -228,8 +231,10 @@ exports.Queue = new Schema({
         ],
         required: true,
         message: {
-            type: path => `Строковый код состояния обработки сообщения очереди обмена (${path}) имеет некорректный тип данных (ожидалось - String)`,
+            type: path =>
-            enum: path => `Значение строкового кода состояния обработки сообщения очереди обмена (${path}) не поддерживается`,
+                `Строковый код состояния обработки сообщения очереди обмена (${path}) имеет некорректный тип данных (ожидалось - String)`,
+            enum: path =>
+                `Значение строкового кода состояния обработки сообщения очереди обмена (${path}) не поддерживается`,
             required: path => `Не указан строковый код состояния обработки сообщения очереди обмена (${path})`
         }
     },
@@ -238,7 +243,8 @@ exports.Queue = new Schema({
         type: String,
         required: false,
         message: {
-            type: path => `Информация от обработчика сообщения очереди обмена (${path}) имеет некорректный тип данных (ожидалось - String)`,
+            type: path =>
+                `Информация от обработчика сообщения очереди обмена (${path}) имеет некорректный тип данных (ожидалось - String)`,
             required: path => `Не указана информация от обработчика сообщения очереди обмена (${path})`
         }
     },
@@ -247,7 +253,8 @@ exports.Queue = new Schema({
         type: Number,
         required: false,
         message: {
-            type: path => `Идентификатор связанного сообщения очереди обмена (${path}) имеет некорректный тип данных (ожидалось - Number)`,
+            type: path =>
+                `Идентификатор связанного сообщения очереди обмена (${path}) имеет некорректный тип данных (ожидалось - Number)`,
             required: path => `Не указан идентификатор связанного сообщения очереди обмена (${path})`
         }
     },
@@ -256,7 +263,8 @@ exports.Queue = new Schema({
         type: String,
         required: false,
         message: {
-            type: path => `Параметры сообщения очереди обмена (${path}) имеют некорректный тип данных (ожидалось - String)`,
+            type: path =>
+                `Параметры сообщения очереди обмена (${path}) имеют некорректный тип данных (ожидалось - String)`,
             required: path => `Не указаны параметры сообщения очереди обмена (${path})`
         }
     },
@@ -265,7 +273,8 @@ exports.Queue = new Schema({
         type: String,
         required: false,
         message: {
-            type: path => `Параметры ответа на сообщение очереди обмена (${path}) имеют некорректный тип данных (ожидалось - String)`,
+            type: path =>
+                `Параметры ответа на сообщение очереди обмена (${path}) имеют некорректный тип данных (ожидалось - String)`,
             required: path => `Не указаны параметры ответа на сообщение очереди обмена (${path})`
         }
     },
@@ -274,7 +283,8 @@ exports.Queue = new Schema({
         type: Number,
         required: true,
         message: {
-            type: path => `Приоритет в очереди обмена (${path}) имеет некорректный тип данных (ожидалось - Number)`,
+            type: path => 
+                `Приоритет в очереди обмена (${path}) имеет некорректный тип данных (ожидалось - Number)`,
             required: path => `Не указан приоритет в очереди обмена (${path})`
         }
     }
@@ -287,7 +297,8 @@ exports.QueueMsg = new Schema({
         type: Buffer,
         required: true,
         message: {
-            type: path => `Данные сообщения очереди обмена (${path}) имеют некорректный тип данных (ожидалось - Buffer)`,
+            type: path =>
+                `Данные сообщения очереди обмена (${path}) имеют некорректный тип данных (ожидалось - Buffer)`,
             required: path => `Не указаны данные сообщения очереди обмена (${path})`
         }
     }
@@ -300,7 +311,8 @@ exports.QueueResp = new Schema({
         type: Buffer,
         required: true,
         message: {
-            type: path => `Данные ответа сообщения очереди обмена (${path}) имеют некорректный тип данных (ожидалось - Buffer)`,
+            type: path =>
+                `Данные ответа сообщения очереди обмена (${path}) имеют некорректный тип данных (ожидалось - Buffer)`,
             required: path => `Не указаны данные ответа сообщения очереди обмена (${path})`
         }
     }
@@ -314,7 +326,8 @@ exports.QueuePrcResult = new Schema({
         enum: [SPRC_RESP_RESULT_OK, SPRC_RESP_RESULT_ERR, SPRC_RESP_RESULT_UNAUTH],
         required: true,
         message: {
-            type: path => `Состояние обработки сообщения очереди обмена (${path}) имеет некорректный тип данных (ожидалось - String)`,
+            type: path =>
+                `Состояние обработки сообщения очереди обмена (${path}) имеет некорректный тип данных (ожидалось - String)`,
             enum: path => `Значение состояния обработки сообщения очереди обмена (${path}) не поддерживается`,
             required: path => `Не указано состояние обработки сообщения очереди обмена (${path})`
         }
@@ -324,7 +337,8 @@ exports.QueuePrcResult = new Schema({
         type: String,
         required: false,
         message: {
-            type: path => `Параметры ответа на сообщение очереди обмена (${path}) имеют некорректный тип данных (ожидалось - String)`,
+            type: path =>
+                `Параметры ответа на сообщение очереди обмена (${path}) имеют некорректный тип данных (ожидалось - String)`,
             required: path => `Не указаны параметры ответа на сообщение очереди обмена (${path})`
         }
     },
@@ -333,7 +347,8 @@ exports.QueuePrcResult = new Schema({
         type: String,
         required: true,
         message: {
-            type: path => `Информация от обработчика сообщения очереди обмена (${path}) имеет некорректный тип данных (ожидалось - String)`,
+            type: path =>
+                `Информация от обработчика сообщения очереди обмена (${path}) имеет некорректный тип данных (ожидалось - String)`,
             required: path => `Не указана информация от обработчика сообщения очереди обмена (${path})`
         }
     }

models/prms_db_connector.js

@@ -22,9 +22,7 @@ const {
     NQUEUE_EXEC_STATE_OK,
     NQUEUE_EXEC_STATE_ERR,
     NQUEUE_RESET_DATA_NO,
-    NQUEUE_RESET_DATA_YES,
+    NQUEUE_RESET_DATA_YES
-    NQUEUE_IN_PROGRESS_NO,
-    NQUEUE_IN_PROGRESS_YES
 } = require("./obj_queue"); //Схемы валидации сообщения очереди обмена
 
 //----------
@@ -330,29 +328,6 @@ exports.putQueue = new Schema({
     }
 });
 
-//Схема валидации параметров функции установки признака "В работе" позиции очереди
-exports.setInProgress = new Schema({
-    //Идентификатор позиции очереди
-    nQueueId: {
-        type: Number,
-        required: true,
-        message: {
-            type: path => `Идентификатор позиции очереди (${path}) имеет некорректный тип данных (ожидалось - Number)`,
-            required: path => `Не указан идентификатор позиции очереди (${path})`
-        }
-    },
-    nInProgress: {
-        type: Number,
-        enum: [NQUEUE_IN_PROGRESS_NO, NQUEUE_IN_PROGRESS_YES],
-        required: true,
-        message: {
-            type: path => `Признак "В работе" (${path}) имеет некорректный тип данных (ожидалось - Number)`,
-            enum: path => `Значение признака "В работе" (${path}) не поддерживается`,
-            required: path => `Не указано значение признака "В работе" (${path})`
-        }
-    }
-});
-
 //Схема валидации параметров функции считывания исходящих сообщений
 exports.getOutgoing = new Schema({
     //Количество считываемых сообщений очереди

models/prms_out_queue.js

@@ -42,7 +42,8 @@ exports.OutQueue = new Schema({
         type: DBConnector,
         required: true,
         message: {
-            type: path => `Объект для взаимодействия с БД (${path}) имеет некорректный тип данных (ожидалось - DBConnector)`,
+            type: path =>
+                `Объект для взаимодействия с БД (${path}) имеет некорректный тип данных (ожидалось - DBConnector)`,
             required: path => `Не указан объект для взаимодействия с БД (${path})`
         }
     },
@@ -51,7 +52,8 @@ exports.OutQueue = new Schema({
         type: Logger,
         required: true,
         message: {
-            type: path => `Объект для протоколирования работы (${path}) имеет некорректный тип данных (ожидалось - Logger)`,
+            type: path =>
+                `Объект для протоколирования работы (${path}) имеет некорректный тип данных (ожидалось - Logger)`,
             required: path => `Не указаны объект для протоколирования работы (${path})`
         }
     },
@@ -60,7 +62,8 @@ exports.OutQueue = new Schema({
         type: Notifier,
         required: true,
         message: {
-            type: path => `Объект для рассылки уведомлений (${path}) имеет некорректный тип данных (ожидалось - Notifier)`,
+            type: path =>
+                `Объект для рассылки уведомлений (${path}) имеет некорректный тип данных (ожидалось - Notifier)`,
             required: path => `Не указан объект для рассылки уведомлений (${path})`
         }
     },
@@ -129,7 +132,8 @@ exports.startQueueProcessor = new Schema({
         use: { validateChildProcess },
         required: true,
         message: {
-            validateChildProcess: path => `Процесс обработчика (${path}) имеет некорректный тип данных (ожидалось - ChildProcess)`,
+            validateChildProcess: path =>
+                `Процесс обработчика (${path}) имеет некорректный тип данных (ожидалось - ChildProcess)`,
             required: path => `Не указан процесс обработчика (${path})`
         }
     }
@@ -145,6 +149,16 @@ exports.stopQueueProcessor = new Schema({
             type: path => `Идентификатор сообщения (${path}) имеет некорректный тип данных (ожидалось - Number)`,
             required: path => `Не указан идентификатор сообщения (${path})`
         }
+    },
+    //Процесс обработчика
+    proc: {
+        use: { validateChildProcess },
+        required: true,
+        message: {
+            validateChildProcess: path =>
+                `Процесс обработчика (${path}) имеет некорректный тип данных (ожидалось - ChildProcess)`,
+            required: path => `Не указан процесс обработчика (${path})`
+        }
     }
 });
 

modules/parus_oracle_db.js

@@ -412,18 +412,6 @@ const putQueue = async prms => {
     return queueData.RCQUEUE[0];
 };
 
-//Установка признака "В работе" в сообщении очереди
-const setInProgress = async prms => {
-    await executeStored({
-        connection: prms.connection,
-        sName: "PKG_EXS.QUEUE_IN_PROGRESS_SET",
-        inPrms: {
-            NEXSQUEUE: prms.nQueueId,
-            NIN_PROGRESS: prms.nInProgress
-        }
-    });
-};
-
 //Считывание очередной порции исходящих сообщений из очереди
 const getQueueOutgoing = async prms => {
     let queueOutgoingData = await executeStored({
@@ -578,7 +566,6 @@ exports.getServiceExpiredQueueInfo = getServiceExpiredQueueInfo;
 exports.log = log;
 exports.getQueue = getQueue;
 exports.putQueue = putQueue;
-exports.setInProgress = setInProgress;
 exports.getQueueOutgoing = getQueueOutgoing;
 exports.setQueueState = setQueueState;
 exports.getQueueMsg = getQueueMsg;

modules/parus_pg_db.js

@@ -373,18 +373,6 @@ const putQueue = async prms => {
     return queueData.RCQUEUE[0];
 };
 
-//Установка признака "В работе" в сообщении очереди
-const setInProgress = async prms => {
-    await executeStored({
-        connection: prms.connection,
-        sName: "PKG_EXS$QUEUE_IN_PROGRESS_SET",
-        inPrms: {
-            NEXSQUEUE: prms.nQueueId,
-            NIN_PROGRESS: prms.nInProgress
-        }
-    });
-};
-
 //Считывание очередной порции исходящих сообщений из очереди
 const getQueueOutgoing = async prms => {
     let queueOutgoingData = await executeStored({
@@ -539,7 +527,6 @@ exports.getServiceExpiredQueueInfo = getServiceExpiredQueueInfo;
 exports.log = log;
 exports.getQueue = getQueue;
 exports.putQueue = putQueue;
-exports.setInProgress = setInProgress;
 exports.getQueueOutgoing = getQueueOutgoing;
 exports.setQueueState = setQueueState;
 exports.getQueueMsg = getQueueMsg;