Skip to content
OccurrentOccurrent

DCB with domain events

Layer 1: Domain events

The Dynamic Consistency Boundary section in layer 0 covers tags, criteria and the DcbEventStore, which all work with CloudEvents. This section covers the parts that work with your domain events: an application service, deciders and subscriptions for a DCB boundary.

The DCB Application Service

Section titled “The DCB Application Service”

Running that cycle by hand for every command gets repetitive, and it is easy to forget the retry. DcbApplicationService does the read, decide, tag, and append for you, retrying automatically on a DcbAppendConditionNotFulfilledException (five attempts by default, with exponential backoff). execute(criteria, fn) reads the events matching criteria, hands them to your function, converts and tags whatever new domain events it returns, and appends them under the same boundary. The service needs a way to derive DCB tags for the events it appends, supplied once as a TagGenerator when constructing GenericDcbApplicationService, or per call through DcbExecuteOptions.tagGenerator(...). That built-in retry sits inside whatever transaction is open around execute, so if you open one yourself, retry at your own boundary instead. See Retry and Transactions.

The Java signature returns Optional<DcbAppendResult>, empty when your function decided there was nothing to do. The Kotlin extension executeOrNull returns a nullable DcbAppendResult instead, so a no-op command reads as null rather than an Optional.

When a boundary matches many events and working through them on every command becomes too slow, you can accelerate it with a DCB snapshot.

DcbCriteria boundary = DcbCriteria.tagsAnyOf(Tag.of("course", courseId), Tag.of("student", studentId));


Optional<DcbAppendResult> result = applicationService.execute(boundary, events -> {
    if (isCourseFull(events, courseId) || isStudentAtEnrollmentLimit(events, studentId)) {
        return List.of();
    }
    return List.of(new StudentEnrolledInCourse(courseId, studentId));
});

Deriving Tags From Annotations

Section titled “Deriving Tags From Annotations”

Hand-writing a TagGenerator works, but for the common case, where an event’s own fields are the tag values, AnnotationTagGenerator derives one from @DcbTag. Annotate a record component, field, or no-arg getter. The tag key comes from @DcbTag’s value (or its alias key), and falls back to the member’s own name when neither is set. The tag value is the member’s runtime value converted with toString(). A null or blank value skips that tag rather than failing.

public record StudentEnrolledInCourse(
        @DcbTag("course") String courseId,
        @DcbTag("student") String studentId) {
}


TagGenerator<StudentEnrolledInCourse> tagGenerator = new AnnotationTagGenerator<>();
// tagGenerator.tags(event) returns {Tag.of("course", courseId), Tag.of("student", studentId)}

Coupling a Decider to a Boundary

Section titled “Coupling a Decider to a Boundary”

A DcbDecider couples three things a feature would otherwise have to keep in sync by hand: the plain Decider (decide and evolve), a function from command to the DcbCriteria boundary that command needs, and a TagGenerator for the events it emits. Build one with DcbDecider.create(...) (or DcbDecider.from around an existing Decider), or the Kotlin dcbDecider(...) factory, which reads a little more naturally at the call site. In Kotlin you can also turn an existing Decider into a DcbDecider with the toDcb { ... } extension, supplying the criteria and tags for the boundary.

Once a decider is a DcbDecider, it is self-describing: given a command, it knows both what to decide and where to read from. Running it is a single call, the Kotlin DSL’s execute(command, dcbDecider) or, from Java, a DcbDeciderApplicationService wrapping the DCB application service. Either one asks the decider for the command’s boundary, reads the matching events, decides, tags the new events with the decider’s own TagGenerator, and appends, all without you naming a DcbCriteria at the call site.

DcbDecider<EnrollStudent, EnrollmentState, DomainEvent> enrollmentDecider = DcbDecider.create(
        new EnrollmentState(),
        (command, state) -> decide(command, state),
        (state, event) -> evolve(state, event),
        command -> DcbCriteria.tagsAnyOf(Tag.of("course", command.courseId()), Tag.of("student", command.studentId())),
        event -> tagsFor(event)
);


