CITKParus/P8-ExchangeService
7
0
Fork 1
Code Issues Pull Requests Projects Releases Wiki Activity

ЦИТК-659 - Возможность работы нескольких серверов параллельно

Browse Source
This commit is contained in:
Dollerino 2025-05-15 19:03:29 +03:00
parent 45d149dd23
commit 0e2a5b6bd2
8 changed files with 320 additions and 145 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
config.js
View File

@ -9,6 +9,8 @@
//Общие параметры
let common = {
//Наименование сервера приложений
sServerName: "",
//Версия сервера приложений
sVersion: "8.5.6.1",
//Релиз сервера приложений

10
core/app.js
View File

@ -177,7 +177,7 @@ class ParusAppServer {
if (this.dbConn.bConnected) {
await this.logger.warn("Отключение сервера приложений от БД...");
try {
await this.dbConn.disconnect();
await this.dbConn.disconnect({ bServerClose: true });
} catch (e) {
await this.logger.error(`Ошибка отключения от БД: ${e.sCode}: ${e.sMessage}`);
}
@ -201,6 +201,12 @@ class ParusAppServer {
this.dbConn = new db.DBConnector({
connectSettings: {
...prms.config.dbConnect,
exsSrv: {
sServerName: prms.config.common.sServerName,
sServerIP: `порт - ${prms.config.inComing.nPort}, доступные IP - ${
prms.config.inComing.sHost === "0.0.0.0" ? getIPs().join("; ") : prms.config.inComing.sHost
}`
},
sRelease: prms.config.common.sRelease,
bControlSystemVersion: prms.config.common.bControlSystemVersion,
nPoolMin: prms.config.inComing.nPoolMin,
@ -269,7 +275,7 @@ class ParusAppServer {
this.srvAvlCtrl.on(sac.SEVT_SERVICE_AVAILABLE_CONTROLLER_STOPPED, this.onServiceACStopped);
//Подключаемся к БД
await this.logger.info("Подключение сервера приложений к БД...");
await this.dbConn.connect();
await this.dbConn.connect({ bServerStart: true });
}
//Останов сервера
async stop(terminateTimeout) {

38
core/db_connector.js
View File

@ -73,6 +73,7 @@ class DBConnector extends EventEmitter {
//Инициализируем остальные свойства
this.connection = null;
this.bConnected = false;
this.nExsSrv = null;
} else {
throw new ServerError(SERR_MODULES_NO_MODULE_SPECIFIED, "Не указано имя подключаемого модуля-коннектора!");
}
@ -94,12 +95,23 @@ class DBConnector extends EventEmitter {
}
}
//Подключиться к БД
async connect() {
async connect(prms) {
//Подключаемся только если ещё не подключены
if (!this.bConnected) {
//Проверяем структуру переданного объекта с параметрами для подключения к БД
let sCheckResult = validateObject(prms, prmsDBConnectorSchema.connect, "Параметры функции подключения к БД");
//Если структура объекта в норме
if (!sCheckResult) {
try {
//Подготовим параметры для передачи в БД
let connectData = _.cloneDeep(prms);
connectData.connectSettings = this.connectSettings;
//Подключаемся
this.connection = await this.connector.connect(this.connectSettings);
let res = await this.connector.connect(connectData);
//Сохраняем информацию о подключении
this.connection = res.pool;
//Сохраняем информацию о сервере интеграции
this.nExsSrv = res.nExsSrv;
//Выставим внутренний флаг подключения
this.bConnected = true;
//Расскажем всем, что подключились
@ -109,15 +121,26 @@ class DBConnector extends EventEmitter {
} catch (e) {
throw new ServerError(SERR_DB_CONNECT, e.message);
}
} else {
throw new ServerError(SERR_OBJECT_BAD_INTERFACE, sCheckResult);
}
}
}
//Отключиться от БД
async disconnect() {
async disconnect(prms) {
//Смысл отключаться есть только когда мы подключены, в противном случае - зачем тратить время
if (this.bConnected) {
//Проверяем структуру переданного объекта с параметрами для подключения к БД
let sCheckResult = validateObject(prms, prmsDBConnectorSchema.disconnect, "Параметры функции отключения от БД");
//Если структура объекта в норме
if (!sCheckResult) {
try {
//Подготовим параметры для передачи в БД
let disconnectData = _.cloneDeep(prms);
disconnectData.connection = this.connection;
disconnectData.nExsSrv = this.nExsSrv;
//Отключаемся
await this.connector.disconnect({ connection: this.connection });
await this.connector.disconnect(disconnectData);
//Забываем подключение и удаляем флаги подключенности
this.connection = null;
this.bConnected = false;
@ -128,6 +151,9 @@ class DBConnector extends EventEmitter {
} catch (e) {
throw new ServerError(SERR_DB_DISCONNECT, e.message);
}
} else {
throw new ServerError(SERR_OBJECT_BAD_INTERFACE, sCheckResult);
}
}
}
//Получить список сервисов
@ -136,7 +162,7 @@ class DBConnector extends EventEmitter {
if (this.bConnected) {
try {
//Забираем список сервисов и декорируем его заготовками под список функций
let srvs = await this.connector.getServices({ connection: this.connection });
let srvs = await this.connector.getServices({ connection: this.connection, nExsSrv: this.nExsSrv });
srvs.forEach(s => {
s.functions = [];
});
@ -386,6 +412,7 @@ class DBConnector extends EventEmitter {
//Подготовим параметры для передачи в БД
let logData = _.cloneDeep(prms);
logData.connection = this.connection;
logData.nExsSrv = this.nExsSrv;
//И выполним запись в журнал
let res = await this.connector.log(logData);
//Валидируем полученный ответ
@ -521,6 +548,7 @@ class DBConnector extends EventEmitter {
//Подготовим параметры для передачи в БД
let getQueueOutgoingData = _.cloneDeep(prms);
getQueueOutgoingData.connection = this.connection;
getQueueOutgoingData.nExsSrv = this.nExsSrv;
//Выполняем считывание из БД
let res = await this.connector.getQueueOutgoing(getQueueOutgoingData);
//Валидируем полученный ответ

6
core/out_queue_processor.js
View File

@ -604,7 +604,7 @@ const processTask = async prms => {
logger.removeDBConnector();
});
//Подключаемся к БД
await dbConn.connect();
await dbConn.connect({ bServerStart: false });
//Считываем запись очереди
q = await dbConn.getQueue({ nQueueId: prms.task.nQueueId });
//Далее работаем от статуса считанной записи
@ -709,12 +709,12 @@ const processTask = async prms => {
}
}
//Отключаемся от БД
if (dbConn) await dbConn.disconnect();
if (dbConn) await dbConn.disconnect({ bServerClose: false });
//Отправляем успех
sendOKResult();
} catch (e) {
//Отключаемся от БД
if (dbConn) await dbConn.disconnect();
if (dbConn) await dbConn.disconnect({ bServerClose: false });
//Отправляем ошибку
if (e instanceof ServerError && e.sCode == SERR_UNAUTH) sendUnAuthResult();
else sendErrorResult({ sMessage: makeErrorText(e) });

21
models/obj_config.js
View File

@ -13,6 +13,9 @@ const Schema = require("validate"); //Схемы валидации
// Тело модуля
//-------------
//Функция проверки длины значения наименование сервера приложений
const validateServerName = val => val.length <= 40;
//Функция проверки значения таймаута останова сервера
const validateTerminateTimeout = val => val >= 1000 && val <= 120000 && Number.isInteger(val);
@ -46,8 +49,21 @@ const validatePoolMaxInComing = val => val >= 1 && val <= 1000;
//Функция проверки значения шага инкремента подключений к БД в пуле обработчика входящих сообщений
const validatePoolIncrementInComing = val => val >= 0 && val <= 1000;
//Функция проверки наименования сервера приложений в сессии БД
const validateSessionAppName = val => val === "PARUS$ExchangeServer";
//Схема валидации общих параметров сервера приложений
const common = new Schema({
//Наименование сервера приложений
sServerName: {
type: String,
required: false,
use: { validateServerName },
message: {
type: path => `Наименование сервера приложений (${path}) имеет некорректный тип данных (ожидалось - String)`,
validateServerName: path => `Длина наименования сервера приложений (${path}) не может превышать 40 символов`
}
},
//Версия сервера приложений
sVersion: {
type: String,
@ -130,9 +146,12 @@ const dbConnect = new Schema({
sSessionAppName: {
type: String,
required: true,
use: { validateSessionAppName },
message: {
type: path => `Наименование сервера приложений в сессии БД (${path}) имеет некорректный тип данных (ожидалось - String)`,
required: path => `Не указано наименование сервера приложений в сессии БД (${path})`
required: path => `Не указано наименование сервера приложений в сессии БД (${path})`,
validateSessionAppName: path =>
`Наименование сервера приложений в сессии БД поддерживает только значение "PARUS$ExchangeServer" (${path})`
}
},
//Наименование подключаемого модуля обслуживания БД

86
models/prms_db_connector.js
View File

@ -74,6 +74,32 @@ exports.DBConnector = new Schema({
}
});
//Схема валидации параметров функции подключения к БД
exports.connect = new Schema({
//Признак запуска сервера интеграции
bServerStart: {
type: Boolean,
required: true,
message: {
type: path => `Признак запуска сервера интеграции (${path}) имеет некорректный тип данных (ожидалось - Boolean)`,
required: path => `Не указан признак запуска сервера интеграции (${path})`
}
}
});
//Схема валидации параметров функции отключения от БД
exports.disconnect = new Schema({
//Признак выключения сервера интеграции
bServerClose: {
type: Boolean,
required: true,
message: {
type: path => `Признак выключения сервера интеграции (${path}) имеет некорректный тип данных (ожидалось - Boolean)`,
required: path => `Не указан признак выключения сервера интеграции (${path})`
}
}
});
//Схема валидации параметров функции получения списка функций сервиса
exports.getServiceFunctions = new Schema({
//Идентификатор сервиса
@ -174,8 +200,7 @@ exports.putServiceAuthInQueue = new Schema({
required: false,
enum: [NFORCE_YES, NFORCE_NO],
message: {
type: path =>
`Признак принудительного завершения сессии (${path}) имеет некорректный тип данных (ожидалось - Number)`,
type: path => `Признак принудительного завершения сессии (${path}) имеет некорректный тип данных (ожидалось - Number)`,
enum: path => `Значение признака принудительного завершения сессии (${path}) не поддерживается`,
required: path => `Не указан признак принудительного завершения сессии (${path})`
}
@ -203,8 +228,7 @@ exports.putLog = new Schema({
enum: [NLOG_STATE_INF, NLOG_STATE_WRN, NLOG_STATE_ERR],
required: true,
message: {
type: path =>
`Тип сообщения журнала работы сервиса (${path}) имеет некорректный тип данных (ожидалось - Number)`,
type: path => `Тип сообщения журнала работы сервиса (${path}) имеет некорректный тип данных (ожидалось - Number)`,
enum: path => `Значение типа сообщения журнала работы сервиса (${path}) не поддерживается`,
required: path => `Не указан тип сообщения журнала работы сервиса (${path})`
}
@ -214,8 +238,7 @@ exports.putLog = new Schema({
type: String,
required: false,
message: {
type: path =>
`Сообщение журнала работы сервиса (${path}) имеет некорректный тип данных (ожидалось - String)`,
type: path => `Сообщение журнала работы сервиса (${path}) имеет некорректный тип данных (ожидалось - String)`,
required: path => `Не указано сообщение журнала работы сервиса (${path})`
}
},
@ -236,8 +259,7 @@ exports.putLog = new Schema({
message: {
type: path =>
`Идентификатор связанной функции-обработчика сообщения журнала работы сервиса (${path}) имеет некорректный тип данных (ожидалось - Number)`,
required: path =>
`Не указан идентификатор связанной функции-обработчика сообщения журнала работы сервиса (${path})`
required: path => `Не указан идентификатор связанной функции-обработчика сообщения журнала работы сервиса (${path})`
}
},
//Идентификатор связанной позиции очереди обмена
@ -247,8 +269,7 @@ exports.putLog = new Schema({
message: {
type: path =>
`Идентификатор связанной позиции очереди обмена сообщения журнала работы сервиса (${path}) имеет некорректный тип данных (ожидалось - Number)`,
required: path =>
`Не указан идентификатор связанной позиции очереди обмена сообщения журнала работы сервиса (${path})`
required: path => `Не указан идентификатор связанной позиции очереди обмена сообщения журнала работы сервиса (${path})`
}
}
});
@ -260,8 +281,7 @@ exports.getQueue = new Schema({
type: Number,
required: true,
message: {
type: path =>
`Идентификатор позиции очереди обмена (${path}) имеет некорректный тип данных (ожидалось - Number)`,
type: path => `Идентификатор позиции очереди обмена (${path}) имеет некорректный тип данных (ожидалось - Number)`,
required: path => `Не указан идентификатор позиции очереди обмена (${path})`
}
}
@ -274,8 +294,7 @@ exports.putQueue = new Schema({
type: Number,
required: true,
message: {
type: path =>
`Идентификатор функции сервиса обработчика позиции очереди (${path}) имеет некорректный тип данных (ожидалось - Number)`,
type: path => `Идентификатор функции сервиса обработчика позиции очереди (${path}) имеет некорректный тип данных (ожидалось - Number)`,
required: path => `Не указан идентификатор функции сервиса обработчика позиции очереди (${path})`
}
},
@ -284,8 +303,7 @@ exports.putQueue = new Schema({
use: { validateBuffer },
required: false,
message: {
validateBuffer: path =>
`Данные сообщения очереди обмена (${path}) имеют некорректный тип данных (ожидалось - null или Buffer)`,
validateBuffer: path => `Данные сообщения очереди обмена (${path}) имеют некорректный тип данных (ожидалось - null или Buffer)`,
required: path => `Не указаны данные сообщения очереди обмена (${path})`
}
},
@ -294,8 +312,7 @@ exports.putQueue = new Schema({
type: Number,
required: false,
message: {
type: path =>
`Идентификатор связанной позиции очереди обмена (${path}) имеет некорректный тип данных (ожидалось - Number)`,
type: path => `Идентификатор связанной позиции очереди обмена (${path}) имеет некорректный тип данных (ожидалось - Number)`,
required: path => `Не указан идентификатор связанной позиции очереди обмена (${path})`
}
},
@ -304,8 +321,7 @@ exports.putQueue = new Schema({
type: Number,
required: false,
message: {
type: path =>
`Идентификатор связанной организации (${path}) имеет некорректный тип данных (ожидалось - Number)`,
type: path => `Идентификатор связанной организации (${path}) имеет некорректный тип данных (ожидалось - Number)`,
required: path => `Не указан идентификатор связанной организации (${path})`
}
},
@ -314,8 +330,7 @@ exports.putQueue = new Schema({
type: Number,
required: false,
message: {
type: path =>
`Идентификатор связанного документа (${path}) имеет некорректный тип данных (ожидалось - Number)`,
type: path => `Идентификатор связанного документа (${path}) имеет некорректный тип данных (ожидалось - Number)`,
required: path => `Не указан идентификатор связанного документа (${path})`
}
},
@ -346,8 +361,7 @@ exports.getOutgoing = new Schema({
type: Number,
required: true,
message: {
type: path =>
`Количество считываемых сообщений очереди (${path}) имеет некорректный тип данных (ожидалось - Number)`,
type: path => `Количество считываемых сообщений очереди (${path}) имеет некорректный тип данных (ожидалось - Number)`,
required: path => `Не указано количество считываемых сообщений очереди (${path})`
}
}
@ -400,8 +414,7 @@ exports.setQueueState = new Schema({
enum: [NINC_EXEC_CNT_NO, NINC_EXEC_CNT_YES],
required: false,
message: {
type: path =>
`Флаг инкремента количества исполнений (${path}) имеет некорректный тип данных (ожидалось - Number)`,
type: path => `Флаг инкремента количества исполнений (${path}) имеет некорректный тип данных (ожидалось - Number)`,
enum: path => `Значение флага инкремента количества исполнений (${path}) не поддерживается`,
required: path => `Не указан флаг икремента количества исполнений (${path})`
}
@ -448,8 +461,7 @@ exports.setQueueMsg = new Schema({
use: { validateBuffer },
required: true,
message: {
validateBuffer: path =>
`Данные сообщения очереди обмена (${path}) имеют некорректный тип данных (ожидалось - null или Buffer)`,
validateBuffer: path => `Данные сообщения очереди обмена (${path}) имеют некорректный тип данных (ожидалось - null или Buffer)`,
required: path => `Не указаны данные сообщения очереди обмена (${path})`
}
}
@ -471,8 +483,7 @@ exports.setQueueOptions = new Schema({
type: String,
required: true,
message: {
validateBuffer: path =>
`Парамметры сообщения очереди обмена (${path}) имеют некорректный тип данных (ожидалось - String)`,
validateBuffer: path => `Парамметры сообщения очереди обмена (${path}) имеют некорректный тип данных (ожидалось - String)`,
required: path => `Не указаны параметры сообщения очереди обмена (${path})`
}
}
@ -507,8 +518,7 @@ exports.setQueueResp = new Schema({
use: { validateBuffer },
required: true,
message: {
validateBuffer: path =>
`Данные ответа сообщения очереди обмена (${path}) имеют некорректный тип данных (ожидалось - null или Buffer)`,
validateBuffer: path => `Данные ответа сообщения очереди обмена (${path}) имеют некорректный тип данных (ожидалось - null или Buffer)`,
required: path => `Не указаны данные ответа сообщения очереди обмена (${path})`
}
},
@ -518,8 +528,7 @@ exports.setQueueResp = new Schema({
enum: [NIS_ORIGINAL_NO, NIS_ORIGINAL_YES],
required: true,
message: {
type: path =>
`Признак передачи оригинала ответа (${path}) имеет некорректный тип данных (ожидалось - Number)`,
type: path => `Признак передачи оригинала ответа (${path}) имеет некорректный тип данных (ожидалось - Number)`,
enum: path => `Значение признака передачи оригинала ответа (${path}) не поддерживается`,
required: path => `Не указан признак передачи оригинала ответа (${path})`
}
@ -542,8 +551,7 @@ exports.setQueueOptionsResp = new Schema({
type: String,
required: true,
message: {
validateBuffer: path =>
`Парамметры ответа на сообщение очереди обмена (${path}) имеют некорректный тип данных (ожидалось - String)`,
validateBuffer: path => `Парамметры ответа на сообщение очереди обмена (${path}) имеют некорректный тип данных (ожидалось - String)`,
required: path => `Не указаны параметры ответа на сообщение очереди обмена (${path})`
}
}
@ -565,8 +573,7 @@ exports.setQueueAppSrvResult = new Schema({
use: { validateBuffer },
required: true,
message: {
validateBuffer: path =>
`Данные сообщения очереди обмена (${path}) имеют некорректный тип данных (ожидалось - null или Buffer)`,
validateBuffer: path => `Данные сообщения очереди обмена (${path}) имеют некорректный тип данных (ожидалось - null или Buffer)`,
required: path => `Не указаны данные сообщения очереди обмена (${path})`
}
},
@ -575,8 +582,7 @@ exports.setQueueAppSrvResult = new Schema({
use: { validateBuffer },
required: true,
message: {
validateBuffer: path =>
`Данные ответа сообщения очереди обмена (${path}) имеют некорректный тип данных (ожидалось - null или Buffer)`,
validateBuffer: path => `Данные ответа сообщения очереди обмена (${path}) имеют некорректный тип данных (ожидалось - null или Buffer)`,
required: path => `Не указаны данные ответа сообщения очереди обмена (${path})`
}
}

141
modules/parus_oracle_db.js
View File

@ -20,6 +20,7 @@ const DT_DATE = oracledb.DB_TYPE_DATE; //Тип данных "Дата" БД
const DT_CLOB = oracledb.DB_TYPE_CLOB; //Тип данных "Текстовые данные" БД
const DT_BLOB = oracledb.DB_TYPE_BLOB; //Тип данных "Двоичные данные" БД
const DT_CURSOR = oracledb.CURSOR; //Тип данных "Курсор" БД
const DT_VARCHAR_LENGTH = 32767; //Длина типа "Строка"
//------------
// Тело модуля
@ -44,7 +45,8 @@ const makeStoredPrms = (inPrms, outPrms) => {
for (i in outPrms) {
prms[i] = {
type: outPrms[i],
dir: oracledb.BIND_OUT
dir: oracledb.BIND_OUT,
...(outPrms[i] === DT_VARCHAR ? { maxSize: DT_VARCHAR_LENGTH } : {})
};
break;
}
@ -122,57 +124,85 @@ const executeStored = async prms => {
if (outResult) return outResult;
};
//Проверка допустимого количества воркеров
const checkWorkers = async prms => {
let workersData = await executeStored({
//Проверка условий при запуске сервиса интеграции
const checkAppStart = async prms => {
let res = await executeStored({
connection: prms.connection,
sName: "PKG_EXS.UTL_LIC_CLNT_COUNT_GET",
outPrms: {
LIC_CNT: DT_NUMBER
sName: "PKG_EXS.UTL_APPSRV_START_CHECK",
inPrms: {
NCONTROL_VERSION: prms.bControlSystemVersion ? 1 : 0,
SEXS_RELEASE_DATE: prms.sRelease,
NWORKERS: prms.nWorkers,
SEXSSRV: prms.sServerName,
SMODULE_NAME: prms.sModuleName
},
isFunction: true
outPrms: {
SERR_TEXT: DT_VARCHAR
},
isFunction: false
});
if (workersData.LIC_CNT === 0) {
throw new Error(`Не определено количество лицензий для приложения "ExchangeServer".`);
}
if (prms.nMaxWorkers > workersData.LIC_CNT - 1) {
throw new Error(
`Недопустимое значение параметра "Количество одновременно обрабатываемых исходящих сообщений" ("outGoing.nMaxWorkers") файла конфигурации сервиса приложений ("config.js"). Максимальное количество одновременно обрабатываемых исходящих сообщений - ${
workersData.LIC_CNT - 1
}.`
);
//Если есть ошибки запуска
if (res.SERR_TEXT) {
throw new Error(res.SERR_TEXT);
}
};
//Проверка соответствия релизов сервера приложений и системы
const checkRelease = async prms => {
let releaseData = await executeStored({
//Инициализация сервера интеграции
const initServer = async prms => {
let res = await executeStored({
connection: prms.connection,
sName: "PKG_EXS.UTL_PRODUCT_RELEASE_GET",
outPrms: {
DB_RELEASE: DT_VARCHAR
sName: "PKG_EXS.EXSSRV_INIT",
inPrms: {
SEXSSRV: prms.sServerName,
SIP: prms.sServerIP,
NWORKERS: prms.nMaxWorkers
},
isFunction: true
outPrms: {
NEXSSRV: DT_NUMBER
},
isFunction: false
});
if (releaseData.DB_RELEASE !== prms.sRelease) {
throw new Error(`Версия сервера приложений (${prms.sRelease}) не соответствует версии системы (${releaseData.DB_RELEASE}).`);
//Если рег. номер сервера интеграции не определен
if (!res.NEXSSRV) {
throw new Error(`Ошибка считывания сервера интеграции с мнемокодом "${prms.sServerName}".`);
}
//Возвращаем рег. номер сервера интеграции
return res.NEXSSRV;
};
//Очистка информации о сервере при закрытии сервиса
const clearServer = async prms => {
await executeStored({
connection: prms.connection,
sName: "PKG_EXS.EXSSRV_CLEAR",
inPrms: {
NEXSSRV: prms.nExsSrv
},
isFunction: false
});
};
//Подключение к БД
const connect = async prms => {
try {
let pool = await oracledb.createPool({
user: prms.sUser,
password: prms.sPassword,
connectString: prms.sConnectString,
//Инициализируем результат подключения
let connection = { pool: null, nExsSrv: null };
//Определяем наименование модуля сессии
let moduleName = prms.connectSettings.exsSrv.sServerName
? `${prms.connectSettings.sSessionAppName}$${prms.connectSettings.exsSrv.sServerName}`
: prms.connectSettings.sSessionAppName;
//Создаем пул подключения
connection.pool = await oracledb.createPool({
user: prms.connectSettings.sUser,
password: prms.connectSettings.sPassword,
connectString: prms.connectSettings.sConnectString,
queueTimeout: 600000,
poolMin: prms.nPoolMin ? prms.nPoolMin : 4,
poolMax: prms.nPoolMax ? prms.nPoolMax : 4,
poolIncrement: prms.nPoolIncrement ? prms.nPoolIncrement : 0,
poolMin: prms.connectSettings.nPoolMin ? prms.connectSettings.nPoolMin : 4,
poolMax: prms.connectSettings.nPoolMax ? prms.connectSettings.nPoolMax : 4,
poolIncrement: prms.connectSettings.nPoolIncrement ? prms.connectSettings.nPoolIncrement : 0,
sessionCallback: (connection, requestedTag, callback) => {
if (prms.sSessionAppName) connection.module = prms.sSessionAppName;
connection.execute(`ALTER SESSION SET CURRENT_SCHEMA=${prms.sSchema}`).then(
connection.module = moduleName;
connection.execute(`ALTER SESSION SET CURRENT_SCHEMA=${prms.connectSettings.sSchema}`).then(
r => {
callback(null, connection);
},
@ -182,11 +212,29 @@ const connect = async prms => {
);
}
});
if (prms.bControlSystemVersion) {
await checkRelease({ sRelease: prms.sRelease, connection: pool });
//Если это запуск сервиса интеграции
if (prms.bServerStart) {
//Проверяем общие условия запуска
await checkAppStart({
bControlSystemVersion: prms.connectSettings.bControlSystemVersion,
sRelease: prms.connectSettings.sRelease,
nWorkers: prms.connectSettings.nMaxWorkers,
sServerName: prms.connectSettings.exsSrv.sServerName,
sModuleName: moduleName,
connection: connection.pool
});
//Если сервис интеграции разделяется на сервера
if (prms.connectSettings.exsSrv.sServerName) {
//Инициализируем информацию о сервере интеграции
connection.nExsSrv = await initServer({
sServerName: prms.connectSettings.exsSrv.sServerName,
sServerIP: prms.connectSettings.exsSrv.sServerIP,
nMaxWorkers: prms.connectSettings.nMaxWorkers,
connection: connection.pool
});
}
await checkWorkers({ nMaxWorkers: prms.nMaxWorkers, connection: pool });
return pool;
}
return connection;
} catch (e) {
throw new Error(e.message);
}
@ -195,6 +243,12 @@ const connect = async prms => {
//Отключение от БД
const disconnect = async prms => {
try {
//Если указана информация о сервере и это закрытие сервиса интеграции
if (prms.nExsSrv && prms.bServerClose) {
//Очищаем информацию о сервере
await clearServer({ nExsSrv: prms.nExsSrv, connection: prms.connection });
}
//Отключаем от базы данных
await prms.connection.close(NPOOL_DRAIN_TIME);
return;
} catch (e) {
@ -207,6 +261,9 @@ const getServices = async prms => {
let servicesData = await executeStored({
connection: prms.connection,
sName: "PKG_EXS.SERVICES_GET",
inPrms: {
NEXSSRV: prms.nExsSrv
},
outPrms: {
RCSERVICES: DT_CURSOR
}
@ -315,7 +372,8 @@ const log = async prms => {
SMSG: prms.sMsg,
NEXSSERVICE: prms.nServiceId,
NEXSSERVICEFN: prms.nServiceFnId,
NEXSQUEUE: prms.nQueueId
NEXSQUEUE: prms.nQueueId,
NEXSSRV: prms.nExsSrv
},
outPrms: { RCLOG: DT_CURSOR }
});
@ -361,7 +419,8 @@ const getQueueOutgoing = async prms => {
connection: prms.connection,
sName: "PKG_EXS.QUEUE_SRV_TYPE_SEND_GET",
inPrms: {
NPORTION_SIZE: prms.nPortionSize
NPORTION_SIZE: prms.nPortionSize,
NEXSSRV: prms.nExsSrv
},
outPrms: { RCQUEUES: DT_CURSOR }
});

119
modules/parus_pg_db.js
View File

@ -130,58 +130,103 @@ const executeStored = async prms => {
if (outResult) return outResult;
};
//Проверка допустимого количества воркеров
const checkWorkers = async prms => {
let workersData = await executeStored({
//Проверка условий при запуске сервиса интеграции
const checkAppStart = async prms => {
let res = await executeStored({
connection: prms.connection,
sName: "PKG_EXS$UTL_LIC_CLNT_COUNT_GET",
sName: "PKG_EXS$UTL_APPSRV_START_CHECK",
inPrms: {
NCONTROL_VERSION: prms.bControlSystemVersion ? 1 : 0,
SEXS_RELEASE_DATE: prms.sRelease,
NWORKERS: prms.nWorkers,
SEXSSRV: prms.sServerName,
SMODULE_NAME: prms.sModuleName
},
outPrms: {
LIC_CNT: DT_NUMBER
SERR_TEXT: DT_VARCHAR
}
});
if (workersData.LIC_CNT === 0) {
throw new Error(`Не определено количество лицензий для приложения "ExchangeServer".`);
}
if (prms.nMaxWorkers > workersData.LIC_CNT - 1) {
throw new Error(
`Недопустимое значение параметра "Количество одновременно обрабатываемых исходящих сообщений" ("outGoing.nMaxWorkers") файла конфигурации сервиса приложений ("config.js"). Максимальное количество одновременно обрабатываемых исходящих сообщений - ${
workersData.LIC_CNT - 1
}.`
);
//Если есть ошибки запуска
if (res.SERR_TEXT) {
throw new Error(res.SERR_TEXT);
}
};
//Проверка соответствия релизов сервера приложений и системы
const checkRelease = async prms => {
let releaseData = await executeStored({
//Инициализация сервера интеграции
const initServer = async prms => {
let res = await executeStored({
connection: prms.connection,
sName: "PKG_EXS$UTL_PRODUCT_RELEASE_GET",
sName: "PKG_EXS$EXSSRV_INIT",
inPrms: {
SEXSSRV: prms.sServerName,
SIP: prms.sServerIP,
NWORKERS: prms.nMaxWorkers
},
outPrms: {
DB_RELEASE: DT_VARCHAR
NEXSSRV: DT_NUMBER
}
});
if (releaseData.DB_RELEASE !== prms.sRelease) {
throw new Error(`Версия сервера приложений (${prms.sRelease}) не соответствует версии системы (${releaseData.DB_RELEASE}).`);
//Если рег. номер сервера интеграции потерялся
if (!res.NEXSSRV) {
throw new Error(`Ошибка считывания сервера интеграции с мнемокодом "${prms.sServerName}".`);
}
//Возвращаем рег. номер сервера интеграции
return res.NEXSSRV;
};
//Очистка информации о сервере при закрытии сервиса
const clearServer = async prms => {
await executeStored({
connection: prms.connection,
sName: "PKG_EXS$EXSSRV_CLEAR",
inPrms: {
NEXSSRV: prms.nExsSrv
}
});
};
//Подключение к БД
const connect = async prms => {
try {
let pool = new pg.Pool({
connectionString: `postgres://${prms.sUser}:${prms.sPassword}@${prms.sConnectString}`,
//Инициализируем результат подключения
let connection = { pool: null, nExsSrv: null };
//Определяем наименование модуля сессии
let moduleName = prms.connectSettings.exsSrv.sServerName
? `${prms.connectSettings.sSessionAppName}$${prms.connectSettings.exsSrv.sServerName}`
: prms.connectSettings.sSessionAppName;
//Создаем пул подключения
connection.pool = new pg.Pool({
connectionString: `postgres://${prms.connectSettings.sUser}:${prms.connectSettings.sPassword}@${prms.connectSettings.sConnectString}`,
connectionTimeoutMillis: 600000,
min: prms.nPoolMin ? prms.nPoolMin : 4,
max: prms.nPoolMax ? prms.nPoolMax : 4
min: prms.connectSettings.nPoolMin ? prms.connectSettings.nPoolMin : 4,
max: prms.connectSettings.nPoolMax ? prms.connectSettings.nPoolMax : 4
});
pool.on("acquire", async client => {
await client.query(`select ALTER_SESSION_SET_SCHEMA($1);`, [prms.sSchema]);
await client.query(`select ALTER_SESSION_SET_SCHEMA($1);`, [prms.connectSettings.sSchema]);
});
//Если это запуск сервиса интеграции
if (prms.bServerStart) {
//Проверяем общие условия запуска
await checkAppStart({
bControlSystemVersion: prms.connectSettings.bControlSystemVersion,
sRelease: prms.connectSettings.sRelease,
nWorkers: prms.connectSettings.nMaxWorkers,
sServerName: prms.connectSettings.exsSrv.sServerName,
sModuleName: moduleName,
connection: connection.pool
});
//Если сервис интеграции разделяется на сервера
if (prms.connectSettings.exsSrv.sServerName) {
//Инициализируем информацию о сервере интеграции
connection.nExsSrv = await initServer({
sServerName: prms.connectSettings.exsSrv.sServerName,
sServerIP: prms.connectSettings.exsSrv.sServerIP,
nMaxWorkers: prms.connectSettings.nMaxWorkers,
connection: connection.pool
});
if (prms.bControlSystemVersion) {
await checkRelease({ sRelease: prms.sRelease, connection: pool });
}
await checkWorkers({ nMaxWorkers: prms.nMaxWorkers, connection: pool });
return pool;
}
return connection;
} catch (e) {
throw new Error(e.message);
}
@ -190,6 +235,11 @@ const connect = async prms => {
//Отключение от БД
const disconnect = async prms => {
try {
//Если указана информация о сервере и это закрытие сервиса интеграции
if (prms.nExsSrv && prms.bServerClose) {
//Очищаем информацию о сервере
await clearServer({ nExsSrv: prms.nExsSrv, connection: prms.connection });
}
await prms.connection.end();
} catch (e) {
throw new Error(e.message);
@ -201,6 +251,9 @@ const getServices = async prms => {
let servicesData = await executeStored({
connection: prms.connection,
sName: "PKG_EXS$SERVICES_GET",
inPrms: {
NEXSSRV: prms.nExsSrv
},
outPrms: {
RCSERVICES: DT_CURSOR
}
@ -308,7 +361,8 @@ const log = async prms => {
SMSG: prms.sMsg,
NEXSSERVICE: prms.nServiceId,
NEXSSERVICEFN: prms.nServiceFnId,
NEXSQUEUE: prms.nQueueId
NEXSQUEUE: prms.nQueueId,
NEXSSRV: prms.nExsSrv
},
outPrms: { RCLOG: DT_CURSOR }
});
@ -354,7 +408,8 @@ const getQueueOutgoing = async prms => {
connection: prms.connection,
sName: "PKG_EXS$QUEUE_SRV_TYPE_SEND_GET",
inPrms: {
NPORTION_SIZE: prms.nPortionSize
NPORTION_SIZE: prms.nPortionSize,
NEXSSRV: prms.nExsSrv
},
outPrms: { RCQUEUES: DT_CURSOR }
});