Event Flow is a lightweight Java framework for building event-driven applications. It provides the structural backbone for publishing, routing, and processing events — so you can focus on business logic instead of wiring infrastructure.
- Features
- Architecture
- Modules
- Installation
- Quick Start
- Core Components
- Lifecycle Tracking
- Usage Examples
- Configuration
- Interaction Diagrams
- Flexible Routing — Event channels with configurable transports
- Multiple Transports — LocalQueue (in-JVM) and Apache Kafka out of the box, with extension points for custom transports
- Annotation-Based — Event handling via
@EventListener - Interface-Based — Event handling via
EventSubscriberinterface - POJO/Record Events — Support for plain Java objects without
Eventinterface - Idempotency — Event deduplication based on UID
- Transactional Publishing — Send events after transaction commit
- Structured Logging — Decorators for publisher and dispatcher with machine-parseable JSON output
- Retry Mechanism — Exponential backoff with configurable parameters
- Lifecycle Tracking — End-to-end event lifecycle with persistent storage, status tracking, automatic retry of failed events, and acknowledgment-based monitoring
- Extensible Serialization — JSON and MessagePack with support for custom formats
┌─────────────────────────────────────────────────────────────────────────┐
│ Event Flow Architecture │
├─────────────────────────────────────────────────────────────────────────┤
│ │
│ ┌──────────────┐ ┌─────────────────────────────────────────┐ │
│ │ Service │────────▶│ Event Channels │ │
│ │ (Publisher) │ │ ┌─────────────┐ ┌─────────────────┐ │ │
│ └──────────────┘ │ │ Internal │ │ External │ │ │
│ │ │ (in-JVM) │ │ (Kafka) │ │ │
│ │ └──────┬──────┘ └────────┬────────┘ │ │
│ │ │ │ │ │
│ └─────────┼───────────────── ┼────────────┘ │
│ │ │ │
│ ┌─────────▼───────────────── ▼────────────┐ │
│ │ Outgoing Event Transports │ │
│ │ ┌─────────────┐ ┌─────────────────┐ │ │
│ │ │ LocalQueue │ │ Kafka │ │ │
│ │ │ Queue │ │ Producer │ │ │
│ │ └─────────────┘ └─────────────────┘ │ │
│ └─────────────────────────────────────────┘ │
│ │
└─────────────────────────────────────────────────────────────────────────┘
│
│ External Communication
│ (Network / Message Broker)
▼
┌─────────────────────────────────────────────────────────────────────────┐
│ External Event Flow (Kafka) │
├─────────────────────────────────────────────────────────────────────────┤
│ │
│ ┌─────────────────────────────────────────┐ │
│ │ Incoming Event Transports │ │
│ │ ┌─────────────┐ ┌─────────────────┐ │ │
│ │ │ LocalQueue │ │ Kafka │ │ │
│ │ │ Queue │ │ Consumer │ │ │
│ │ └─────────────┘ └─────────────────┘ │ │
│ └─────────┬─────────────────┬─────────────┘ │
│ │ │ │
│ ┌─────────▼─────────────────▼────────────┐ │
│ │ Event Dispatcher │ │
│ │ │ │
│ └─────────┬─────────────────┬────────────┘ │
│ │ │ │
│ ┌─────────▼─────────────────▼────────────┐ │
│ │ Handler Registry │ │
│ │ ┌─────────────┐ ┌─────────────────┐ │ │
│ │ │ Annotation │ │ Interface │ │ │
│ │ │ Based │ │ Based │ │ │
│ │ └─────────────┘ └─────────────────┘ │ │
│ └─────────┬─────────────────┬────────────┘ │
│ │ │ │
│ ┌─────────▼─────────────────▼────────────┐ │
│ │ Event Handlers │ │
│ │ (@EventListener / EventSubscriber) │ │
│ └────────────────────────────────────────┘ │
│ │
└─────────────────────────────────────────────────────────────────────────┘
┌────────────────────────────────────────────────────────────────────────────┐
│ Microservice A (Event Producer) │
│ │
│ ┌──────────┐ ┌───────────┐ ┌──────────┐ ┌──────────────────────┐ │
│ │ Service │──▶│ Publisher │──▶│ Channel │──▶│ KafkaOutTransport │ │
│ └──────────┘ └───────────┘ └──────────┘ └──────────┬───────────┘ │
└────────────────────────────────────────────────────────────┼───────────────┘
│ produce()
▼
═══════════════════════════════════════════════════
Apache Kafka — Topic: "events"
═══════════════════════════════════════════════════
┃ ┃
consume() ┃ consume() ┃
│ ┃ │ ┃
▼ ┃ ▼ ┃
┌─────────────────────────────────────┐ ┌─────────────────────────────────────┐
│ Microservice B (Consumer) │ │ Microservice C (Consumer) │
│ │ │ │
│ ┌──────────────────┐ │ │ ┌──────────────────┐ │
│ │ KafkaInTransport │ │ │ │ KafkaInTransport │ │
│ └──────────────────┘ │ │ └──────────────────┘ │
│ │ │ │ │ │
│ ┌────────▼─────────┐ │ │ ┌────────▼─────────┐ │
│ │ EventDispatcher │ │ │ │ EventDispatcher │ │
│ └────────┬─────────┘ │ │ └────────┬─────────┘ │
│ │ dispatch() │ │ │ dispatch() │
│ │ │ │ │ │
│ ┌────────▼─────────┐ │ │ ┌────────▼────────┐ │
│ │ HandlerRegistry │ │ │ │ HandlerRegistry │ │
│ └────────┬─────────┘ │ │ └────────┬────────┘ │
│ │ getHandlers() │ │ │ getHandlers() │
│ ▼ │ │ ▼ │
│ ┌──────────────────┐ │ │ ┌──────────────────┐ │
│ │ Handler 1 │ │ │ │ Handler 2 │ │
│ │ Handler 2 │ │ │ │ Handler 3 │ │
│ └──────────────────┘ │ │ └──────────────────┘ │
└─────────────────────────────────────┘ └─────────────────────────────────────┘
┌─────────────────────────────────────────────────────────────────────────────┐
│ Single Application │
│ │
│ ┌──────────┐ ┌───────────┐ ┌──────────┐ ┌──────────────────────┐ │
│ │ Service │──▶│ Publisher │──▶│ Channel │──▶│LocalQueueOutTransport│ │
│ └──────────┘ └───────────┘ └──────────┘ └──────────┬───────────┘ │
│ │ offer(event) │
│ ▼ │
│ ┌───────────────────────────────────────────────────────┐ │
│ │ BlockingDeque<Event> │ │
│ │ (shared queue) │ │
│ └─┬─────────────────────────────────────────────────────┘ │
│ │ take(event) │
│ ▼ │
│ ┌──────────────────┐ ┌──────────────────┐ ┌──────────────────────┐ │
│ │ LocalQueueInTrans│──▶│ EventDispatcher │──▶│ EventHandlerRegistry │ │
│ └──────────────────┘ └────────┬─────────┘ └──────────┬───────────┘ │
│ │ dispatch() │ getHandlers() │
│ │ ▼ │
│ │ ┌──────────────────────┐ │
│ └─────────────▶│ Handlers │ │
│ async execute │ (virtual threads) │ │
│ └──────────────────────┘ │
└─────────────────────────────────────────────────────────────────────────────┘
| Module | Description | Documentation |
|---|---|---|
| event-flow-core | Core module — framework-agnostic, pure Java 21+ | README |
| event-flow-spring | Spring Boot auto-configuration with YAML | README |
Add the dependency to your pom.xml:
<dependency>
<groupId>io.github.vovten</groupId>
<artifactId>event-flow</artifactId>
<version>1.1.0</version>
</dependency>For Spring Boot integration:
<dependency>
<groupId>io.github.vovten</groupId>
<artifactId>event-flow-spring</artifactId>
<version>1.1.0</version>
</dependency>implementation 'io.github.vovten:event-flow:1.1.0'
// For Spring Boot:
implementation 'io.github.vovten:event-flow-spring:1.1.0'- Java 21+
- Apache Kafka 3.6.0+ (optional, for external events)
Events can be defined in two ways:
a) Use the @Event annotation (recommended) — cleaner POJO/record, channels from annotation:
@Event(channels = InternalEventChannel.class)
public record OrderShipped(String orderId, String customerId) {}b) Implement the Event interface — full control over type and channels:
public record OrderCreatedEvent(String orderId, String customerId) implements Event {
@Override
public Class<?> type() {
return OrderCreatedEvent.class;
}
}Both approaches work with
publish(),prepare(), and handler registration. Recommended:@Eventannotation for cleaner code; useimplements Eventonly when you need full control.
// Create local queue transport for in-JVM event delivery
var transports = new LocalQueueTransportsBuilder()
.queueSize(1000)
.build();
// Create a channel for internal (in-application) events
// For external events (Kafka), use ExternalEventChannel with Kafka transports
EventChannel internalChannel = new InternalEventChannel(
List.of(transports.publisher())
);
// Create publisher (add externalChannel here for external events)
EventPublisher eventPublisher = EventPublisherBuilder.create(internalChannel)
.retryable(3, Duration.ofMillis(100), 2.0)
.buildAndLog();
// Create handler registry
EventHandlerRegistry handlerRegistry = EventHandlerRegistryBuilder.create()
.withAnnotationListeners()
.withInterfaceListeners()
.buildAndLog();
// Create dispatcher
EventDispatcher eventDispatcher = EventDispatcherBuilder.create()
.executor(Executors.newVirtualThreadPerTaskExecutor())
.handlerRegistry(handlerRegistry)
.transports(List.of(transports.dispatcher()))
.concurrencyLimit(100)
.buildAndLog();
// Start the dispatcher
eventDispatcher.start(eventDispatcher::dispatch);Handle the event directly:
public class OrderEventHandler {
@EventListener
public void handleOrderCreated(OrderCreatedEvent event) {
System.out.println("Order created: " + event.orderId());
}
}
// Register the handler
handlerRegistry.register(new OrderEventHandler());Or receive the full Envelope with metadata (eventId, processId, occurredAt, etc.):
public class OrderEnvelopeHandler {
@EventListener(OrderShipped.class)
public void handleOrderShipped(Envelope<OrderShipped> envelope) {
OrderShipped event = envelope.payload();
System.out.println("Order " + event.orderId()
+ " processed with id " + envelope.eventId());
}
}
// Register the envelope handler
handlerRegistry.register(new OrderEnvelopeHandler());When the handler parameter is
Envelope<T>, the@EventListenerannotation must specify the payload type explicitly (e.g.,@EventListener(OrderShipped.class)). This is especially useful for POJO/record events annotated with@Event— they are automatically wrapped in anEnvelopeupon publishing.
public class OrderService {
private final EventPublisher eventPublisher;
public OrderService(EventPublisher eventPublisher) {
this.eventPublisher = eventPublisher;
}
public void createOrder(String customerId) {
// Business logic...
eventPublisher.publish(new OrderCreatedEvent("order-123", customerId));
}
}Base interface for all events. Defines event type and publication channels.
public interface Event {
Class<?> type();
default List<Class<? extends EventChannel>> channels() {
return List.of(InternalEventChannel.class);
}
default String asJson() {
return EventUtils.toJson(this);
}
}TraceableEvent — extends Event with tracing fields: eventId (UUID), processId (correlation), occurredAt (timestamp).
Wrapper for domain events that adds technical metadata. Automatically captures:
eventId(UUID) — unique event identifierprocessId(UUID) — correlation ID (e.g., saga ID)occurredAt(Instant) — event timestampmetadata(Map) — custom key-value pairspayload— the actual domain object
The envelope implements Event interface, so it passes through existing transport infrastructure.
Creating Envelopes:
// Auto-generated metadata
Envelope<OrderCreatedEvent> envelope = Envelope.of(new OrderCreatedEvent("123"));
// With custom processId (correlation)
UUID processId = UUID.fromString("...");
Envelope<OrderCreatedEvent> envelope = Envelope.of(new OrderCreatedEvent("123"), processId);
// With explicit channels
Envelope<OrderCreatedEvent> envelope = Envelope.of(
new OrderCreatedEvent("123"),
List.of(ExternalEventChannel.class)
);Channels from @Event Annotation:
POJO/record classes can use the @Event annotation to specify default channels:
@Event(channels = ExternalEventChannel.class)
public record OrderCreatedEvent(String orderId) {}A channel defines event delivery routes through transports.
public interface EventChannel {
String name();
List<OutTransport> transports();
CompletableFuture<SendResults> send(Event event);
}Built-in Channels:
InternalEventChannel— for internal in-application deliveryExternalEventChannel— for external delivery to other applications/microservices
Publishes events to configured channels.
Creating via Builder:
EventPublisher publisher = EventPublisherBuilder.create(internalChannel, externalChannel)
.retryable(3, Duration.ofMillis(100), 2.0)
.build();EventPublisherBuilder — fluent builder for creating publishers with flexible configuration:
| Method | Description |
|---|---|
create(...) |
Create builder with event channels (required) |
retryable() |
Enable retry with default settings (3 attempts, 100ms initial delay, 2.0 multiplier) |
retryable(max, delay, multiplier) |
Enable retry with custom settings |
withDecorator(fn) |
Add custom decorator to the publisher chain |
build() |
Build the publisher |
buildAndLog() |
Build the publisher and log the configuration |
loggable() |
Enable structured logging (JSON, 1024 char payload limit) |
loggable(maxPayloadLength) |
Enable structured logging with custom payload truncation |
loggable(maxPayloadLength, excludedEvents) |
Enable structured logging with event type exclusion |
loggable(maxPayloadLength, excludedEvents, logLevels) |
Enable structured logging with per-event log level overrides |
Delivers events from transports to handlers.
public interface EventDispatcher {
void dispatch(Event event);
void register(Object listener);
boolean isRegistered(Object listener);
void start(Consumer<Event> handler);
void stop();
}EventDispatcherBuilder — fluent builder for creating dispatchers:
| Method | Description |
|---|---|
executor(...) |
Configure ExecutorService (required) |
handlerRegistry(...) |
Handler registry (required) |
transports(...) |
List of incoming transports |
concurrencyLimit(n) |
Concurrency limiting via Semaphore |
idempotent() |
Enable idempotency (deduplication by UID) |
idempotent(maxSize, ttl) |
Enable idempotency with custom settings |
withDecorator(fn) |
Add custom decorator |
build() |
Build the dispatcher |
buildAndLog() |
Build the dispatcher and log the configuration |
loggable() |
Enable structured logging (JSON, 1024 char payload limit) |
loggable(maxPayloadLength) |
Enable structured logging with custom payload truncation |
loggable(maxPayloadLength, excludedEvents) |
Enable structured logging with event type exclusion |
loggable(maxPayloadLength, excludedEvents, logLevels) |
Enable structured logging with per-event log level overrides |
Registry for discovering and managing event handlers.
public interface EventHandlerRegistry {
List<EventHandler> getHandlers(Event event);
void register(Object listener);
void unregister(Object listener);
boolean isRegistered(Object listener);
void merge(EventHandlerRegistry registry);
int handlerCount();
String name();
}Built-in implementations:
EventListenerRegistry— discovery via@EventListenerannotationEventSubscriberRegistry— discovery viaEventSubscriberinterfaceCompositeEventHandlerRegistry— combines multiple registries
EventHandlerRegistryBuilder — fluent builder for creating registries:
| Method | Description |
|---|---|
withAnnotationListeners() |
Enable discovery via @EventListener annotation |
withInterfaceListeners() |
Enable discovery via EventSubscriber interface |
withCustomRegistry(registry) |
Add a custom registry |
withDecorator(fn) |
Add a decorator |
build() |
Build the registry |
buildAndLog() |
Build the registry and log the configuration |
An enum that controls how an event's journey is tracked. Three levels:
| Level | Behaviour |
|---|---|
NONE |
Fire-and-forget — event passes through without any persistence |
PERSISTED |
Event is stored in the EventStore (with UNDEFINED status) but not actively tracked |
MANAGED |
Full lifecycle tracking — event status transitions through NEW → PUBLISHED → HANDLED, with automatic retry on failure |
Set via @Event(lifecycle = ...) annotation or the lifecycle() default method on Event.
Persistence layer for lifecycle tracking. Stores serialised events and tracks their status as they flow through the system.
JdbcEventStore— production-grade, backed by a relational database (PostgreSQL, H2, MySQL, etc.)InMemoryEventStore— in-JVMConcurrentHashMap-backed store for testing and single-JVM scenarios
Decorator for EventPublisher that persists events to the EventStore before they are sent and updates their status (NEW → PUBLISHED/FAILED) after publishing. See the Lifecycle Tracking section.
Decorator for EventDispatcher that publishes SuccessAck or FailureAck events back to the source channels after handler execution — enabling end-to-end status tracking. See the Lifecycle Tracking section.
Periodically scans the EventStore for failed (FAILED), stuck (PUBLISHED), and orphaned (NEW) events and re-publishes them with exponential backoff. See the Lifecycle Tracking section.
An EventSubscriber that processes incoming lifecycle acknowledgment events (SuccessAck/FailureAck) and updates the event status in the EventStore. Filters by service name to allow multiple publisher instances to share the same channel. See the Lifecycle Tracking section.
Transports for event delivery.
Incoming Transports (InTransport):
LocalQueueInTransport— receive from local queueKafkaInTransport— receive from Kafka topics
Outgoing Transports (OutTransport):
LocalQueueOutTransport— send to local queueKafkaOutTransport— send to Kafka topicBroadcastKafkaOutTransport— send to all Kafka topic partitions
Utility for creating paired incoming/outgoing transports based on a local queue:
var transports = new LocalQueueTransportsBuilder()
.queueSize(1000)
.build();
EventChannel channel = new InternalEventChannel(
List.of(transports.publisher())
);
EventDispatcher dispatcher = EventDispatcherBuilder.create()
.executor(Executors.newVirtualThreadPerTaskExecutor())
.handlerRegistry(handlerRegistry)
.transports(List.of(transports.dispatcher()))
.build();
dispatcher.start(dispatcher::dispatch);EventSerializer — serialization interface with magic byte prefix:
0x01— JSON0x02— MessagePack
EventSerializerFactory — factory with automatic format detection:
// Register a custom serializer
EventSerializerFactory.getInstance().register(new MyCustomSerializer());EventTypeRegistry — security whitelist for allowed event classes:
// Allow a package (default: io.github.vovten.eventflow.*)
EventTypeRegistry.allowPackage("com.example.events");
// Allow a specific class
EventTypeRegistry.allowClass(MyEvent.class);Event Flow provides an end-to-end lifecycle tracking system that answers three questions about every event:
- Was it saved? — the event persisted before publication (crash recovery)
- Was it delivered? — the event reached all target channels successfully
- Was it handled? — all registered handlers processed the event without errors
This turns event publishing from a fire-and-forget operation into a reliable, observable process — essential for critical business events (orders, payments, notifications) where you need guarantees and visibility.
| Scenario | Without lifecycle | With lifecycle (MANAGED) |
|---|---|---|
| Service crashes mid-publish | Event lost | Event safe in store, retried on restart |
| Handler throws an exception | Silent failure | Status updated to FAILED, automatic retry |
| Ack lost in transit | Event stuck in limbo | Detected as PUBLISHED → retried |
| Debugging production issues | Logs only | Queryable event store with full history |
You choose the level per event — NONE for high-throughput fire-and-forget, PERSISTED for audit without monitoring, MANAGED when you need guarantees.
┌─────────────────────────── PUBLISHER ────────────────────────────┐
│ │
│ Event ──► EventLifecyclePublisher ──► EventStore ──► Channel │
│ │ │ │
│ │ saves & tracks status │ │
│ │ (NEW→PUBLISHED/FAILED) │ │
│ │ │ │
│ ◄── EventRetryScheduler scans FAILED, PUBLISHED, NEW ──► retry │
│ │
└──────────────────────────────────────────────────────────────────┘
│ ▲
│ SuccessAck / FailureAck │
▼ │
┌────────────────────────── DISPATCHER ────────────────────────────┐
│ │
│ Channel ──► EventLifecycleDispatcher ──► Handlers │
│ │ │
│ │ publishes ack after handler execution │
│ │
│ ◄── AckHandler processes ack, updates store (HANDLED/FAILED) │
│ │
└──────────────────────────────────────────────────────────────────┘
The flow:
EventLifecyclePublishersaves the event toEventStorewith statusNEW, then publishes it- On successful delivery, status updates to
PUBLISHED; on failure, toFAILED EventLifecycleDispatcher(on the consumer side) publishesSuccessAckorFailureAckback through the channelAckHandler(on the publisher side) catches the ack and updates status toHANDLEDorFAILEDEventRetrySchedulerperiodically rescansFAILED, stuckPUBLISHED, and orphanedNEWevents and retries them with exponential backoff
| Level | Behaviour | Use case |
|---|---|---|
NONE |
No persistence, no tracking. Event passes straight through | High-throughput notifications, ephemeral events |
PERSISTED |
Event saved to store but status stays UNDEFINED. No retry, no ack |
Audit trail without operational guarantees |
MANAGED |
Full tracking: NEW → PUBLISHED → HANDLED, automatic retry on failure, ack-based end-to-end confirmation |
Orders, payments, critical business events |
Set via @Event annotation:
@Event(channels = InternalEventChannel.class, lifecycle = EventLifecycle.MANAGED)
public record OrderCreated(String orderId) {}Or via Event.lifecycle() default method — the annotation takes precedence when both are present.
Enable lifecycle tracking in Spring Boot with three properties:
event-flow:
publisher:
lifecycle:
enabled: true # Enable lifecycle-aware publishing
service-name: order-service # Required! Identifies this service for ack filtering
store:
type: db # "db" (PostgreSQL/H2/MySQL) or "in-memory" (testing)
table-name: event_store
auto-init-schema: true # Disable in production, manage DDL via Flyway
retry:
enabled: true
max-retries: 3
retry-interval: 30s
min-age: 30s # Backoff: 30s, then 60s, then 120s
dispatcher:
lifecycle:
enabled: true # Enable ack generation on the dispatcher sideThe service-name is mandatory — it identifies this service instance so that AckHandler only processes acknowledgments meant for it (multiple services can share the same ack channel without interference).
EventRetryScheduler scans for three categories of events that need retry:
| Status | Why it happens |
|---|---|
FAILED |
Publication or handler threw an error |
PUBLISHED |
Event was sent but ack was lost in transit (stuck) |
NEW |
Event was saved but the application crashed before publishing finished (orphaned) |
Backoff is exponential: delay = minAge × 2^retryCount. First retry at 30s, second at 60s, third at 120s.
EventCleanupScheduler periodically deletes old terminal events (HANDLED and UNDEFINED) from the EventStore to prevent unbounded growth. Deletion is performed in configurable batches with pauses between batches to reduce database load.
event-flow:
publisher:
lifecycle:
cleanup:
enabled: true
max-age: 7d # Events older than this are deleted
batch-size: 500 # Rows per DELETE
interval: 60m # How often the scheduler runs
pause-between-batches: 100ms # Throttle between batchesSafety:
- Only terminal statuses (
HANDLED,UNDEFINED) are cleaned up —FAILEDevents are preserved for manual inspection - A single cycle deletes at most 100 000 events; leftover events are picked up by the next cycle
- A random jitter (up to
interval) is added to the first run to avoid thundering herd when multiple instances start
The EventStore interface has two built-in implementations:
JdbcEventStore— production grade, backed by a relational database. Table is created automatically by default. Dialect-specific DDL scripts are shipped atio/github/vovten/eventflow/lifecycle/store/db/event-store-<dialect>.sqlfor manual migration tooling.InMemoryEventStore—ConcurrentHashMap-backed, for testing and single-JVM scenarios where persistence is not needed.
You can also implement EventStore with your own backend (Redis, MongoDB, etc.) and configure it via store.type.
Event Flow provides multiple ways to publish events:
| Method | Envelope | Channels | Metadata | Message Size |
|---|---|---|---|---|
implements Event |
❌ No | From event | Minimal | ⚡ Smallest |
prepare().publish() |
✅ Auto | Custom | Custom | Medium |
Note: When POJO/record or
prepare()is used, anEnvelopeis automatically created wrapping the payload with additional metadata (eventId, processId, occurredAt). This increases message size but adds correlation/tracing capabilities.
The most convenient approach for most use cases:
// Define event with default channels via annotation
@Event(channels = {InternalEventChannel.class, ExternalEventChannel.class})
public record OrderCreatedEvent(long orderId) {}
// Publish with custom metadata (channels from annotation are used automatically)
eventPublisher.prepare(new OrderCreatedEvent(1))
.withProcessId(processId)
.publish();This gives you:
- Channels from
@Eventannotation (no need to specify in code) - Auto-generated eventId and occurredAt (timestamps)
- Custom metadata via builder (processId, etc.)
- Envelope for correlation/tracing
Implement Event interface for minimum overhead — no Envelope wrapper, smallest message size:
public record OrderCreatedEvent(String orderId, String email) implements Event {
@Override
public Class<?> type() {
return OrderCreatedEvent.class;
}
@Override
public List<Class<? extends EventChannel>> channels() {
return List.of(InternalEventChannel.class, ExternalEventChannel.class);
}
}
eventPublisher.publish(new OrderCreatedEvent("order-123", "user@example.com"));Use cases: High-throughput scenarios, microservice-to-microservice communication, Kafka topics.
Publish any Java object directly — Envelope is created automatically:
// Simple POJO with @Event annotation
@Event(channels = InternalEventChannel.class)
public record OrderCreated(String orderId, String email) {}
eventPublisher.publish(new OrderCreated("order-123", "user@example.com"));Envelope with auto-generated metadata:
eventId— random UUIDprocessId— nulloccurredAt— current timestampmetadata— empty- Channels —
InternalEventChannel(default)
Use the builder for custom metadata and channels — same Envelope is created internally:
eventPublisher.prepare(new OrderCreated("order-123", "user@example.com"))
.withMetadata("key1", "data1")
.withMetadata("key2", "data2")
.withChannels(InternalEventChannel.class, ExternalEventChannel.class)
.withProcessId(UUID.randomUUID())
.withOccurredAt(Instant.now())
.publish();Note: Channels specified via
withChannels()have priority over channels defined in@Eventannotation on the payload class.
Available builder methods:
withMetadata(key, value)— add single metadata entrywithMetadata(Map)— add multiple metadata entrieswithChannel(channel)— set single channel (convenience alias)withChannels(c1)— set one channelwithChannels(c1, c2)— set two channelswithChannels(c1, c2, c3)— set three channelswithChannels(List)— set arbitrary number of channelswithProcessId(UUID)— correlation ID (e.g., saga ID)withOccurredAt(Instant)— custom event timestamppublish()— send the event
Specify default channels on the POJO/record class:
@Event(channels = ExternalEventChannel.class)
public record OrderShipped(String orderId, Instant shippedAt) {}
@Event(channels = {InternalEventChannel.class, ExternalEventChannel.class})
public record OrderDelivered(String orderId, Instant deliveredAt) {}
eventPublisher.publish(new OrderShipped("order-123", Instant.now()));| Approach | Best For | Envelope | Message Size |
|---|---|---|---|
implements Event |
High throughput, Kafka, microservices | ❌ | Smallest |
publish(POJO/record) |
Simple notifications, internal events | ✅ | Medium |
publish(POJO/record) + @Event |
Default routing configuration | ✅ | Medium |
prepare().publish() |
Custom metadata, dynamic routing | ✅ | Medium |
public record UserRegisteredEvent(String userId, String email) implements Event {
@Override
public Class<?> type() {
return UserRegisteredEvent.class;
}
@Override
public List<Class<? extends EventChannel>> channels() {
return List.of(InternalEventChannel.class, ExternalEventChannel.class);
}
}public class NotificationEventSubscriber implements EventSubscriber {
@Override
public List<Class<? extends Event>> events() {
return List.of(UserRegisteredEvent.class);
}
@Override
public void onEvent(Event event) {
if (event instanceof UserRegisteredEvent e) {
sendWelcomeEmail(e.email());
}
}
private void sendWelcomeEmail(String email) {
// Email sending logic
}
}
// Register the handler
handlerRegistry.register(new NotificationEventSubscriber());Handlers can receive the entire Envelope including metadata:
public class OrderEventHandler {
@EventListener
public void handleOrder(Envelope<OrderPlacedEvent> envelope) {
// Access payload
OrderPlacedEvent event = envelope.payload();
// Access metadata
UUID eventId = envelope.eventId();
UUID processId = envelope.processId();
Instant occurredAt = envelope.occurredAt();
Map<String, String> metadata = envelope.metadata();
System.out.println("Processed: " + event.orderId());
}
}Note: When using Envelope as a handler parameter, you must specify the payload type in the annotation:
@EventListener(OrderPlacedEvent.class)
public void handleOrder(Envelope<OrderPlacedEvent> envelope) {
// Handle envelope
}// Outgoing transport
OutTransport kafkaOut = new KafkaOutTransport(
"localhost:9092", // bootstrap servers
"events" // topic
);
EventChannel externalChannel = new ExternalEventChannel(
List.of(kafkaOut)
);
// Incoming transport
InTransport kafkaIn = new KafkaInTransport(
"localhost:9092", // bootstrap servers
"events", // topics (comma-separated)
"event-dispatcher" // group.id
);
// Dispatcher with Kafka transport
EventDispatcher dispatcher = EventDispatcherBuilder.create()
.executor(Executors.newVirtualThreadPerTaskExecutor())
.handlerRegistry(handlerRegistry)
.transports(List.of(kafkaIn))
.build();
dispatcher.start(dispatcher::dispatch);EventDispatcher idempotentDispatcher = EventDispatcherBuilder.create()
.executor(Executors.newVirtualThreadPerTaskExecutor())
.handlerRegistry(handlerRegistry)
.transports(List.of(inTransport))
.idempotent(10000, Duration.ofMinutes(5)) // max 10000 entries, 5 min TTL
.build();For transactional publishing in Spring Boot applications, see event-flow-spring/README.md.
For detailed configuration examples, see:
- Event Flow Core — LocalQueue, Kafka, custom transports, serialization
- Event Flow Spring — YAML auto-configuration, transactional publishing, retry support, lifecycle tracking
| Property | Default | Description |
|---|---|---|
event-flow.publisher.lifecycle.enabled |
false |
Enable lifecycle-aware event publishing |
event-flow.publisher.lifecycle.service-name |
"" |
Required! Service name for ack filtering |
event-flow.publisher.lifecycle.store.type |
db |
Store type: db, in-memory, or custom |
event-flow.publisher.lifecycle.store.table-name |
event_store |
Custom table name (for db type) |
event-flow.publisher.lifecycle.store.auto-init-schema |
true |
Auto-create table on startup |
event-flow.publisher.lifecycle.retry.enabled |
true |
Enable automatic retry of failed events |
event-flow.publisher.lifecycle.retry.max-retries |
3 |
Maximum retry attempts |
event-flow.publisher.lifecycle.retry.retry-interval |
30s |
Interval between retry cycles |
event-flow.publisher.lifecycle.retry.min-age |
30s |
Base backoff for exponential retry |
event-flow.dispatcher.lifecycle.enabled |
false |
Enable ack-based lifecycle tracking on dispatcher |
LocalQueue is a built-in in-JVM transport for internal event exchange. See event-flow-core/README.md for details.
Kafka transport for external event communication. See event-flow-core/README.md for configuration examples.
Implement OutTransport or InTransport interfaces to add custom transport types. See event-flow-core/README.md for an example.
Event Flow provides structured JSON logging decorators for both the publisher and the dispatcher. Each log entry captures: event status, envelope metadata (eventId, processId, occurredAt), payload (truncated), handler/transport results, duration, and distributed tracing context (traceId, spanId, deliveredFrom).
Use buildAndLog() or the loggable() builder methods:
// Publisher
EventPublisher publisher = EventPublisherBuilder.create(channel)
.loggable() // defaults: 1024 char payload
.loggable(500) // custom payload truncation
.loggable(500, Set.of("HeartbeatEvent")) // exclude noisy events
.loggable(500, Set.of(), Map.of("HeartbeatEvent", "ERROR")) // with log overrides
.build();
// Dispatcher
EventDispatcher dispatcher = EventDispatcherBuilder.create()
.executor(executor)
.handlerRegistry(registry)
.loggable()
.build();By default, log level is determined by the outcome:
| Outcome | Default level |
|---|---|
| All handlers/transports succeed | INFO |
| Partial success (some fail) | WARN |
| All fail or exception | ERROR |
With logLevels you can override the minimum log level for specific event types. The override acts as a threshold:
| Override | ERROR outcome | WARN outcome | INFO outcome |
|---|---|---|---|
ERROR |
log.error |
suppressed | suppressed |
WARN |
log.error |
log.warn |
suppressed |
INFO |
log.error |
log.warn |
log.info |
Example: Suppress logging for a high-frequency heartbeat event, only show errors:
event-flow:
dispatcher:
logging:
enabled: true
log-levels:
HeartbeatEvent: ERROR
HealthCheckEvent: WARNIn Spring Boot, these settings go into event-flow.yml or application.yml. When using the builder directly (without Spring), pass the map via loggable(maxPayloadLength, excludedEvents, logLevels):
Map<String, String> logLevels = Map.of(
"HeartbeatEvent", "ERROR",
"HealthCheckEvent", "WARN"
);
EventDispatcher dispatcher = EventDispatcherBuilder.create()
.executor(executor)
.handlerRegistry(registry)
.loggable(1024, Set.of(), logLevels)
.build();Valid level names: TRACE, DEBUG, INFO, WARN, ERROR.
┌─────────┐ ┌──────────────┐ ┌────────────┐ ┌───────────┐ ┌──────────┐ ┌───────────────┐
│ Service │ │EventPublisher│ │EventChannel│ │LocalQueue │ │Dispatcher│ │HandlerRegistry│
└────┬────┘ └──────┬───────┘ └─────┬──────┘ └─────┬─────┘ └────┬─────┘ └───────┬───────┘
│ │ │ │ │ │
│ publish(event)│ │ │ │ │
│──────────────▶│ │ │ │ │
│ │ send(event) │ │ │ │
│ │────────────────▶│ │ │ │
│ │ │ offer(event) │ │ │
│ │ │──────────────▶│ │ │
│ │ │ │ │ │
│ │ │ │ take(event) │ │
│ │ │ │─────────────▶│ │
│ │ │ │ │ │
│ │ │ │ │getHandlers(event)
│ │ │ │ │───────────────▶│
│ │ │ │ │◀───────────────│
│ │ │ │ │ [handlers] │
│ │ │ │ │ │
│ │ │ │ │ async execute │
│ │ │ │ │───────────────▶│ onEvent(event)
│ │ │ │ │ │
│◀──────────────│ (CompletableFuture) │ │ │
┌─────────┐ ┌──────────────┐ ┌──────────────┐ ┌──────────┐ ┌──────────────┐
│ Service │ │EventPublisher│ │KafkaOutTransp│ │ Kafka │ │KafkaInTransp │
└────┬────┘ └──────┬───────┘ └──────┬───────┘ └────┬─────┘ └──────┬───────┘
│ │ │ │ │
│ publish(event)│ │ │ │
│──────────────▶│ │ │ │
│ │ send(event) │ │ │
│ │─────────────────▶│ │ │
│ │ │ produce(event) │ │
│ │ │───────────────▶│ │
│ │ │ │ │
│ │ │ ack(offset) │ │
│ │ │◀───────────────│ │
│ │◀─────────────────│ │ │
│ │ SendResult │ │ │
│◀──────────────│ │ │ │
│ │ │ │ │
│ │ │ │ poll(event) │
│ │ │ │────────────────────▶│
│ │ │ │ │
│ │ │ │ event │
│ │ │ │◀────────────────────│
┌──────────┐ ┌───────────────────┐ ┌──────────────┐ ┌────────────────┐ ┌──────────┐ ┌─────────┐
│ Service │ │EventLifecycle │ │ EventStore │ │ EventChannel │ │Lifecycle │ │AckHandler│
│ │ │ Publisher │ │ (persistent)│ │ + Dispatcher │ │Dispatcher│ │ │
└────┬─────┘ └────────┬──────────┘ └──────┬───────┘ └───────┬────────┘ └────┬─────┘ └────┬────┘
│ │ │ │ │ │
│ publish(event) │ │ │ │ │
│────────────────▶│ │ │ │ │
│ │ │ │ │ │
│ │ save(NEW) │ │ │ │
│ │───────────────────▶│ │ │ │
│ │ │ │ │ │
│ │ send(event) │ │ │ │
│ │──────────────────────────────────────▶│ │ │
│ │ │ │ │ │
│ │ │ │ dispatch(event)│ │
│ │ │ │───────────────▶│ │
│ │ │ │ │ │
│ │ │ │ │ handlers │
│ │ │ │ │─────────▶ │
│ │ │ │ │◀──────── │
│ │ │ │ │ results │
│ │ │ │ │ │
│ │ update(PUBLISHED) │ │ │ │
│ │◀───────────────────│ │ │ │
│ │ │ │ │ │
│ │ │ │ │ publish ack │
│ │ │ │ │─────────────│──────────▶
│ │ │ │ │ (Success/ │ │
│ │ │ │ │ Failure) │ │
│ │ │ │ │ │ │
│ │ │ │ │ │ ack event │
│ │ │ update(HANDLED/ │ │ │◀──────────│
│ │ │ FAILED) │ │ │ │
│ │◀───────────────────│ │ │ │ │
│◀────────────────│ (CompletableFuture)│ │ │ │ │
┌──────────────┐ ┌───────────────┐ ┌─────────────────────────────────────┐
│ Dispatcher │ │HandlerRegistry│ │ Event Handlers │
└──────┬───────┘ └───────┬───────┘ └──────────────┬──────────────────────┘
│ │ │
│ dispatch(event) │ │
│──────────────────▶│ │
│ │ │
│ │ getHandlers(event) │
│ │─────────────────────────▶│
│ │ │
│ │ [Handler1, Handler2, ...]│
│ │◀─────────────────────────│
│ │ │
│ handlers │ │
│◀──────────────────│ │
│ │ │
│ executor.execute(Handler1) │
│─────────────────────────────────────────────▶│ Handler1.onEvent()
│ │ │
│ executor.execute(Handler2) │
│─────────────────────────────────────────────▶│ Handler2.onEvent()
│ │ │
│ ... │ │
│ │ │
- Event Flow Core — Detailed core module documentation
- Event Flow Spring — Spring Boot auto-configuration
- Javadoc
- Source Code
- Usage Examples
- Fork the repository
- Create a branch (
git checkout -b feature/amazing-feature) - Commit your changes (
git commit -m 'Add amazing feature') - Push to the branch (
git push origin feature/amazing-feature) - Open a Pull Request
This project is licensed under the Apache License 2.0 — see the LICENSE file for details.
- Vladimir Aleshkov (@vovten)