Working with Events 

Events — an inter-service communication mechanism in the Citeck platform that allows microservices to react to changes in the system without direct dependencies on each other.

Events are generated when the state of objects changes: when records are created, updated, or deleted, when the status changes, attributes are modified, tasks are assigned, and other actions. Subscribers receive only the attributes they need — without modifying the event source.

The mechanism supports two delivery modes:

Non-transactional — delivery via RabbitMQ after the transaction completes (loose coupling between services);
Transactional — synchronous delivery within the current transaction (allows interrupting the transaction or performing actions “here and now”).

Events 2.0 

Event reactions are built on RabbitMQ and event models in the ecos-events library. RabbitMQ connections are established on both the producer and consumer sides. Example: sending notifications in response to an event — status changed, attribute modified, task assigned, etc.

Architecture 

../../_images/notification_through_events_2.0.png

How It Works 

Events in Citeck allow changing the set of attributes required by an event subscriber without modifying the event source. At system startup, all subscribers register in Zookeeper the list of event types they need and the event attributes they are interested in.

The application that can emit events of this type sees that there are subscribers to these events in the system, and when events occur, it prepares the required list of attributes and sends them to RabbitMQ (for non-transactional listeners) or directly to the listener (for transactional listeners) via a synchronous request.

Attributes are described in the Records API format and can take full advantage of that API.

Transactional Listeners 

Transactional listeners allow reacting to events “here and now” without waiting for the transaction to complete. Such listeners can be used to add system state validation with the ability to abort the transaction, or for any other actions that must be performed within the transaction.

An important consideration: by adding transactional listeners, you automatically introduce a hard dependency of event-generating microservices on the microservice containing the listener. That is, if the microservice with the listener is unavailable, the event-generating microservice will not be able to function fully.

Examples 

Subscribing to an event with arbitrary attributes (Kotlin)

eventsService.addListener<DataValue> {
    withTransactional(true)
    withEventType(RecordCreatedEvent.TYPE)
    withAction {
        println("Event for record with type: ${it["type"].asText()} Display name: ${it["disp"].asText()}")
    }
    withDataClass(DataValue::class.java)
    withAttributes(mapOf("type" to "rec._type?id", "disp" to "rec?disp"))
}

Subscribing to an event with arbitrary attributes (Java):

Map<String, String> attributes = new HashMap<>();
attributes.put("type", "rec._type?id");
attributes.put("disp", "rec?disp");

eventsService.<DataValue>addListener(b -> {
    b.withTransactional(true);
    b.withEventType(RecordCreatedEvent.TYPE);
    b.withActionJ((event) -> {
        System.out.printf(
            "Event for record with type: %s Display name: %s%n",
            event.get("type").asText(),
            event.get("disp").asText()
        );
    });
    b.withDataClass(DataValue.class);
    b.withAttributes(attributes);
    return Unit.INSTANCE;
});

Example of an event listener for a specific data type 

import org.springframework.stereotype.Component
import ru.citeck.ecos.events2.EventsService
import ru.citeck.ecos.events2.type.RecordChangedEvent
import ru.citeck.ecos.events2.type.RecordCreatedEvent
import ru.citeck.ecos.events2.type.RecordDeletedEvent
import ru.citeck.ecos.records2.RecordRef
import ru.citeck.ecos.records2.predicate.model.Predicates.eq
import ru.citeck.ecos.records3.record.atts.schema.annotation.AttName
import java.time.Instant
import javax.annotation.PostConstruct

@Component
class SomeTypeEventsListener(
    private val eventsService: EventsService,
) {

    companion object {
        private const val YOUR_TYPE = "ID вашего типа данных"
    }

    @PostConstruct
    fun init() {
        eventsService.addListener<RecordUpdated> {
            withTransactional(true)
            withEventType(RecordChangedEvent.TYPE)
            withDataClass(RecordUpdated::class.java)
            withFilter(eq("typeDef.id", YOUR_TYPE))
            withAction { event ->
                println("Запись была обновлена: " + event.record + ", создал: " + event.user + ", время: " + event.time)
//                Ваша логика при событии Обновления записи.....
            }
        }

        eventsService.addListener<RecordCreated> {
            withTransactional(true)
            withEventType(RecordCreatedEvent.TYPE)
            withDataClass(RecordCreated::class.java)
            withFilter(eq("typeDef.id", YOUR_TYPE))
            withAction { event ->
                println("Создана новая запись: " + event.record + ", создал: " + event.user + ", время: " + event.time)
//                Ваша логика при событии Создания записи.....
            }
        }

        eventsService.addListener<RecordDeleted> {
            withTransactional(true)
            withEventType(RecordDeletedEvent.TYPE)
            withDataClass(RecordDeleted::class.java)
            withFilter(eq("typeDef.id", YOUR_TYPE))
            withAction { event ->
                println("Запись была удалена: " + event.record + ", удалил: " + event.user + ", время: " + event.time)
//                Ваша логика при событии Удаления записи.....
            }
        }

//        И еще много других Listener-ов для уже реализованных эвентов или ваших собственных
//        Например для RecordStatusChangedEvent, RecordDraftStatusChangedEvent, RecordContentChangedEvent и тд.
    }

//    В data классах определяем набор необходимых нам данных, которые хотим достать из Event-а.
//    Можно ознакомиться с классом RecordEventTypes.kt из библиотеки ecos-events2 для более подробного понимания какие данные можно получить

    data class RecordUpdated(
        @AttName("record?id")
        val record: RecordRef,
        @AttName("\$event.time")
        val time: Instant,
        @AttName("\$event.user")
        val user: String,
    )

    data class RecordCreated(
        @AttName("record?id")
        val record: RecordRef,
        @AttName("\$event.time")
        val time: Instant,
        @AttName("\$event.user")
        val user: String,
    )

    data class RecordDeleted(
        @AttName("record?id")
        val record: RecordRef,
        @AttName("\$event.time")
        val time: Instant,
        @AttName("\$event.user")
        val user: String
    )

}

Notes: