EventLocal
библиотека для организации работы микросервисов для маленьких проектов требующих работать на разнесённых физически серверах.
Первый взгляд на проблему
Требовалось отделить бизнес логику от реализации запросов через такие средства коммуникации как брокер сообщений, телеграм месседжер, база данных и RestApi. Идея состояла в том, что в бизнес логике создавался бы контроллер. Его методы помечались TS декораторами, которые привязывали исполнение этого метода к конкретному виду реализации.
Для примера обработка сообщений:
Импортируем библиотеки и создадим класс контроллера:
import { EventHeandlerMethod, MessageHeandlerMethod, API } from "./EventCore";
export class TemperatureController {
@EventHeandlerMethod("temperature.change")
public static UpdateTemperature(s: string) {
console.log(s);
}
}
В нашем случае мы подписались на доменные события поступающие через RabbitMQ в канале temperature.change.
Наш класс контроллера ничего не знает о реализации запросов к нему. в этот класс мы можем поместить вызов основной бизнес логики.
Если нам необходимо сменить тип вызова на RestApi то мы просто заменим декоратор
@API('put','/v1/temperature')
public static UpdateTemperature(res, req) {
console.log(s);
}
В данном примере взаимодействие с параметрами res и req будет производиться также, как и с библиотекой Express.
Бывает так, что иногда необходимо взаимодействовать с сервисом через такой канал связи как TelegramBot. Для этого необходимо указать настройки в /.env
@MessageHeandlerMethod(/^Привет,\s(.+)/i)
public static OtherCommand(s: string) {
console.log(s);
}
Внимание! При работе с этим каналом нужно иметь в виду, что можно создавать только один экземпляр взаимодействия с API телеграм.
В качестве входного параметра будет первая скобка регулярного выражения. А если вызвать процедуру return [string]. То данное сообщение будет отправлено обратно в чат текущего отправившего пользователя. В случае отправки return [boolean] будет такой же эффект, но на значение true ответ будет "исполнено", а на значение false - "неудача"
Если функция ничего не возвращает, то данная функция будет возвращать пользователю сообщение "Принято".
Отправка событий
Для обратного взаимодействия предусмотрено несколько методов библиотеки EventLocal
- eventLocal.sendEvent(quaue: string, message: string) - для брокера сообщений.
- eventLocal.sendMessage(chatId: number, message: string) - для телеграм сообщений.
- eventLocal.sendApi(host, port, path) - для get запросов.
При запуске эта библиотека будет пытаться зарегистрироваться в сервисе регистрации. Обычно я его запускаю на порте 1989.
Протокол регистрации
Пример:
-
Отправка запроса на адрес 192.168.1.4:1989
-
В запросе должно быть тело с информацией о версии сервиса, порт, хост, и название для вызова сервиса по нему.
-
Получить в ответ статус 200
-
Если ответ не 200 повторить попытку через 5 секунд
-
Далее сервис регистрации будет проверять точку жизни приложения /vX/health, где X это версия, на предмет ответа статуса 200 каждые 3 секунды.
-
Если точка не будет опрошена, сервис будет пытаться зарегистрироваться снова.
Хранилище данных
По умолчанию создаётся база данных aux_event и далее в ней три таблицы: events, entities, snapshots. Как описано в книге Микросервисы - Криса Ричардсона. Взаимодействие с хранилищем должно происходить через класс репозитория. Он может быть наследован и реализовывать интерфейс:
export class AggregateRepository {
save() {
//
};
find() {
//
};
update() {
//
};
}
Конфигурирование и начало работы
Для конфигурирования требуется создать файл .env и указать его при запуске программы:
node -r ./.env ./FILE_NAME
Конфигурирование производится силами пакета dotenv
- SERVICE_NAME='caht' - Имя сервиса для регистрации
- JWT_TOKEN_KEY='KEY' - Ключ для проверки токенов JWT (если не указан, то не использует проверку прозрачных токенов)
- TELEGRAM_ENABLE=1 - включить поддержку телеграмм API, если не указан, то недоступен
- TELEGRAM_PROXY_HOST - IP адрес для прокси телеграма
- TELEGRAM_PROXY_PORT - порт для прокси
- TELEGRAM_API - апи полученный телеграмм ботом
- TELEGRAM_COMMAND_CHAT_ID=465739287 - номер чата, для телеграмм сообщений
- SERVICE_PORT=8088 - порт, на котором будет запущен сервис
- SERVICE_VERSION_API=1 - этот указатель требуется для регистрации сервиса
- BROKER_HOST='{LOGIN}:{PASSWORD}@localhost:5672' - Настройки брокера сообщений. Обратите внимание, что при работе внутри сети доступ без пароля будет заблокирован, ознакомьтесь с правами и конфигурированием в RebbitMQ
Требования к ключу безопасности
Доступный алгоритм HS256
Полезная нагрузка:
- user_id - номер пользователя
- mask - битовая маска прав.
Работа с доменной моделью
Всвязи с тем, что в данной библиотеке используется событийная модель, к доменной модели выдвигаются следующие требования. Каждый доменный класс должен наследовать ReflectionMutableCommandProccessingAggregate. Создаются 2 метода, process(command: Command) - для создания событий и apply(domenEvent: DomenEvent) для их применения.
Метод process вызывает репозиторий при создании, поиске и обновлении данных. После того как события будут созданы, они будут применены к классу через метод apply.
В конструктор репозитория добавляется пустой экземпляр домена. Это необходимо, что бы абстрагировать репозиторий от заполнения очевидных данных, из экземпляра класса берётся конструктор для создания прототипов, а также имя класса для заполнения полей событий и команд.
Класс сделует расширять дополнительной надстройкой, которая будет иметь свою таблицу в базе данных с нужными полями.
Команды
Команды должны быть названы глаголами, указывающими на действие, которое должно быть сделано.
export default class TestCommand extends Command {
}
export class CreateTest extends TestCommand {
constructor(testDetail: TestDetail) {
super();
this.version = "1";
this.entity_id = testDetail.id;
this.entity = testDetail;
}
}
События
События должны быть названы глаголом в страдательном залоге в прошедшем впемени. События отражают уже совершённые действия с сущностями.
export class TestEvent extends DomenEvent {}
export class TestCreated extends TestEvent {}
Домен
export class Test implements ReflectionMutableCommandProccessingAggregate {
...
apply(event: TestEvent): void {
switch (event.event_type) {
case "TestCreated":
this.id = event.entity_id;
this.name = event.event_data.name;
return;
case ...
}
}
process(command: TestCommand): TestEvent[] {
if (command instanceof CreateTest) {
command.entity.id = command.entity_id;
return [
new TestCreated({
event_type: "TestCreated",
event_data: command.entity,
entity_type: "Test",
entity_id: command.entity_id,
triggering_event: null
})
];
}
...
}
Сервис
export default class TestService {
constructor(private repository: AggregateRepository<Test>) {}
public async createTest(testDetail: TestDetail) {
return await this.repository.save(new CreateTest(testDetail));
}
public async renameTest(id: string, name: string) {
return await this.repository.update(new RenameTest(id, name));
}
public async find(id) {
return await this.repository.find(id);
}
}