Skip to content
OccurrentOccurrent

Command dispatch

Layer 1: Domain events · Modules: occurrent-command-dispatch, occurrent-command-dispatch-dcb, occurrent-command-dispatch-annotation

A saga or a policy needs to issue commands without knowing how those commands get turned into events. CommandDispatcher<C> is the interface for that. It has one method you must write, void dispatch(C command), so a plain lambda is a valid dispatcher. Delivery is at-least-once, so whatever dispatch calls into must be safe to run twice on the same command. Running it twice should never append the same events a second time.

There is a second, optional method, void dispatchAll(List<C> commands). A saga hands one reaction’s whole command list to it in a single call, and the default implementation just dispatches them one at a time, which is what a plain lambda dispatcher gets.

That matters because a saga dispatches before it saves its own state. If the third of three commands fails, the first two have already been dispatched, the state is not saved, and the redelivery runs the reaction again from the top. There is no per-command progress marker, so a command that keeps failing re-issues the ones before it every time. For a receiver backed by an ApplicationService that costs nothing, because it re-reads the stream and the target rejects a command it has already applied. For something external and not idempotent, such as sending an email or charging a card, that repeated prefix is the thing to watch out for.

Every dispatcher from CommandDispatchers already overrides dispatchAll, so you get the batching without asking for it. A run of consecutive commands aimed at the same stream, or at the same DCB boundary, becomes one append, and a failure anywhere in that run leaves none of it written. Only consecutive commands aimed at the same target are batched, because dispatch preserves order. A reaction that issues one command to order A, one to order B, then another to order A is three appends rather than two, since the two order-A commands aren’t next to each other. The one exception is DcbCommandDispatchers.invocation(...), because two invocations sharing a boundary can each carry their own TagGenerator and a single append can only be tagged one way. Write your own override when the target is something Occurrent knows nothing about and you can write a batch together, for example inside one transaction.

Overriding it does not make dispatch exactly-once, and Occurrent does not promise that. It closes the partial-write window only for the dispatcher whose dispatchAll you overrode, and the contract stays at-least-once either way.

When you decorate a dispatcher, to add logging or metrics for example, extend ForwardingCommandDispatcher rather than implementing the interface again. dispatchAll is a default method, so a decorator that overrides dispatch and delegates silently turns the delegate’s one-append dispatchAll back into one dispatch call per command, reopening the partial-progress window described above. The base class forwards both methods to the delegate, so you override only what you change.

<dependency>
    <groupId>org.occurrent</groupId>
    <artifactId>occurrent-command-dispatch</artifactId>
    <version>0.34.0</version>
</dependency>

ApplicationService is the write engine underneath. It takes a stream id and a function that calls your domain model, reads the stream, lets the function decide what to append, and appends it. DeciderApplicationService wraps it so that the function is a decider. A CommandDispatcher is what the saga calls instead of the ApplicationService directly. It resolves which stream the command targets, then hands the command to the ApplicationService (or an equivalent write path) that does the actual read, decide, and append. The saga only knows the command, not the stream id or the decider behind it.

To resolve the target stream, a CommandDispatcher implementation needs a StreamIdResolver<C>, another single-method interface, String streamId(C command). Write one by hand when the stream id has to be computed from the command, or derive it automatically with @TargetStreamId.

Convenience Factories

Section titled “Convenience Factories”

Wiring a CommandDispatcher by hand means picking an ApplicationService, a Decider, and (for the stream case) a StreamIdResolver, and gluing them together yourself. CommandDispatchers.decider(...) and DcbCommandDispatchers.decider(...) do that gluing for you, one factory per style of decider.

For a stream-keyed decider, org.occurrent.command.CommandDispatchers.decider(deciderApplicationService, decider, streamIdResolver) builds a CommandDispatcher<C> that resolves the stream id, then executes the decider through the application service. It lives in occurrent-command-dispatch, the same module as CommandDispatcher itself, but using it also pulls in occurrent-decider, a light, optional dependency you’d otherwise add yourself.

public record PlaceOrder(String orderId, String productId, int quantity) {
}


Decider<PlaceOrder, OrderState, OrderEvent> placeOrderDecider = ...;
DeciderApplicationService<OrderEvent> applicationService = ...;
StreamIdResolver<PlaceOrder> streamIdResolver = PlaceOrder::orderId;


CommandDispatcher<PlaceOrder> dispatcher = CommandDispatchers.decider(applicationService, placeOrderDecider, streamIdResolver);


dispatcher.dispatch(new PlaceOrder(orderId, productId, quantity));

A dispatcher usually handles a whole family of related commands, not just one, so the resolver typically branches over the command type instead of reading a single field:

public sealed interface OrderCommand permits PlaceOrder, CancelOrder {
}


public record PlaceOrder(String orderId, String productId, int quantity) implements OrderCommand {
}


public record CancelOrder(String orderId) implements OrderCommand {
}


StreamIdResolver<OrderCommand> streamIdResolver = command -> switch (command) {
    case PlaceOrder c -> c.orderId();
    case CancelOrder c -> c.orderId();
};

When every command in the family already has a field that holds the stream id, you can skip writing the resolver and derive it from an annotation instead, as shown in Deriving the Stream Id From Annotations.

For a DcbDecider, org.occurrent.command.dcb.DcbCommandDispatchers.decider(dcbDeciderApplicationService, dcbDecider) takes only those two arguments. A DcbDecider already carries its own DcbCriteria and TagGenerator, so there’s no stream id to resolve. This factory lives in its own module, occurrent-command-dispatch-dcb, kept separate from occurrent-command-dispatch so that dispatching commands to plain, stream-keyed deciders doesn’t drag in the DCB and CloudEvent stack.

<dependency>
    <groupId>org.occurrent</groupId>
    <artifactId>occurrent-command-dispatch-dcb</artifactId>
    <version>0.34.0</version>
</dependency>
 
DcbDecider<EnrollStudent, EnrollmentState, DomainEvent> enrollmentDecider = ...;
DcbDeciderApplicationService<DomainEvent> applicationService = ...;


CommandDispatcher<EnrollStudent> dispatcher = DcbCommandDispatchers.decider(applicationService, enrollmentDecider);


dispatcher.dispatch(new EnrollStudent(courseId, studentId));

There is a third factory for the case where you have no command types at all, CommandDispatchers.invocation(applicationService), with DcbCommandDispatchers.invocation(...) beside it. Those take an Invocation, a command that carries the domain function to run instead of naming one. See sagas without command types.

Deriving the Stream Id From Annotations

Section titled “Deriving the Stream Id From Annotations”

Put @TargetStreamId on the field, record component, or getter that already holds the command’s target stream id, and AnnotationStreamIdResolver reads it with reflection, so you don’t have to write a StreamIdResolver by hand.

public record EnrollStudent(@TargetStreamId String courseId, String studentId) {
}


StreamIdResolver<EnrollStudent> streamIdResolver = new AnnotationStreamIdResolver<>();
// streamIdResolver.streamId(command) returns command.courseId()

Only one field, record component, or getter per command class can carry @TargetStreamId. AnnotationStreamIdResolver also accepts a different annotation of your own, for a command that already carries one you’d rather reuse.

<dependency>
    <groupId>org.occurrent</groupId>
    <artifactId>occurrent-command-dispatch-annotation</artifactId>
    <version>0.34.0</version>
</dependency>

This is the single-stream match to deriving DCB tags from annotations. @DcbTag marks the fields that become the tags for a DCB boundary spanning several streams. @TargetStreamId marks the one field that already is the id of a single stream.