Определение проблемы
В проекте Lakey используется модуль Vello внутренней системы обмена сообщениями Prologue для управления внешними событиями и их сохранения, чтобы сделать их доступными в контексте LLM.
Событийный преобразователь 'TypeAwareEventConverter', предоставляемый Prologue по умолчанию, считывает значение 'payloadClass' из заголовка при получении сообщения и выполняет десериализацию после проверки, существует ли имя класса в класспасе запущенной службы.
Этот метод работает без проблем как внутри одного сервиса, так и при обмене событиями в зависимости от модуля события внешнего сервиса, поскольку класс события уже существует в класспасе службы. Однако с не определёнными и опубликованными событиями ситуация иная. Класс неопределённого события отсутствует в класспасе Lakey, что приводит к возникновению исключения, и, в конечном итоге, событие не обрабатывается должным образом и теряется.
Поскольку логика внутри Lakey также изменяется каждый раз, когда схема события внешнего сервиса изменяется, возникает сильная связь между службами. Мы думали о том, как сохранить события без зависимости от модуля события внешнего сервиса и реализовали 'LakeyEventConverter' для решения этой проблемы.
Существующий способ работы EventConverter
1. Преобразует полезную нагрузку сообщения в строку с заданной кодировкой.
2. Если значение 'payloadClass' отсутствует, происходит десериализация в тип 'Map'.
3. Если значение 'payloadClass' присутствует, класс загружается с помощью 'Class.forName()', после чего происходит десериализация в соответствующий тип.
4. Если загрузка класса завершается неудачей, возникнет исключение 'EventConversionFailed'.
Решение: LakeyEventConverter
Чтобы решить проблему 4, мы унаследовали 'TypeAwareEventConverter' и переопределили метод 'convert()'. Это позволяет сохранять события, в которых произошла ошибка преобразования, в интерпретируемом сыром виде.
public class LakeyEventConverter extends TypeAwareEventConverter {
private final String encoding;
public LakeyEventConverter(
@Value("${vizend.prologue.vello.encoding:UTF-8}") String encoding) {
super(encoding);
this.encoding = encoding;
}
@Override
public Object convert(VelloMessage message) {
try {
Object object = super.convert(message);
if (object instanceof LakeyOutgoingDomainEvent out) {
return out.toIncomingEvent();
}
return object;
} catch (EventConvertException e) {
String payloadJson = new String(message.getPayload(), Charset.forName(encoding));
Map<String, Object> payload = JsonUtil.fromJson(payloadJson, Map.class);
String fqcn = message.getPayloadClass();
String simpleClassName = fqcn.substring(fqcn.lastIndexOf('.') + 1);
return UnreferencedEvent.builder()
.eventType(simpleClassName)
.headers(message.getHeaders())
.payload(payload)
.messageId(message.getId())
.subject(message.getSubject())
.receivedAt(LocalDateTime.now())
.build();
}
}
}Обработка 'LakeyEventConverter' делится на два случая.
1. В случае успешного преобразования
Если преобразование родительским конвертером ('TypeAwareEventConverter') прошло успешно, но дело касается собственных событий Lakey или событийных классов, от которых мы зависим, возвращается преобразованный объект без дополнительной постобработки.
2. Если произошла ошибка преобразования типа, как и при внешнем событии
Если родительский класс выбрасывает 'EventConvertException', это будет перехвачено и обработано отдельно. В этом случае полезная нагрузка десериализуется в 'Map<String, Object>', и из FQCN, содержащегося в заголовке, извлекается только простое имя класса, которое используется как 'eventType'. Затем на его основе создается и возвращается объект 'UnreferencedEvent'.
При этом сохраняется не только полезная нагрузка, но и следующая информация.
* Идентификатор сообщения, тема получения (subject), заголовок сообщения, время получения, исходная полезная нагрузка
UnreferencedEvent
UnreferencedEvent - это объект, содержащий события, для которых нет класса ссылки. Как следует из названия, Lakey не может узнать тип, но это событие всё же имеет ценность для сбора.
@Getter
@Builder
public class UnreferencedEvent {
private String eventType;
private Map<String, String> headers;
private Map<String, Object> payload;
private LocalDateTime receivedAt;
private String messageId;
private String subject;
}Здесь важным значением является не только полезная нагрузка. messageId необходим для обработки дубликатов, а subject используется для проверки темы, из которой пришло событие. headers и receivedAt помогут позже отследить источник события и время его получения.
От UnreferencedEvent до RawEvent
Когда LakeyEventConverter возвращает UnreferencedEvent, обработчик UnreferencedEventHandler с аннотацией @EventHandler из Prologue получает это событие. Обработчик не проводит никакого бизнес-анализа и передает его в RawEventFlow.register().
@Component
@RequiredArgsConstructor
public class UnreferencedEventHandler {
private final RawEventFlow rawEventFlow;
@EventHandler
public void handleUnreferencedEvent(UnreferencedEvent unreferencedEvent) {
if (unreferencedEvent == null) {
log.warn("UnreferencedEvent is null");
return;
}
rawEventFlow.register(unreferencedEvent);
}
}В RawEventFlow.register() проверяются три условия.
public void register(UnreferencedEvent unreferencedEvent) {
if (!rawEventAction.isTargetEvent(unreferencedEvent)) {
return;
}
RawEventCdo rawEventCdo = rawEventAction.convertToRawEventCdo(unreferencedEvent);
if (rawEventLogic.existsByEventId(rawEventCdo.getEventId())) {
return;
}
String rawEventId = rawEventLogic.registerRawEvent(rawEventCdo);
eventProxy.produceEvent(RawEventProjectionRequestedEvent.newInstance(rawEventId));
}1. Проверка целевой темы
Не все события, приходящие извне, сохраняются. Обрабатываются только темы, входящие в targetTopics, установленные в LakeyConfigProperties. На начальном этапе фильтруются события, не предназначенные для сбора, чтобы уменьшить ненужное сохранение и последующую обработку.
2. Проверка на дублирование события
messageId используется как eventId для проверки, было ли уже сохранено это событие. Если то же событие приходит снова, оно сохраняется только один раз.
3. Сохранение в RawEvent
Если это целевая тема и нет дублирования, мы преобразуем UnreferencedEvent в RawEventCdo и сохраняем. После сохранения мы публикуем RawEventProjectionRequestedEvent. Фактическая предобработка данных происходит на этапе обработки этого события.
Завершение
Причина, по которой была создана эта структура, ясна. Это нужно, чтобы не потерять само входящее событие, даже если тип события заранее неизвестен.
Существующий 'TypeAwareEventConverter' выбрасывал исключение и прекращал обработку при встрече с отсутствующим в класспасе типом. В отличие от него, 'LakeyEventConverter' сохраняет исключенное событие в общественной сущности под названием 'UnreferencedEvent'. В результате Lakey смог собирать события, не полагаясь напрямую на классы событий внешних служб.
Если при обработке событий необходимо ослабить связь между службами, буду признателен за ваше внимание.
Ted