Система событий

Серверные события связывают изменение доменной сущности с дополнительной реакцией: проверкой, пересчётом, автоматизацией процесса, сбросом кеша или публикацией сообщения во внешнюю шину. Центральная точка выполнения — ru.bgcrm.event.EventProcessor.

Эта система не является очередью фоновых задач. По умолчанию обработчики выполняются синхронно внутри вызывающей операции и влияют на её результат и транзакцию.

Серверные события ru.bgcrm.event.Event и клиентские события ru.bgcrm.event.client.ClientEvent решают разные задачи. Первые вызывают Java/Kotlin-обработчики на сервере. Вторые передают команды интерфейсу пользователя через JSON-response или Redis.

Основные типы

Тип Назначение

Event

Базовый Serializable-контракт. Каждое событие предоставляет timestamp.

AbstractEvent

Базовая реализация с временем создания и идентификатором ERP-инстанса.

UserEvent

Событие пользовательского действия. Хранит DynActionForm, через неё предоставляет пользователя и позволяет обработчику добавить данные или клиентские события в HTTP-response.

EventListener<E>

Обработчик события типа E. Метод notify(event, connectionSet) может изменить данные в текущей транзакции или отклонить операцию исключением.

DynamicEventListener

Базовый класс обработчика из динамического Java-кода. Для декларативной диспетчеризации служит AnnotatedEventListener.

EventProcessor

Хранит локальные подписки, вызывает системные и динамические обработчики, записывает телеметрию и передаёт поддерживаемые события в MessageBus.

События обычно группируются по моменту выполнения. Например, ProcessChangingEvent и ParamChangingEvent создаются до изменения, а ProcessChangedEvent и ParamChangedEvent — после него. Это соглашение конкретного прикладного API, а не автоматическое поведение EventProcessor: место вызова и состав полей определяет код, который создаёт событие.

Порядок обработки

Обычный EventProcessor.processEvent(…​) выполняет следующие шаги:

  1. Использует текущий ServerContext. Если контекста нет, создаёт транзакционный контекст пользователя из UserEvent либо системного пользователя.

  2. Находит подписчиков по точному runtime-классу события и последовательно вызывает их в порядке регистрации.

  3. Если передано имя динамического класса, создаёт его экземпляр и вызывает после системных подписчиков.

  4. Передаёт событие в MessageBus, если не отключена обработка системных listeners.

  5. Возвращает управление вызывающему коду. Commit выполняет владелец ServerContext, а не EventProcessor.

Подписка на базовый класс не получает события его наследников. Ключом реестра является точный результат event.getClass(). Если нужны несколько типов, подпишите listener на каждый из них. Параметр includeSupertypes у @OnEvent относится к внутренней диспетчеризации AnnotatedEventListener и не меняет реестр EventProcessor.

Если listener выбросил BGMessageException, BGException или runtime-исключение, выполнение прекращается: следующие обработчики и MessageBus не вызываются. Исключение поднимается владельцу операции; в HTTP-запросе это обычно означает отсутствие commit и ответ с ошибкой.

Порядок регистрации сохраняется внутри списка одного класса события, но отдельного API приоритетов нет. Не стройте бизнес-логику на том, что два независимых listener всегда будут зарегистрированы в определённой последовательности. Если шаги обязаны выполняться в строгом порядке, объедините их в одном обработчике или выразите порядок в сервисе.

Варианты вызова

Вызов Поведение

processEvent(event)

Использует master-соединение текущего ServerContext; без контекста создаёт новый и подтверждает его после успешной обработки.

processEvent(event, connectionSet)

Вызывает системных подписчиков и MessageBus, используя переданный набор соединений.

processEvent(event, className, connectionSet)

После системных подписчиков вызывает динамический класс className, затем MessageBus.

processEvent(event, className, connectionSet, false)

Вызывает только указанный динамический класс. Системные подписчики и MessageBus пропускаются.

processEventWithoutConnection(event)

Передаёт listener null вместо ConnectionSet. Подходит только обработчикам, которые гарантированно не обращаются к БД и не требуют журналирования через EventProcessorLogDAO.

