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

ЦИТК-901 - Поддержка MQTT/Kafka - косметические исправления

Browse Source
This commit is contained in:
Mikhail Chechnev 2024-09-27 13:21:14 +03:00
parent 9b25f10fb4
commit 67aafc25b7
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);
}
//Обработка сообщения
//Обработка сообщения HTTP/HTTPS
async processMessage(prms) {
//Проверяем структуру переданного объекта для обработки
let sCheckResult = validateObject(prms, prmsInQueueSchema.processMessage, "Параметры функции обработки входящего сообщения");
@ -382,7 +382,6 @@ class InQueue extends EventEmitter {
throw new ServerError(SERR_OBJECT_BAD_INTERFACE, sCheckResult);
}
}
//Обработка MQ сообщения
async processMQMessage({ message, service, fn, sProtocol }) {
//Буфер для сообщения очереди
@ -528,7 +527,6 @@ class InQueue extends EventEmitter {
}
}
}
//Запуск обработки очереди входящих сообщений
async startProcessing(prms) {
//Проверяем структуру переданного объекта для старта
@ -703,7 +701,6 @@ class InQueue extends EventEmitter {
throw new ServerError(SERR_OBJECT_BAD_INTERFACE, sCheckResult);
}
}
//Закрытие подключений
stopConnections() {
//Если у нас есть соединения с MQTT
@ -729,7 +726,6 @@ class InQueue extends EventEmitter {
}
}
}
//Остановка обработки очереди исходящих сообщений
stopProcessing() {
//Выставляем флаг неработы

10
core/out_queue_processor.js
View File

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