// DcbDeciderApplicationService resolves the boundary, decides, tags, and appends in one call
var deciderApplicationService = new DcbDeciderApplicationService<>(applicationService);
Optional<DcbAppendResult> result = deciderApplicationService.execute(
        new EnrollStudent(courseId, studentId),
        enrollmentDecider);

DcbDeciderApplicationService also has an execute(DcbCriteria, List<C>, DcbDecider<C, S, E>) overload for a caller that has already resolved the boundary itself, so the criteria don’t have to come from the decider a second time.

Subscribing to DCB Events

Section titled “Subscribing to DCB Events”

A DCB read model needs to see DCB events as they are written, and for a durable one, catch up on history first. @DcbSubscription is the declarative, framework-managed way to do that. It is the DCB counterpart to @StreamSubscription, filtering by event types and tags instead of an Occurrent Filter, and starting at a position instead of a time.

For an ephemeral, per-connection subscription you start and cancel by hand, a Server-Sent-Events feed scoped to one request, for example, inject the DcbSubscriptions DSL instead and call subscribe (or the Kotlin subscribeDcb) directly, passing a DcbCriteria and a DcbStartAt. DcbStartAt.beginning() replays the whole DCB sequence by position before switching to live delivery, the same catch-up behavior @DcbSubscription’s startAt = DcbStartPosition.BEGINNING gives you with the Spring Boot starter.

@DcbSubscription(id = "courseDashboard", startAt = DcbStartPosition.BEGINNING)
void onEvent(CourseEvent event) {
    dashboard.update(event);
}

@DcbSubscription filters by the DCB query built from its eventTypes (matched as any-of, and taken from the handler’s event parameter when left empty) and its tags (matched as all-of, each in "key:value" form and validated at startup so a malformed tag fails fast). startAt = DcbStartPosition.BEGINNING replays from the start of the DCB sequence, or use startAtDcbPosition to resume after a specific position instead.

@DcbSubscription(
        id = "courseDashboard",
        eventTypes = {CourseRegistered.class, StudentEnrolled.class},
        tags = {"course:123"},
        startAt = DcbStartPosition.BEGINNING)
void onEvent(CourseEvent event) {
    dashboard.update(event);
}

Reading DCB Metadata

Section titled “Reading DCB Metadata”

A DCB subscription handler can read the event’s metadata the same way a stream subscription can. Declare an org.occurrent.cloudevents.EventMetadata parameter for the generic parts (stream id, version, position, and any CloudEvent extension), or an org.occurrent.dsl.dcb.DcbEventMetadata parameter for a DCB-focused view that also exposes the event’s tags and its position as an OptionalLong. DcbEventMetadata wraps an EventMetadata, so eventMetadata() always gets you back to the generic view. In Kotlin there is also a dcbTags extension property on EventMetadata, for handlers that take the generic type.

@DcbSubscription(id = "courseDashboard")
void onEvent(CourseEvent event, DcbEventMetadata metadata) {
    Set<Tag> tags = metadata.dcbTags();
    OptionalLong position = metadata.position();
    long streamVersion = metadata.eventMetadata().getStreamVersion();
    dashboard.update(event);
}

Reactive DCB

Section titled “Reactive DCB”

Everything above has a reactive counterpart, the same way Occurrent pairs a blocking and a reactive API everywhere else. org.occurrent.eventstore.api.dcb.reactor.DcbEventStore returns Mono<DcbEventStream> from read and Mono<DcbAppendResult> from append, using the same criteria, tags, and append-condition types as the blocking store. The reactive DcbApplicationService.execute(criteria, fn) returns a Mono<DcbAppendResult> that completes empty when the domain function produced no new events. The domain function itself stays a plain synchronous Function<List<E>, List<E>>, since deciding is a pure computation over a list the read has already materialized, only the read and the append are reactive. DcbSubscriptions has a Project Reactor variant too, so a live subscription or a from-the-beginning catch-up works the same way over Flux as it does over the blocking callback API.