Новый код внутри сервиса должен использовать соединения текущего контекста. Для совместимости со старым контрактом передавайте ServerContextConnectionSet либо используйте EventProcessorWrapper.

val context = ServerContext.get()
EventProcessor.processEvent(
    ExampleChangedEvent(entityId),
    ServerContextConnectionSet(context)
)

Регистрация системных обработчиков

Подписка хранит переданный экземпляр listener в памяти JVM:

class ExampleChangedListener : EventListener<ExampleChangedEvent> {
    override fun notify(event: ExampleChangedEvent, conSet: ConnectionSet) {
        ServerContext.getService<ExampleService>().recalculate(event.entityId)
    }
}

EventProcessor.subscribe(ExampleChangedListener(), ExampleChangedEvent::class.java)

Ядро создаёт системные listeners в KernelSystemListeners. Плагины регистрируют свои обработчики при инициализации. Динамические классы из настройки createOnStart создаются при старте; административная операция полной перекомпиляции вызывает их регистрацию повторно.

subscribe не проверяет дубли. Повторное создание регистратора приводит к повторному вызову одного и того же поведения. Если класс может инициализироваться повторно, сначала удалите старые экземпляры:

EventProcessor.unsubscribe(ExampleChangedListener::class.java.name)
EventProcessor.subscribe(ExampleChangedListener(), ExampleChangedEvent::class.java)

unsubscribe удаляет из всех списков каждый listener с указанным полным именем класса.

Зарегистрированный экземпляр может одновременно обслуживать события из разных HTTP-потоков. Системный listener должен быть stateless или потокобезопасным. Не храните событие, ConnectionSet, ServerContext, service/repository или промежуточный результат в полях экземпляра.

Динамические обработчики

Имя динамического listener обычно приходит из настройки типа процесса, параметра или другой конфигурируемой сущности. Если имя передано в processEvent, для вызова создаётся новый экземпляр через DynamicClassManager. DynamicEventListener, зарегистрированный обычным subscribe, остаётся общим экземпляром и подчиняется требованиям потокобезопасности системного listener.

В обычном режиме DynamicEventListener выполняется во внутреннем executor, а вызывающий поток синхронно ждёт результат. Максимальное время ожидания определяется в следующем порядке:

  1. @EventProcessingTimeout(timeout = …​) на классе listener;

  2. настройка event.processTimeout;

  3. устаревшая настройка dynamicEventListenerTimeOut;

  4. 1000 мс по умолчанию.

При подключённом debugger (jdwp) динамический listener выполняется прямо в вызывающем потоке без timeout, чтобы breakpoint не приводил к ошибке ожидания.

Timeout ограничивает только ожидание. Текущая реализация не вызывает Future.cancel(…​), поэтому скрипт может продолжить работу после того, как вызывающая операция уже получила ошибку и начала закрывать контекст. Динамический обработчик обязан укладываться в короткий лимит и не должен запускать долгий HTTP-вызов, ожидание или цикл. Переносите такую работу в AsyncWorker через submitAfterCommit.

AnnotatedEventListener упрощает один динамический обработчик с большим числом поддерживаемых событий: методы выбираются по аннотациям @OnEvent, @OnProcessCreated, @OnStatusChanged, @OnParamChanged и другим. Полный каталог и правила сигнатур приведены на отдельной странице.

Транзакция и побочные эффекты

Listener участвует в текущей операции:

  • видит неподтверждённые изменения через тот же master connection;

  • может добавить другие изменения в общую транзакцию;

  • может остановить операцию исключением;

  • не должен самостоятельно подтверждать или закрывать соединения контекста.

Событие с суффиксом Changed означает «основное изменение выполнено в текущей транзакции», а не «изменение уже подтверждено». Внешняя система, slave БД и другой ERP-инстанс могут ещё не видеть результат.

Долгий или внешний побочный эффект запускайте после commit:

val user = ServerContext.get().user
val entityId = event.entityId

AsyncWorker.submitAfterCommit("example_sync", user) {
    syncEntity(entityId)
}

