forked from CITKParus/P8-ExchangeService
Обработка входящих: вынес обработку сообщения в отдельную функцию, настроил сохранения тела входящего сообщения в БД, настроил индивидуальные и общий обработчики ошибок
This commit is contained in:
Mikhail Chechnev
parent
ff7dee27d1
commit
52320550da
119
core/in_queue.js
119
core/in_queue.js
@ -10,10 +10,13 @@
|
||||
const _ = require("lodash"); //Работа с массивами и коллекциями
|
||||
const EventEmitter = require("events"); //Обработчик пользовательских событий
|
||||
const express = require("express"); //WEB-сервер Express
|
||||
const bodyParser = require("body-parser"); //Модуль для Express (разбор тела входящего запроса)
|
||||
const { ServerError } = require("./server_errors"); //Типовая ошибка
|
||||
const { SERR_OBJECT_BAD_INTERFACE } = require("./constants"); //Общесистемные константы
|
||||
const { makeErrorText, validateObject } = require("./utils"); //Вспомогательные функции
|
||||
const { SERR_OBJECT_BAD_INTERFACE, SERR_WEB_SERVER } = require("./constants"); //Общесистемные константы
|
||||
const { makeErrorText, validateObject, buildURL } = require("./utils"); //Вспомогательные функции
|
||||
const { NINC_EXEC_CNT_YES } = require("../models/prms_db_connector"); //Схемы валидации параметров функций модуля взаимодействия с БД
|
||||
const objServiceSchema = require("../models/obj_service"); //Схемы валидации сервиса
|
||||
const objServiceFnSchema = require("../models/obj_service_function"); //Схемы валидации функции сервиса
|
||||
const objQueueSchema = require("../models/obj_queue"); //Схемы валидации сообщения очереди
|
||||
const prmsInQueueSchema = require("../models/prms_in_queue"); //Схемы валидации параметров функций класса
|
||||
|
||||
@ -60,13 +63,55 @@ class InQueue extends EventEmitter {
|
||||
//Уведомление о запуске обработчика очереди
|
||||
notifyStarted() {
|
||||
//Оповестим подписчиков о запуске
|
||||
this.emit(SEVT_IN_QUEUE_STARTED);
|
||||
this.emit(SEVT_IN_QUEUE_STARTED, this.inComing.nPort);
|
||||
}
|
||||
//Уведомление об остановке обработчика очереди
|
||||
notifyStopped() {
|
||||
//Оповестим подписчиков об останове
|
||||
this.emit(SEVT_IN_QUEUE_STOPPED);
|
||||
}
|
||||
//Обработка сообщения
|
||||
async processMessage(prms) {
|
||||
//Проверяем структуру переданного объекта для старта
|
||||
let sCheckResult = validateObject(
|
||||
prms,
|
||||
prmsInQueueSchema.processMessage,
|
||||
"Параметры функции обработки входящего сообщения"
|
||||
);
|
||||
//Если структура объекта в норме
|
||||
if (!sCheckResult) {
|
||||
//Определимся с телом сообщения
|
||||
let blMsg = null;
|
||||
//Для POST сообщений - это тело запроса
|
||||
if (prms.function.nFnPrmsType == objServiceFnSchema.NFN_PRMS_TYPE_POST) {
|
||||
blMsg = prms.req.body && !_.isEmpty(prms.req.body) ? prms.req.body : null;
|
||||
} else {
|
||||
//Для GET - параметры запроса
|
||||
if (!_.isEmpty(prms.req.query)) blMsg = new Buffer(JSON.stringify(prms.req.query));
|
||||
}
|
||||
//Кладём сообщение в очередь
|
||||
let q = await this.dbConn.putQueue({
|
||||
nServiceFnId: prms.function.nId,
|
||||
blMsg
|
||||
});
|
||||
//Скажем что пришло новое входящее сообщение
|
||||
await this.logger.info(
|
||||
`Новое входящее сообщение от ${prms.req.connection.address().address} для фукнции ${
|
||||
prms.function.sCode
|
||||
} (${buildURL({ sSrvRoot: prms.service.sSrvRoot, sFnURL: prms.function.sFnURL })})`,
|
||||
{ nQueueId: q.nId }
|
||||
);
|
||||
prms.res
|
||||
.status(200)
|
||||
.send(
|
||||
`<html><body><center><br><h1>Сервер приложений ПП Пурс 8</h1><h3>Функция сервиса: ${
|
||||
prms.service.sName
|
||||
}/${prms.function.sCode}</h3></center></body></html>`
|
||||
);
|
||||
} else {
|
||||
throw new ServerError(SERR_OBJECT_BAD_INTERFACE, sCheckResult);
|
||||
}
|
||||
}
|
||||
//Запуск обработки очереди входящих сообщений
|
||||
startProcessing(prms) {
|
||||
//Проверяем структуру переданного объекта для старта
|
||||
@ -81,15 +126,74 @@ class InQueue extends EventEmitter {
|
||||
this.bWorking = true;
|
||||
//запоминаем список обслуживаемых сервисов
|
||||
this.services = prms.services;
|
||||
//Запускаем сервер
|
||||
this.webApp.use("*", (req, res) => {
|
||||
res.status(200).send("<html><body><center><h1>Сервер приложений ПП Пурс 8</h1></center></body></html>");
|
||||
//Конфигурируем сервер - обработка тела сообщения
|
||||
this.webApp.use(bodyParser.raw({ limit: `${this.inComing.nMsgMaxSize}mb`, type: "*/*" }));
|
||||
//Конфигурируем сервер - обходим все сервисы, работающие на приём сообщений
|
||||
_.forEach(_.filter(this.services, { nSrvType: objServiceSchema.NSRV_TYPE_RECIVE }), srvs => {
|
||||
//Для любых запросов к корневому адресу сервиса - ответ о том, что это за сервис, и что он работает
|
||||
this.webApp.all(srvs.sSrvRoot, (req, res) => {
|
||||
res.status(200).send(
|
||||
`<html><body><center><br><h1>Сервер приложений ПП Пурс 8</h1><h3>Сервис: ${
|
||||
srvs.sName
|
||||
}</h3></center></body></html>`
|
||||
);
|
||||
});
|
||||
//!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!! КОНТРОЛЬ ЗАПУСКА!!!!!!!!!!!!!!
|
||||
//Для всех функций сервиса...
|
||||
_.forEach(srvs.functions, fn => {
|
||||
//...собственный обработчик, в зависимости от указанного способа передачи параметров
|
||||
this.webApp[fn.nFnPrmsType == objServiceFnSchema.NFN_PRMS_TYPE_POST ? "post" : "get"](
|
||||
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(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)));
|
||||
}
|
||||
);
|
||||
});
|
||||
});
|
||||
//Запросы на адреса, не входящие в состав объявленных сервисов - 404 NOT FOUND
|
||||
this.webApp.use("*", (req, res) => {
|
||||
res.status(404).send(
|
||||
"<html><body><center><br><h1>Сервер приложений ПП Пурс 8</h1><h3>Запрошенный адрес не найден</h3></center></body></html>"
|
||||
);
|
||||
});
|
||||
//Ошибки, не отработанные индивидуальными обработчиками - 500 SERVER ERROR
|
||||
this.webApp.use(async (err, req, res, next) => {
|
||||
//Протоколируем в журнал работы сервера
|
||||
await this.logger.error(makeErrorText(new ServerError(SERR_WEB_SERVER, err.message)));
|
||||
//Отправим ошибку клиенту
|
||||
res.status(500).send(makeErrorText(new ServerError(SERR_WEB_SERVER, err.message)));
|
||||
});
|
||||
//Запускаем сервер
|
||||
this.srv = this.webApp.listen(this.inComing.nPort, () => {
|
||||
//И оповещаем всех что запустились
|
||||
this.notifyStarted();
|
||||
});
|
||||
this.srv.on("error", e => {
|
||||
throw new ServerError(e.code, `Фатальная ошибка обработчика очереди входящих сообщений: ${e.message}`);
|
||||
});
|
||||
} else {
|
||||
throw new ServerError(SERR_OBJECT_BAD_INTERFACE, sCheckResult);
|
||||
}
|
||||
@ -99,7 +203,6 @@ class InQueue extends EventEmitter {
|
||||
//Выставляем флаг неработы
|
||||
this.bWorking = false;
|
||||
//Останавливаем WEB-сервер (если создавался)
|
||||
//!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!! КОНТРОЛЬ ОСТАНОВА!!!!!!!!!!!!!!
|
||||
if (this.srv) {
|
||||
this.srv.close(() => {
|
||||
//Оповещаем всхес, что остановились
|
||||
|
||||
@ -8,10 +8,13 @@
|
||||
//----------------------
|
||||
|
||||
const Schema = require("validate"); //Схемы валидации
|
||||
const { IncomingMessage, ServerResponse } = require("http"); //Работа с HTTP протоколом
|
||||
const { inComing } = require("./obj_config"); //Схемы валидации конфигурации сервера приложений
|
||||
const { defServices } = require("./obj_services"); //Схема валидации списка сервисов
|
||||
const { DBConnector } = require("../core/db_connector"); //Класс взаимодействия в БД
|
||||
const { Logger } = require("../core/logger"); //Класс для протоколирования работы
|
||||
const { Service } = require("./obj_service"); //Схема валидации сервиса
|
||||
const { ServiceFunction } = require("./obj_service_function"); //Схема валидации функции сервиса
|
||||
|
||||
//------------------
|
||||
// Интерфейс модуля
|
||||
@ -49,6 +52,46 @@ exports.InQueue = new Schema({
|
||||
}
|
||||
});
|
||||
|
||||
//Схема валидации параметров функции обработки входящего сообщения
|
||||
exports.processMessage = new Schema({
|
||||
//Входящее сообщение
|
||||
req: {
|
||||
type: IncomingMessage,
|
||||
required: true,
|
||||
message: {
|
||||
type: path =>
|
||||
`Входящее сообщение (${path}) имеет некорректный тип данных (ожидалось - IncomingMessage, см. документацию к Node.JS HTTP - https://nodejs.org/dist/latest-v10.x/docs/api/http.html#http_class_http_incomingmessage)`,
|
||||
required: path => `Не указано входящее сообщение (${path})`
|
||||
}
|
||||
},
|
||||
//Ответ на входящее сообщение
|
||||
res: {
|
||||
type: ServerResponse,
|
||||
required: true,
|
||||
message: {
|
||||
type: path =>
|
||||
`Ответ на входящие сообщение (${path}) имеет некорректный тип данных (ожидалось - ServerResponse, см. документацию к Node.JS HTTP - https://nodejs.org/dist/latest-v10.x/docs/api/http.html#http_class_http_serverresponse)`,
|
||||
required: path => `Не указан ответ на входящее сообщение (${path})`
|
||||
}
|
||||
},
|
||||
//Cервис-обработчик
|
||||
service: {
|
||||
schema: Service,
|
||||
required: true,
|
||||
message: {
|
||||
required: path => `Не указан сервис для обработки входящего сообщения (${path})`
|
||||
}
|
||||
},
|
||||
//Функция сервиса-обработчика
|
||||
function: {
|
||||
schema: ServiceFunction,
|
||||
required: true,
|
||||
message: {
|
||||
required: path => `Не указана функция сервиса для обработки входящего сообщения (${path})`
|
||||
}
|
||||
}
|
||||
});
|
||||
|
||||
//Схема валидации параметров функции запуска обслуживания очереди
|
||||
exports.startProcessing = new Schema({
|
||||
//Список обслуживаемых сервисов
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user