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

Merge pull request 'master' (#1) from CITKParus/P8-ExchangeService:master into master

Browse Source 
Reviewed-on: Dollerok/P8-ExchangeService#1
This commit is contained in:
Dollerok 2024-09-27 13:54:29 +03:00
commit 86a19155c6
2 changed files with 6 additions and 10 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
core/in_queue.js
View File

@ -107,7 +107,7 @@ class InQueue extends EventEmitter {
//Оповестим подписчиков об останове //Оповестим подписчиков об останове
this.emit(SEVT_IN_QUEUE_STOPPED); this.emit(SEVT_IN_QUEUE_STOPPED);
} }
//Обработка сообщения //Обработка сообщения HTTP/HTTPS
async processMessage(prms) { async processMessage(prms) {
//Проверяем структуру переданного объекта для обработки //Проверяем структуру переданного объекта для обработки
let sCheckResult = validateObject(prms, prmsInQueueSchema.processMessage, "Параметры функции обработки входящего сообщения"); let sCheckResult = validateObject(prms, prmsInQueueSchema.processMessage, "Параметры функции обработки входящего сообщения");
@ -382,7 +382,6 @@ class InQueue extends EventEmitter {
throw new ServerError(SERR_OBJECT_BAD_INTERFACE, sCheckResult); throw new ServerError(SERR_OBJECT_BAD_INTERFACE, sCheckResult);
} }
} }
//Обработка MQ сообщения //Обработка MQ сообщения
async processMQMessage({ message, service, fn, sProtocol }) { async processMQMessage({ message, service, fn, sProtocol }) {
//Буфер для сообщения очереди //Буфер для сообщения очереди
@ -528,7 +527,6 @@ class InQueue extends EventEmitter {
} }
} }
} }
//Запуск обработки очереди входящих сообщений //Запуск обработки очереди входящих сообщений
async startProcessing(prms) { async startProcessing(prms) {
//Проверяем структуру переданного объекта для старта //Проверяем структуру переданного объекта для старта
@ -703,7 +701,6 @@ class InQueue extends EventEmitter {
throw new ServerError(SERR_OBJECT_BAD_INTERFACE, sCheckResult); throw new ServerError(SERR_OBJECT_BAD_INTERFACE, sCheckResult);
} }
} }
//Закрытие подключений //Закрытие подключений
stopConnections() { stopConnections() {
//Если у нас есть соединения с MQTT //Если у нас есть соединения с MQTT
@ -729,7 +726,6 @@ class InQueue extends EventEmitter {
} }
} }
} }
//Остановка обработки очереди исходящих сообщений //Остановка обработки очереди исходящих сообщений
stopProcessing() { stopProcessing() {
//Выставляем флаг неработы //Выставляем флаг неработы

10
core/out_queue_processor.js
View File

@ -171,7 +171,7 @@ const appProcess = async prms => {
//Указываем, что выполнение обработчика "После" невозможно //Указываем, что выполнение обработчика "После" невозможно
bExecuteAfter = false; bExecuteAfter = false;
break; break;
//mqtt и mqtts //MQTT/MQTTS
case [objServiceSchema.SPROTOCOL_MQTT, objServiceSchema.SPROTOCOL_MQTTS].includes(sProtocol): case [objServiceSchema.SPROTOCOL_MQTT, objServiceSchema.SPROTOCOL_MQTTS].includes(sProtocol):
options.url = prms.service.sSrvRoot; options.url = prms.service.sSrvRoot;
options.body = prms.queue.blMsg; options.body = prms.queue.blMsg;
@ -189,7 +189,7 @@ const appProcess = async prms => {
//Указываем, что выполнение обработчика "После" невозможно //Указываем, что выполнение обработчика "После" невозможно
bExecuteAfter = false; bExecuteAfter = false;
break; break;
//Другие //HTTP/HTTPS
default: default:
//Определимся с URL и телом сообщения в зависимости от способа передачи параметров (для POST, PATCH и PUT - данные в теле, для остальных - в URI) //Определимся с URL и телом сообщения в зависимости от способа передачи параметров (для POST, PATCH и PUT - данные в теле, для остальных - в URI)
if ( if (
@ -344,7 +344,7 @@ const appProcess = async prms => {
message: options.body message: options.body
}); });
break; break;
//mqtt и mqtts //MQTT/MQTTS
case [objServiceSchema.SPROTOCOL_MQTT, objServiceSchema.SPROTOCOL_MQTTS].includes(sProtocol): case [objServiceSchema.SPROTOCOL_MQTT, objServiceSchema.SPROTOCOL_MQTTS].includes(sProtocol):
serverResp = await publishMQTT({ serverResp = await publishMQTT({
settings: options.settings, settings: options.settings,
@ -354,9 +354,9 @@ const appProcess = async prms => {
message: options.body message: options.body
}); });
break; break;
//Другие //HTTP/HTTPS
default: default:
//Ждем ответ от удалённого сервера //Установим флаг возврата полного ответа (и тела и заголовков)
options.resolveWithFullResponse = true; options.resolveWithFullResponse = true;
//Отправляем запрос //Отправляем запрос
serverResp = await rqp(options); serverResp = await rqp(options);