Так listener не удерживает HTTP-запрос и JDBC-соединение во время внешнего вызова, а задача не стартует при откате исходной транзакции. Ограничения фоновых задач и надёжные сценарии описаны в разделе про внешние сервисы.

MessageBus

После локальных обработчиков EventProcessor передаёт событие в MessageBus. Шина включается настройкой message_bus.enabled; message_bus.async выбирает синхронную или асинхронную публикацию.

Во внешнее сообщение преобразуются только события, для которых зарегистрирован MessageBuilder. Сейчас это:

  • ParamChangedEvent;

  • ProcessChangedEvent;

  • LinkAddedEvent;

  • LinkRemovedEvent.

Остальные типы MessageBus игнорирует. При синхронном режиме ошибка Kafka поднимается вызывающему коду; при асинхронном — записывается в лог фонового dispatcher.

Вызов MessageBus.dispatch выполняется до commit прикладной транзакции и не использует transactional outbox. В синхронном режиме Kafka-публикация происходит сразу; в асинхронном фактическая отправка конкурирует с завершением транзакции. Возможны оба расхождения: сообщение уже опубликовано, а БД затем откатилась; либо БД подтвердилась, но сообщение не было доставлено. Для интеграции, где потеря или ложная публикация недопустимы, сохраняйте устойчивое задание в той же транзакции и отправляйте его отдельным обработчиком с идемпотентностью и retry.

Ошибки, журнал и телеметрия

Каждый вызов listener публикует метрики:

Метрика Содержание

event_processor_count_total

Количество завершённых вызовов; теги event, class, status.

event_processor_duration_milliseconds

Длительность одного listener с теми же тегами.

Для успеха status равен Successful, для ошибки — failure. Полный каталог находится в справочнике метрик.

Подробный журнал в помесячных таблицах event_processor_log_YYYYMM отключён по умолчанию и управляется настройками:

Настройка Что записывать

logSuccessfulEventsThrow

Успешные вызовы.

logErrorEventsThrow

Вызовы, завершившиеся ошибкой.

logAllEventsThrow

Все вызовы независимо от результата.

Запись содержит класс события, класс listener, connection id, длительность, результат и instance_host_name. Несмотря на имя колонки, текущая реализация записывает туда системное свойство JVM user.name. Журнал создаётся через отдельное соединение и подтверждается независимо от прикладной транзакции. Срок хранения задаёт event_processor_log_live для планировщика очистки логов.

Аннотация @AdditionalLogging включает отдельный logger класса, если глобально установлено enableAdditionalLogging=1.

Добавление нового события

  1. Создайте небольшой неизменяемый класс события. Передавайте идентификаторы и данные, необходимые обработчикам; не помещайте в событие открытые JDBC-ресурсы или service/repository.

  2. Выберите точку публикации и определите её семантику: до изменения, после изменения внутри транзакции или после commit.

  3. Зарегистрируйте listener один раз при инициализации ядра или плагина.

  4. Проверьте успешный сценарий, исключение listener, повторный вызов и конкурентную обработку.

  5. Если реакция внешняя или долгая, запускайте её через AsyncWorker.submitAfterCommit и обеспечьте идемпотентность.

class ExampleChangedEvent(
    val entityId: Int,
    val previousTitle: String,
    val currentTitle: String
) : AbstractEvent()

Не создавайте новый тип, если существующее событие уже означает тот же завершённый факт и содержит достаточные данные. Не переиспользуйте существующее событие только из-за похожего набора полей, если момент или бизнес-смысл отличаются.

Типичные ошибки

  • ожидать, что подписка на Event или другой supertype получит все дочерние события;

  • регистрировать один listener несколько раз при reload;

  • хранить состояние текущего события в singleton-listener;

  • считать Changed гарантией commit;

  • выполнять долгий внешний вызов внутри listener;

  • рассчитывать, что timeout остановил динамический скрипт;

  • отправлять в MessageBus событие, для которого нет MessageBuilder;

  • смешивать серверные и клиентские события;

  • закрывать или подтверждать переданный ConnectionSet внутри listener.