Documentation

The documentation on this page is always for the latest version of Occurrent, currently 0.31.0.

Introduction

Occurrent is an event sourcing library, or if you wish, a set of event sourcing utilities for the JVM, created by Johan Haleby. There are many options for doing event sourcing in Java already so why build another one? There are a few reasons for this besides the intrinsic joy of doing something yourself:

• You should be able to design your domain model without any dependencies to Occurrent or any other library. Your domain model can be expressed with pure functions that returns events. Use Occurrent to store these events. This is a very important design decision! Many people talk about doing this, but I find it rare in practise, and some existing event sourcing frameworks makes this difficult or non-idiomatic.
• Simple: Pick only the libraries you need, no need for an all or nothing solution. If you don’t need subscriptions, then don’t use them! Use the infrastructure that you already have and hook these into Occurrent.
• Occurrent is not a database by itself. The goal is to be a thin wrapper around existing commodity databases that you may already be familiar with.
• Events are stored in a standard format (cloud events). You are responsible for serializing/deserializing the cloud events “body” (data) yourself. While this may seem like a limitation at first, why not just serialize your POJO directly to arbitrary JSON like you’re used to?, it really enables a lot of use cases and peace of mind. For example:
  • It should be possible to hook in various standard components into Occurrent that understands cloud events. For example a component could visualize a distributed tracing graph from the cloud events if using the distributed tracing cloud event extension.
  • Since the current idea is to be as close as possible to the specification even in the database,
    you can use the database to your advantage. For example, you can create custom indexes used for fast and fully consistent domain queries directly on an event stream (or even multiple streams).
• Composable: Function composition and pipes are encouraged. For example pipe the event stream to a rehydration function (any function that converts a stream of events to current state) before calling your domain model.
• Pragmatic: Need consistent projections? You can decide to write projections and events transactionally using tools you already know (such as Spring @Transactional)!
• Interoperable/Portable: Cloud events is a CNCF specification for describing event data in a common way. CloudEvents seeks to dramatically simplify event declaration and delivery across services, platforms, and beyond!
• Use the Occurrent components as lego bricks to compose your own pipelines. Components are designed to be small so that you should be able to re-write them tailored to your own needs if required. Missing a component? You should be able to write one yourself and hook into the rest of the eco-system. Write your own problem/domain specific layer on-top of Occurrent.

Concepts

Event Sourcing

Every system needs to store and update data somehow. Many times this is done by storing the current state of an entity in the database. For example, you might have an entity called Order stored in a order table in a relational database. Everytime something happens to the order, the table is updated with the new information and replacing the previous values. Event Sourcing is a technique that instead stores the changes, represented by events, that occurred for the entity. Events are facts, things that have happened, and they should never be updated. This means that not only can you derive the current state from the set of historic events, but you also know which steps that were involved to reach the current state.

CloudEvents

Cloud events is a CNCF specification for describing event data in a common way. CloudEvents seeks to dramatically simplify event declaration and delivery across services, platforms, and beyond. In Occurrent, you don’t persist your domain events directly to an event store, instead you convert them to a cloud event. You may regard a CloudEvent as a standardized envelope around the data in your domain event.

In practice, this means that instead of storing events in a proprietary or arbitrary format, Occurrent, stores events in accordance with the cloud event specification, even at the data-store level. I.e. you know the structure of your events, even in the database that the event store uses. It’s up to you as a user of the library to convert your domain events into cloud events when writing to the event store. This is extremely powerful, not only does it allow you to design your domains event in any way you find fit (for example without compromises enforced by a JSON serialization library) but it also allows for easier migration, data consistency and features such as (fully-consistent) queries to the event store for certain use cases. A cloud event is made-up by a set of pre-defined attributes described in the cloud event specification. In the context of event sourcing, we can leverage these attributes in the way suggested below:

Note that the table above is to be regarded as a rule of thumb, it’s ok to map things differently if it’s better suited for your application, but it’s a good idea to keep things consistent throughout your organization. To see an example of how this may look in code, refer to the application service documentation.

Occurrent CloudEvent Extensions

Occurrent automatically adds two extension attributes to each cloud event written to the event store:

These are required for Occurrent to operate. A long-term goal of Occurrent is to come up with a standardized set of cloud event extensions that are agreed upon and used by several different vendors.

Since version 0.31.0, Occurrent also stamps every event with a global, monotonically increasing position extension, shared by stream-written and DCB events alike. It gives stream consumers the same total-ordering guarantee that DCB relies on, and stream catch-up reconciles on it rather than on wall-clock time. It is enabled by default for a new store. When a MongoDB store starts up against a collection that already holds events without a position, and position was only on by default rather than turned on explicitly, the store turns position off for itself instead of building the position index over the existing collection, and logs how to enable it. Enabling position explicitly with EventStoreConfig.withStreamPosition() (or the property occurrent.event-store.stream.position=true) keeps it on even over that existing data, in which case backfill the existing events first. A stream-only store can opt out with EventStoreConfig.withoutStreamPosition() (or occurrent.event-store.stream.position=false), but a store with the DCB capability always has it enabled.

In the meantime, it’s quite possible that Occurrent will provide a wider set of optional extensions in the future (such as correlation id and/or sequence number). But for now, it’s up to you as a user to add these if you need them (see CloudEvent Metadata), you would typically do this by creating or extending/wrapping an already existing application service.

CloudEvent Metadata

You can specify metadata to the cloud event by making use of extension attributes. This is the place to add things such as sequence number, correlation id, causation id etc. Actually there’s already a standard way of applying distributed tracing and sequence number generation extensions to cloud events that might be of interest.

EventStore

The event store is a place where you store events. Events are immutable pieces of data describing state changes for a particular stream. A stream is a collection of events that are related, typically but not limited to, a particular entity. For example a stream may include all events for a particular instance of a game or an order.

Occurrent provides an interface, EventStore, that allows to read and write events from the database. The EventStore interface is actually composed of various smaller interfaces since not all databases supports all aspects provided by the EventStore interface. Here’s an example that writes a cloud event to the event store and read it back:

CloudEvent event = CloudEventBuilder.v1()
                    .withId("eventId")
                    .withSource(URI.create("urn:mydomain"))
                    .withType("HelloWorld")
                    .withTime(LocalDateTime.now().atOffset(ZoneOffset.UTC))
                    .withSubject("subject")
                    .withDataContentType("application/json")
                    .withData("{ \"message\" : \"hello\" }".getBytes(StandardCharsets.UTF_8))
                    .build();

// Write                    
eventStore.write("streamId", List.of(event));

// Read
EventStream<CloudEvent> eventStream = eventStore.read("streamId");

// Read filtered events when the store supports it
if (eventStore instanceof ReadEventStreamWithFilter filteredEventStore) {
    EventStream<CloudEvent> filteredEventStream = filteredEventStore.read(
            "streamId",
            StreamReadFilter.type("com.example.NameDefined")
    );
}

val event = CloudEventBuilder.v1()
                    .withId("eventId")
                    .withSource(URI.create("urn:mydomain"))
                    .withType("HelloWorld")
                    .withTime(LocalDateTime.now().atOffset(ZoneOffset.UTC))
                    .withSubject("subject")
                    .withDataContentType("application/json")
                    .withData("{ \"message\" : \"hello\" }".toByteArray())
                    .build()

// Write                    
eventStore.write("streamId", listOf(event))

// Read
val eventStream : EventStream<CloudEvent> = eventStore.read("streamId")

// Read filtered events when the store supports it
val filteredEventStore = eventStore as? ReadEventStreamWithFilter
val filteredEventStream = filteredEventStore?.read(
    "streamId",
    StreamReadFilter.type(NameDefined::class.java.name)
)

Note that when reading the events, the EventStore won’t simply return a Stream of CloudEvent’s, instead it returns a wrapper called EventStream.

EventStream

The EventStream contains the CloudEvent’s for a stream and the version of the stream. The version can be used to guarantee that only one thread/process is allowed to write to the stream at the same time, i.e. optimistic locking. This can be achieved by including the version in a write condition.

Note that reading a stream that doesn’t exist (e.g. eventStore.read("non-existing-id") will return an instance of EventStream with an empty stream of events and 0 as version number. The reason for this is that you can use the same “application service” (a fancy word for a piece of code that loads events from the event store, applies them to the domain model and writes the new events returned to the event store) for both entity creation and subsequent use cases. For example consider this simple domain model:

public class WordGuessingGame {
	public static List<CloudEvent> startNewGame(String gameId, String wordToGuess) {	
		...
	}

	public static List<CloudEvent> guessWord(List<CloudEvent> eventStream, String word) {
		...
	}
}

// Note that the functions might as well be placed directly in a package 
 object WordGuessingGame {
    fun startNewGame(gameId : String, wordToGuess : String) : List<CloudEvent> = ...	
 
    fun guessWord(eventStream : List<CloudEvent>, word : String) : List<CloudEvent> = ...
 }

Then we could write a generic application service that takes a higher-order function (List<CloudEvent>) -> List<CloudEvent>:

public class ApplicationService {

    private final EventStore eventStore;

    public ApplicationService(EventStore eventStore) {
        this.eventStore = eventStore;
    }

    public void execute(String streamId, Function<List<CloudEvent>, List<CloudEvent>> functionThatCallsDomainModel) {
        // Read all events from the event store for a particular stream
        EventStream<CloudEvent> eventStream = eventStore.read(streamId);

        // Invoke the domain model  
        List<CloudEvent> newEvents = functionThatCallsDomainModel.apply(eventStream.eventList());

        // Persist the new events  
        eventStore.write(streamId, eventStream.version(), newEvents);
    }
}

class ApplicationService constructor (val eventStore : EventStore) {

    fun execute(streamId : String, functionThatCallsDomainModel : (List<CloudEvent>) -> List<CloudEvent>) {
        // Read all events from the event store for a particular stream
        val  eventStream : EventStream<CloudEvent> = eventStore.read(streamId)
        
         // Invoke the domain model 
        val newEvents = functionThatCallsDomainModel(eventStream.eventList())

        // Persist the new events
        eventStore.write(streamId, eventStream.version(), newEvents)
    }
}

You could then call the application service like this regardless of you’re starting a new game or not:

// Here we image that we have received the data required to start a new game, e.g. from a REST endpoint. 
String gameId = ...
String wordToGuess = ...;

// Then we invoke the application service to start a game:
applicationService.execute(gameId, __ -> WordGuessingGame.startNewGame(gameId, wordToGuess));  

// Later a player guess a word:
String gameId = ...
String guess = ...;

// We thus invoke the application service again to guess the word:
applicationService.execute(gameId, events -> WordGuessingGame.guessWord(events, guess));

// Here we image that we have received the data required to start a new game, e.g. from a REST endpoint. 
val gameId : String = ...
val wordToGuess : String = ...;

// Then we invoke the application service to start a game:
applicationService.execute(gameId) { 
    WordGuessingGame.startNewGame(gameId, wordToGuess)
}  

// Later a player guess a word:
val gameId : String = ...
val guess : String = ...;

// We thus invoke the application service again to guess the word:
applicationService.execute(gameId) { events ->
    WordGuessingGame.guessWord(events, guess)
}

Writing application services like this is both powerful and simple (once you get used to it). There’s less need for explicit commands and command handlers (the application service is a kind of command handler). You can also use other functional techniques such as partial application to make the code look, arguably, even nicer. It’s also easy to compose several calls to the domain model into one by using standard functional composition techniques. For example in this case you might consider both starting a game and let the player make her first guess from a single request to the REST API. No need to change the domain model to do this, just use function composition.

Write Condition

A “write condition” can be used to specify conditional writes to the event store. Typically, the purpose of this would be to achieve optimistic concurrency control (optimistic locking) of an event stream.

For example, image you have an Account to which you can deposit and withdraw money. A business rule says that it’s not allowed to have a negative balance on an account. Now imagine an account that is shared between two persons and contains 20 EUR. Person “A” wants to withdraw 15 EUR and person “B” wants to withdraw 10 EUR. If they try to do this, an error message should be presented to one of them since the account balance would be negative. But what happens if both persons try to withdraw the money at the same time? Let’s have a look:

// Person A at _time 1_
EventStream<CloudEvent> eventStream = eventStore.read("account1"); // A

// "withdraw" is a pure function in the Account domain model which takes a List
//  of all current events and the amount to withdraw, and returns new events. 
// In this case, a "MoneyWasWithdrawn" event is returned,  since 15 EUR is OK to withdraw.     
List<CloudEvent> events = Account.withdraw(eventStream.eventList(), Money.of(15, EUR));

// We write the new events to the event store  
eventStore.write("account1", events);

// Now in a different thread let's imagine Person B at _time 1_
EventStream<CloudEvent> eventStream = eventStore.read("account1"); // B

// Again we want to withdraw money, and the system will think this is OK, 
// since event streams for A and B has not yet recorded that the balance is negative.   
List<CloudEvent> events = Account.withdraw(eventStream.eventList(), Money.of(10, EUR));

// We write the new events to the event store without any problems! 😱 
// But this shouldn't work since it would violate the business rule!   
eventStore.write("account1", events);

// Person A at _time 1_
val eventStream = eventStore.read("account1") // A

// "withdraw" is a pure function in the Account domain model which takes a List
//  of all current events and the amount to withdraw. It returns a list of 
// new events, in this case only a "MoneyWasWithdrawn" event,  since 15 EUR is OK to withdraw.     
val events = Account.withdraw(eventStream.eventList(), Money.of(15, EUR))

// We write the new events to the event store  
eventStore.write("account1", events)

// Now in a different thread let's imagine Person B at _time 1_
val eventStream = eventStore.read("account1") // B

// Again we want to withdraw money, and the system will think this is OK, 
// since the Account thinks that 10 EUR will have a balance of 10 EUR after 
// the withdrawal.   
val events = Account.withdraw(eventStream.eventList(), Money.of(10, EUR))

// We write the new events to the event store without any problems! 😱 
// But this shouldn't work since it would violate the business rule!   
eventStore.write("account1", events)

To avoid the problem above we want to make use of conditional writes. Let’s see how:

// Person A at _time 1_
EventStream<CloudEvent> eventStream = eventStore.read("account1"); // A
long currentVersion = eventStream.version(); 

// Withdraw money
List<CloudEvent> events = Account.withdraw(eventStream.eventList(), Money.of(15, EUR));

// We write the new events to the event store with a write condition that implies
// that the version of the event stream must be A.   
eventStore.write("account1", currentVersion, events);

// Now in a different thread let's imagine Person B at _time 1_
EventStream<CloudEvent> eventStream = eventStore.read("account1"); // A 
long currentVersion = eventStream.version();

// Again we want to withdraw money, and the system will think this is OK, 
// since event streams for A and B has not yet recorded that the balance is negative.   
List<CloudEvent> events = Account.withdraw(eventStream.eventList(), Money.of(10, EUR));

// We write the new events to the event store with a write condition that implies
// that the version of the event stream must be B. But now Occurrent will throw
// a "org.occurrent.eventstore.api.WriteConditionNotFulfilledException" since, in this
// case A was slightly faster, and the version of the event stream no longer match!
// The entire operation should be retried for person B and when "Account.withdraw(..)"
// is called again it could throw a "CannotWithdrawSinceBalanceWouldBeNegative" exception. 
eventStore.write("account1", currentVersion, events); 

// Person A at _time 1_
val eventStream = eventStore.read("account1") // A
val currentVersion = eventStream.version() 

// Withdraw money
val events = Account.withdraw(eventStream.eventList(), Money.of(15, EUR));

// We write the new events to the event store with a write condition that implies
// that the version of the event stream must be A.   
eventStore.write("account1", currentVersion, events)

// Now in a different thread let's imagine Person B at _time 1_
val eventStream = eventStore.read("account1"); // A 
val currentVersion = eventStream.version()

// Again we want to withdraw money, and the system will think this is OK, 
// since event streams for A and B has not yet recorded that the balance is negative.   
val events = Account.withdraw(eventStream.eventList(), Money.of(10, EUR))

// We write the new events to the event store with a write condition that implies
// that the version of the event stream must be B. But now Occurrent will throw
// a "org.occurrent.eventstore.api.WriteConditionNotFulfilledException" since, in this
// case A was slightly faster, and the version of the event stream no longer match!
// The entire operation should be retried for person B and when "Account.withdraw(..)"
// is called again it could throw a "CannotWithdrawSinceBalanceWouldBeNegative" exception. 
eventStore.write("account1", currentVersion, events) 

What you’ve seen above is a simple, but widely used, form of write condition. Actually, doing eventStore.write("streamId", version, events) is just a shortcut for:

eventStore.write("streamId", WriteCondition.streamVersionEq(version), events);

eventStore.write("streamId", WriteCondition.streamVersionEq(version), events)

But you can compose a more advanced write condition using a Condition:

eventStore.write("streamId", WriteCondition.streamVersion(and(lt(10), ne(5))), events);

eventStore.write("streamId", WriteCondition.streamVersion(and(lt(10), ne(5))), events)

where lt, ne and and is statically imported from org.occurrent.condition.Condition.

EventStore Queries

Since Occurrent builds on-top of existing databases it’s ok, given that you know what you’re doing*, to use the strengths of these databases. One such strength is that databases typically have good querying support. Occurrent exposes this with the EventStoreQueries interface that an EventStore implementation may implement to expose querying capabilities. For example:

OffsetDateTime lastTwoHours = OffsetDateTime.now().minusHours(2); 
// Query the database for all events the last two hours that have "subject" equal to "123" and sort these in descending order
Stream<CloudEvent> events = eventStore.query(subject("123").and(time(gte(lastTwoHours))), SortBy.time(DESCENDING));

val lastTwoHours = OffsetDateTime.now().minusHours(2);
// Query the database for all events the last two hours that have "subject" equal to "123" and sort these in descending order
val events : Stream<CloudEvent> = eventStore.query(subject("123").and(time(gte(lastTwoHours))), SortBy.time(DESCENDING))

The subject and time methods are statically imported from org.occurrent.filter.Filter and gte is statically imported from org.occurrent.condition.Condition.

EventStoreQueries is not bound to a particular stream, rather you can query any stream (or multiple streams at the same time). It also provides the ability to get an “all” stream:

// Return all events in an event store sorted by descending order
Stream<CloudEvent> events = eventStore.all(SortBy.time(DESCENDING));

// Return all events in an event store sorted by descending order
val events : Stream<CloudEvent> = eventStore.all(SortBy.time(DESCENDING))

The EventStoreQueries interface also supports skip and limit capabilities which allows for pagination:

// Skip 42, limit 1024
Stream<CloudEvent> events = eventStore.all(42, 1024);

// Skip 42, limit 1024
val events : Stream<CloudEvent> = eventStore.all(42, 1024)

To get started with an event store refer to Choosing An EventStore.

EventStore Operations

Occurrent event store implementations may optionally also implement the EventStoreOperations interface. It provides means to delete a specific event, or an entire event stream. For example:

// Delete an entire event stream
eventStoreOperations.deleteEventStream("streamId");
// Delete a specific event
eventStoreOperations.deleteEvent("cloudEventId", cloudEventSource);
// This will delete all events in stream "myStream" that has a version less than or equal to 19.
eventStoreOperations.delete(streamId("myStream").and(streamVersion(lte(19L))));

// Delete an entire event stream
eventStoreOperations.deleteEventStream("streamId")
// Delete a specific event
eventStoreOperations.deleteEvent("cloudEventId", cloudEventSource)
// This will delete all events in stream "myStream" that has a version less than or equal to 19.
eventStoreOperations.delete(streamId("myStream").and(streamVersion(lte(19L))))

These are probably operations that you want to use sparingly. Typically, you never want to remove any events, but there are some cases, such as GDPR or other regulations, that requires the deletion of an event or an entire event stream. You should be aware that there are other ways to solve this though. One way would be to encrypt personal data and throw away the key when the user no longer uses the service. Another would be to store personal data outside the event store.

Another reason for deleting events is if you’re implementing something like “closing the books” or certain types of snapshots, and don’t need the old events anymore.

Another feature provided by EventStoreOperations is the ability to update an event. Again, this is not something you normally want to do, but it can be useful for certain strategies of GDPR compliance. For example maybe you want to remove or update personal data in an event when a users unregisters from your service. Here’s an example:

eventStoreOperations.updateEvent("cloudEventId", cloudEventSource, cloudEvent -> {
    return CloudEventBuilder.v1(cloudEvent).withData(removePersonalDetailsFrom(cloudEvent)).build();
});

eventStoreOperations.updateEvent("cloudEventId", cloudEventSource) { cloudEvent -> 
    CloudEventBuilder.v1(cloudEvent).withData(removePersonalDetailsFrom(cloudEvent)).build()
}

Stream Filtering

org.occurrent.eventstore.api.StreamReadFilter is an EventStore read capability for cases where a command or use case only depends on a subset of the events in a stream. It is useful when you want to avoid reading and deserializing events that are irrelevant for a specific decision.

Filtering is a read optimization, not a correctness feature. Do not filter away events that are needed to enforce invariants, rebuild state correctly, or make valid domain decisions.

StreamReadFilter is intentionally scoped to stream reads and validates reserved stream fields such as streamid and streamversion.

Filtered stream reads are exposed via optional capability interfaces:

• org.occurrent.eventstore.api.blocking.ReadEventStreamWithFilter
• org.occurrent.eventstore.api.reactor.ReadEventStreamWithFilter

The following event stores support filtered stream reads:

• InMemoryEventStore
• MongoEventStore
• SpringMongoEventStore
• ReactorMongoEventStore

A blocking example:

if (eventStore instanceof ReadEventStreamWithFilter filteredEventStore) {
    EventStream<CloudEvent> eventStream = filteredEventStore.read(
            "streamId",
            StreamReadFilter.type("com.acme.NameDefined")
    );
}

val filteredEventStore = eventStore as? ReadEventStreamWithFilter
val eventStream = filteredEventStore?.read(
    "streamId",
    StreamReadFilter.type(NameDefined::class.java.name)
)

A reactor example:

if (eventStore instanceof org.occurrent.eventstore.api.reactor.ReadEventStreamWithFilter filteredEventStore) {
    Mono<EventStream<CloudEvent>> eventStream = filteredEventStore.read(
            "streamId",
            StreamReadFilter.type("com.acme.NameDefined")
    );
}

val filteredEventStore = eventStore as? org.occurrent.eventstore.api.reactor.ReadEventStreamWithFilter
val eventStream = filteredEventStore?.read(
    "streamId",
    StreamReadFilter.type(NameDefined::class.java.name)
)

Subscriptions

A subscription is a way to get notified when new events are written to an event store. Typically, a subscription will be used to create views from events (such as projections, sagas, snapshots etc) or create integration events that can be forwarded to another piece of infrastructure such as a message bus. There are two different kinds of API’s, the first one is a blocking API represented by the org.occurrent.subscription.api.blocking.SubscriptionModel interface (in the org.occurrent:occurrent-subscription-api-blocking module), and second one is a reactive API represented by the org.occurrent.subscription.api.reactor.SubscriptionModel interface (in the org.occurrent:occurrent-subscription-api-reactor module).

The blocking API is callback based, which is fine if you’re working with individual events (you can of course use a simple function that aggregates events into batches yourself). If you want to work with streams of data, the reactor SubscriptionModel is probably a better option since it’s using the Flux publisher from project reactor.

Note that it’s fine to use reactive SubscriptionModel, even though the event store is implemented using the blocking api, and vice versa. If the datastore allows it, you can also run subscriptions in a different process than the processes reading and writing to the event store.

To get started with subscriptions refer to Using Subscriptions.

Independently of the blocking-versus-reactive choice above, a subscription is either asynchronous or synchronous. Everything so far is asynchronous: it runs on its own thread and fires after the write commits. A synchronous subscription instead runs inline on the writer thread, before execute returns, so it can update a projection in the write path and, with a transaction, atomically with the write.

Views

Occurrent has a higher-level Projection DSL for maintaining views/projections, described further down. You don’t have to reach for it, though. At its simplest a view is just a subscription in which you create and store the view as you find fit. And even that doesn’t have to be difficult! Here’s a trivial example of a view that maintains the number of ended games. It does so by inceasing the “numberOfEndedGames” field in an (imaginary) database for each “GameEnded” event that is written to the event store:

// An imaginary database API
Database someDatabase = ...
// Subscribe to all "GameEnded" events by starting a subscription named "my-view" 
// and increase "numberOfEndedGames" for each ended game.   
subscriptionModel.subscribe("my-view", filter(type("GameEnded")), cloudEvent -> someDatabase.inc("numberOfEndedGames"));        

// An imaginary database API
val someDatabase : Database = ...
// Subscribe to all "GameEnded" events by starting a subscription named "my-view" 
// and increase "numberOfEndedGames" for each ended game. 
subscriptionModel.subscribe("my-view", filter(type("GameEnded"))) {  
    someDatabase.inc("numberOfEndedGames")
}

Where filter is imported from org.occurrent.subscription.StreamSubscriptionFilter and type is imported from org.occurrent.condition.Condition.

While this is a trivial example it shouldn’t be difficult to create a view that is backed by a JPA entity in a relational database based on a subscription.

When the fold gets more involved, or you want to unit-test it in isolation and reuse it across an asynchronous subscription, a synchronous write, and an on-demand query, reach for Occurrent’s View DSL (org.occurrent:occurrent-view-dsl). It is the read-side counterpart of a decider: a decider folds events and decides new ones, a View folds events into state you read. It is blocking-only, and it is the primitive the higher-level Projection DSL builds on.

So there are three levels for a read model, and you should pick the lowest one that does the job. A plain subscription that writes to your store, as at the top of this section, when the fold is trivial. The View DSL, when you want a fold you can unit-test on its own and store wherever you like. The Projection DSL, when you want the fold, the event types it handles, the view-instance id, and the delivery mode declared together in one place.

Commands

A command is used to represent an intent in an event sourced system, i.e. something that you want to do. They’re different, in a very important way, from events in that commands can fail or be rejected, where-as events cannot. A typical example of a command would be a data structure whose name is defined as an imperative verb, for example PlaceOrder. The resulting event, if the command is processed successfully, could then be OrderPlaced. However, in Occurrent, as explained in more detail in the Command Philosophy section below, you may start off by not using explicit data structures for commands unless you want to. In Occurrent, you can instead use pure functions to represent commands and command handling. Combine this with function composition and you have a powerful way to invoke the domain model (refer to the application service for examples).

Command Philosophy

Occurrent doesn’t contain a built-in command bus. Instead, you’re encouraged to pick any infrastructure component you need to act as the command bus to send commands to another service. Personally, I typically make a call to a REST API or make an RPC invocation instead of using a distributed command bus that routes the commands to my aggregate. There are of course exceptions to this, such as the need for location transparency or if you’re using Decider’s. If you need location transparency, a command bus or an actor model can be of help. But I would argue that you may not always need the complexity by prematurely going down this route if your business requirements doesn’t point you in this direction. Decider’s are a nice alternative, that doesn’t require any additional infrastructure.

But what about internally? For example if a service exposes a REST API and upon receiving a request it publishes a command that’s somehow picked up and routed to a function in your domain model. This is where an application service becomes useful. However, let’s first explore the rationale behind the philosophy of Occurrent. In other frameworks, it’s not uncommon that you define your domain model like this:

public class WordGuessingGame extends AggregateRoot {

    @AggregateId
    private String gameId;
    private String wordToGuess;

    @HandleCommand
    public void handle(StartNewGameCommand startNewGameCommand) {
        // Insert some validation and logic
        ... 
        // Publish an event using the "publish" method from AggregateRoot
        publish(new WordGuessingGameWasStartedEvent(...));
    }
    
    @HandleCommand
    public void handle(GuessWordCommand guessWordCommand) {
        // Some validation and implementation ...
        ...	
        
        // Publish an event using the "publish" method from AggregateRoot
        publish(new WordGuessedEvent(...));
    }

    @HandleEvent
    public void handle(WordGuessingGameWasStartedEvent e) {
        this.gameId = e.getGameId();
        this.wordToGuess = e.getWordToGuess();    
    }        

    ... 
}

 class WordGuessingGame : AggregateRoot() {
 
     @AggregateId
     var gameId : String
     var wordToGuess : String
 
     @HandleCommand
     fun handle(startNewGameCommand : StartNewGameCommand) {
         // Insert some validation and logic
         ... 
         // Publish an event using the "publish" method from AggregateRoot
         publish(WordGuessingGameWasStartedEvent(...))
     }
     
     @HandleCommand
     fun handle(guessWordCommand : GuessWordCommand) {
         // Some validation and implementation ...
         ...	
         
         // Publish an event using the "publish" method from AggregateRoot
         publish(WordGuessedEvent(...))
     }
 
     @HandleEvent
     fun handle(e : WordGuessingGameWasStartedEvent) {
         gameId = e.getGameId()
         wordToGuess = e.getWordToGuess()    
     }        
 
     ... 
 }

Let’s look at a “command” and see what it typically looks like in these frameworks:

public class StartNewGameCommand {
    
    @AggregateId
    private String gameId;
    private String wordToGuess;
    
    public void setGameId(String gameId) {
        this.gameId = gameId;
    }
    
    public String getGameId() {
        return gameId;
    }

    public void setWordToGuess(String wordToGuess) {
        this.gameId = gameId;
    }
    
    public String getWordToGuess() {
        return wordToGuess;
    }
    
    // Equals/hashcode/tostring methods are excluded for breivty
}

data class StartNewGameCommand(@AggregateId var gameId: String, val wordToGuess : String)

Now that we have our WordGuessingGame implementation and a command we can dispatch it to a command bus:

commandbus.dispatch(new StartNewGameCommand("someGameId", "Secret word"));

commandbus.dispatch(StartNewGameCommand("someGameId", "Secret word"))

From a typical Java perspective one could argue that this is not too bad. But it does have a few things one could improve upon from a broader perspective:

1. The WordGuessingGame is complecting several things that may be modelled separately. Data, state, behavior, command- and event routing and event publishing are all defined in the same model (the WordGuessingGame class). It also uses framework specific annotations, classes and inheritance inside your domain model which is something you want to avoid. For small examples like this it arguably doesn’t matter, but if you have complex logic and a large system, it probably will in my experience. Keeping state and behavior separate allows for easier testing, referential transparency and function composition. It allows treating the state as a value which has many benefits.
2. Commands are defined as explicit data structures with framework-specific annotations when arguably they don’t have to. This is fine if you need to serialize the command (in order to send it to another location or to schedule it for the future) but one could argue that you don’t want to couple your commands to some infrastructure. This is of course a trade-off, in Occurrent you’re free to choose any approach you like (i.e. commands/functions can be completely free of framework/library/infrastructure concerns).

Commands in Occurrent

So how would one dispatch commands in Occurrent? As we’ve already mentioned there’s nothing stopping you from using a (distributed) command bus or to create explicit commands, and dispatch them the way we did in the example above. For example, if you’re using Decider’s, it could be nice to have commands explicitly defined. But if you recognize some of the points described above and are looking for a simpler approach, here’s another way to go about. First let’s refactor the domain model to pure functions, without any state or dependencies to Occurrent or any other library/framework.

public class WordGuessingGame {
	public static List<DomainEvent> startNewGame(String gameId, String wordToGuess) {	
		...
	}

	public static List<DomainEvent> guessWord(List<DomainEvent> eventStream, String word) {
		...
	}
}

// Note that the functions might as well be placed directly in a package.
 object WordGuessingGame {
    fun startNewGame(gameId : String, wordToGuess : String) : List<DomainEvent> = ...	
 
    fun guessWord(eventStream : List<DomainEvent>, word : String) : List<DomainEvent> = ...
 }

If you define your behavior like this it’ll be easy to test (and also to compose using normal function composition techniques). There are no side-effects (such as publishing events) which also allows for easier testing and local reasoning.

But where are our commands!? In this example we’ve decided to represent them as functions. I.e. the “command” is modeled as simple function, e.g. startNewGame! This means that the command handling logic is handled by function as well. You don’t need to switch/match over the command since you directly invoke the function itself. Again, you may prefer to actually define your commands explicitly, but in this example we’ll just be using normal functions.

But wait, how are these functions called? Create or copy a generic ApplicationService class like the one below (or use the generic application service provided by Occurrent):

public class ApplicationService {

    private final EventStore eventStore;
    private final Function<CloudEvent, DomainEvent> convertCloudEventToDomainEvent;
    private final Function<DomainEvent, CloudEvent> convertDomainEventToCloudEvent;

    public ApplicationService(EventStore eventStore, 
                              Function<CloudEvent, DomainEvent> convertCloudEventToDomainEvent, 
                              Function<DomainEvent, CloudEvent> convertDomainEventToCloudEvent) {
        this.eventStore = eventStore;
        this.convertCloudEventToDomainEvent = convertCloudEventToDomainEvent;
        this.convertDomainEventToCloudEvent = convertDomainEventToCloudEvent;
    }

    public void execute(String streamId, Function<List<DomainEvent>, List<DomainEvent>> functionThatCallsDomainModel) {
        // Read all events from the event store for a particular stream
        EventStream<CloudEvent> eventStream = eventStore.read(streamId);
        // Convert the cloud events into domain events
        List<DomainEvent> domainEventsInStream = eventStream.events().map(convertCloudEventToDomainEvent).toList();

        // Call a pure function from the domain model which returns a List of domain events
        List<DomainEvent> newDomainEvents = functionThatCallsDomainModel.apply(domainEventsInStream);

        // Convert domain events to cloud events and write them to the event store
        List<CloudEvent> newCloudEvents = newDomainEvents.stream().map(convertDomainEventToCloudEvent).toList();
        eventStore.write(streamId, eventStream.version(), newCloudEvents);
    }
}

class ApplicationService constructor (val eventStore : EventStore, 
                                      val convertCloudEventToDomainEvent : (CloudEvent) -> DomainEvent, 
                                      val convertDomainEventToCloudEvent : (DomainEvent) -> CloudEvent) {

    fun execute(streamId : String, functionThatCallsDomainModel : (List<DomainEvent>) -> List<DomainEvent>) {
        // Read all events from the event store for a particular stream
        val  eventStream : EventStream<CloudEvent> = eventStore.read(streamId)
        // Convert the cloud events into domain events
        val domainEventsInStream : List<DomainEvent> = eventStream.events().map(convertCloudEventToDomainEvent).toList()

        // Call a pure function from the domain model which returns a List of domain events
        val newDomainEvents = functionThatCallsDomainModel(domainEventsInStream)

        // Convert domain events to cloud events and write them to the event store
        val newCloudEvents = newDomainEvents.map(convertDomainEventToCloudEvent)
        eventStore.write(streamId, eventStream.version(), newCloudEvents)
    }
}

and then use the ApplicationService like this:

// A function that converts a CloudEvent to a "domain event"
Function<CloudEvent, DomainEvent> convertCloudEventToDomainEvent = ..
// A function that a "domain event" to a CloudEvent
Function<DomainEvent, CloudEvent> convertDomainEventToCloudEvent = ..
EventStore eventStore = ..
ApplicationService applicationService = new ApplicationService(eventStore, convertCloudEventToDomainEvent, convertDomainEventToCloudEvent);

// Now in your REST API use the application service:
String gameId = ... // From a form parameter
String wordToGuess = .. // From a form parameter
applicationService.execute(gameId, events -> WordGuessingGame.startNewGame(gameId, wordToGuess));

// A function that converts a CloudEvent to a "domain event"
val convertCloudEventToDomainEvent : (CloudEvent) -> DomainEvent = ..
// A function that a "domain event" to a CloudEvent
val convertDomainEventToCloudEvent = (DomainEvent) -> CloudEvent  = ..
val eventStore : EventStore = ..
val applicationService = ApplicationService(eventStore, convertCloudEventToDomainEvent, convertDomainEventToCloudEvent);

// Now in your REST API use the application service:
val gameId = ... // From a form parameter
val wordToGuess = .. // From a form parameter
applicationService.execute(gameId) { events -> 
    WordGuessingGame.startNewGame(gameId, wordToGuess)
}

We’re leveraging higher-order functions instead of using explicit commands.

Command Composition

Many times it’s useful to compose multiple commands into a single unit-of-work for the same stream/aggregate. What this means that you’ll “merge” several commands into one, and they will be executed in an atomic fashion. I.e. either all commands succeed, or all commands fail.

While you’re free to use any means and/or library to achieve this, Occurrent ships with a “command composition” library that you can leverage:

<dependency>
    <groupId>org.occurrent</groupId>
    <artifactId>occurrent-command-composition</artifactId>
    <version>0.31.0</version>
</dependency>

compile 'org.occurrent:occurrent-command-composition:0.31.0'

libraryDependencies += "org.occurrent" % "occurrent-command-composition" % "0.31.0"

@Grab(group='org.occurrent', module='occurrent-command-composition', version='0.31.0') 

[org.occurrent/occurrent-command-composition "0.31.0"]

'org.occurrent:occurrent-command-composition:jar:0.31.0'

<dependency org="org.occurrent" name="occurrent-command-composition" rev="0.31.0" />

As an example consider this simple domain model:

public class WordGuessingGame {
    public List<DomainEvent> startNewGame(List<DomainEvent> events, String gameId, String wordToGuess) {
        // Implementation
    }
    
    public List<DomainEvent> makeGuess(List<DomainEvent> events, String guess) {
        // Implementation
    }
}

object WordGuessingGame {
    fun startNewGame(events : List<DomainEvent>, gameId : String, wordToGuess : String) : List<DomainEvent> {
        // Implementation
    } 

    fun makeGuess(events : List<DomainEvent>, guess : String) : List<DomainEvent> {
        // Implementation
    }    
}

Imagine that for a specific API you want to allow starting a new game and making a guess in the same request. Instead of changing your domain model, you can use function composition! If you statically import composeCommands from org.occurrent.application.composition.command.ListCommandComposition you can do like this:

String gameId = ...
String wordToGuess = ...
String guess = ...
applicationService.execute(gameId, composeCommands(
    events -> WordGuessingGame.startNewGame(events, gameId, wordToGuess),
    events -> WordGuessingGame.makeGuess(events, guess) 
));

val gameId = ...
val wordToGuess = ...
val guess = ...
applicationService.execute(gameId, composeCommands(
    { events -> WordGuessingGame.startNewGame(events, gameId, wordToGuess) }
    { events -> WordGuessingGame.makeGuess(events, guess) } 
))

If you’re using Kotlin you can also make use of the andThen (infix) function for command composition (import org.occurrent.application.composition.command.andThen):

applicationService.execute(gameId,
    { events -> WordGuessingGame.startNewGame(events, gameId, wordToGuess) }
        andThen { events -> WordGuessingGame.makeGuess(events, guess) })

Events returned from WordGuessingGame.startNewGame(..) will be appended to the event stream when calling WordGuessingGame.makeGuess(..) and the new domain events returned by the two functions will be merged and written in an atomic fashion to the event store.

The command composition library also contains some utilities for partial function application that you can use to further enhance the example above (if you like). If you statically import partial method from org.occurrent.application.composition.command.partial.PartialFunctionApplication you can refactor the code above into this:

String gameId = ...
String wordToGuess = ...
String guess = ...
applicationService.execute(gameId, composeCommands(
    partial(WordGuessingGame::startNewGame, gameId, wordToGuess),
    partial(WordGuessingGame::makeGuess, guess)
));

val gameId = ...
val wordToGuess = ...
val guess = ...
applicationService.execute(gameId, composeCommands(
    WordGuessingGame::startNewGame.partial(gameId, wordToGuess)
    WordGuessingGame::makeGuess.partial(guess) 
))

With Kotlin, you can also use andThen (described above) to do:

applicationService.execute(gameId, 
    WordGuessingGame::startNewGame.partial(gameId, wordToGuess)
            andThen WordGuessingGame::makeGuess.partial(guess))

CloudEvent Conversion

To convert between domain events and cloud events you can use the cloud event converter API that’s shipped with Occurrent. This is optional, but components such as the application service and subscription dsl uses a cloud event converter to function. If you’re only using an event store and subscriptions then you don’t need a cloud event converter (or you can roll your own). All cloud event converters implements the org.occurrent.application.converter.CloudEventConverter interface from the org.occurrent:occurrent-cloudevent-converter-api module (see custom cloudevent converter).

Generic CloudEvent Converter

This is a really simple cloud event converter to which you can pass two higher-order functions that converts to and from domain events respectively. To use it depend on:

<dependency>
    <groupId>org.occurrent</groupId>
    <artifactId>occurrent-cloudevent-converter-generic</artifactId>
    <version>0.31.0</version>
</dependency>

compile 'org.occurrent:occurrent-cloudevent-converter-generic:0.31.0'

libraryDependencies += "org.occurrent" % "occurrent-cloudevent-converter-generic" % "0.31.0"

@Grab(group='org.occurrent', module='occurrent-cloudevent-converter-generic', version='0.31.0') 

[org.occurrent/occurrent-cloudevent-converter-generic "0.31.0"]

'org.occurrent:occurrent-cloudevent-converter-generic:jar:0.31.0'

<dependency org="org.occurrent" name="occurrent-cloudevent-converter-generic" rev="0.31.0" />

For example:

Function<CloudEvent, DomainEvent> convertCloudEventToDomainEventFunction = .. // You implement this function
Function<DomainEvent, CloudEvent> convertDomainEventToCloudEventFunction = .. // You implement this function
CloudEventConverter<CloudEvent> cloudEventConverter = new GenericCloudEventConverter<>(convertCloudEventToDomainEventFunction, convertDomainEventToCloudEventFunction);

If your domain model is already using a CloudEvent (and not a custom domain event) then you can just pass a Function.identity() to the GenericCloudEventConverter:

CloudEventConverter<CloudEvent> cloudEventConverter = new GenericCloudEventConverter<>(Function.identity(), Function.identity());

XStream CloudEvent Converter

This cloud event converter uses XStream to convert domain events to cloud events to XML and back. To use it, first depend on this module:

<dependency>
    <groupId>org.occurrent</groupId>
    <artifactId>occurrent-cloudevent-converter-xstream</artifactId>
    <version>0.31.0</version>
</dependency>

compile 'org.occurrent:occurrent-cloudevent-converter-xstream:0.31.0'

libraryDependencies += "org.occurrent" % "occurrent-cloudevent-converter-xstream" % "0.31.0"

@Grab(group='org.occurrent', module='occurrent-cloudevent-converter-xstream', version='0.31.0') 

[org.occurrent/occurrent-cloudevent-converter-xstream "0.31.0"]

'org.occurrent:occurrent-cloudevent-converter-xstream:jar:0.31.0'

<dependency org="org.occurrent" name="occurrent-cloudevent-converter-xstream" rev="0.31.0" />

Next you can instantiate it like this:

XStream xStream = new XStream();
xStream.allowTypeHierarchy(MyDomainEvent.class);
URI cloudEventSource = URI.create("urn:company:domain") 
XStreamCloudEventConverter<MyDomainEvent> cloudEventConverter = new XStreamCloudEventConverter<>(xStream, cloudEventSource);

val xStream = XStream().apply { allowTypeHierarchy(MyDomainEvent::class.java) }
val cloudEventSource = URI.create("urn:company:domain")
val cloudEventConverter = new XStreamCloudEventConverter<>(xStream, cloudEventSource)

You can also configure how different attributes of the domain event should be represented in the cloud event by using the builder, new XStreamCloudEventConverter.Builder<MyDomainEvent>().. build().

Jackson CloudEvent Converter

Jackson 3

Occurrent 0.20.0 provides a Jackson 3-native CloudEvent converter and that is the recommended choice for new code. To use it, first depend on this module:

<dependency>
    <groupId>org.occurrent</groupId>
    <artifactId>occurrent-cloudevent-converter-jackson3</artifactId>
    <version>0.31.0</version>
</dependency>

compile 'org.occurrent:occurrent-cloudevent-converter-jackson3:0.31.0'

libraryDependencies += "org.occurrent" % "occurrent-cloudevent-converter-jackson3" % "0.31.0"

@Grab(group='org.occurrent', module='occurrent-cloudevent-converter-jackson3', version='0.31.0') 

[org.occurrent/occurrent-cloudevent-converter-jackson3 "0.31.0"]

'org.occurrent:occurrent-cloudevent-converter-jackson3:jar:0.31.0'

<dependency org="org.occurrent" name="occurrent-cloudevent-converter-jackson3" rev="0.31.0" />

Next you can instantiate it like this:

import org.occurrent.application.converter.jackson3.JacksonCloudEventConverter;
import tools.jackson.databind.ObjectMapper;
import tools.jackson.databind.json.JsonMapper;

ObjectMapper objectMapper = JsonMapper.builder().build();
URI cloudEventSource = URI.create("urn:company:domain")
JacksonCloudEventConverter<MyDomainEvent> cloudEventConverter = new JacksonCloudEventConverter<>(objectMapper, cloudEventSource);

import org.occurrent.application.converter.jackson3.JacksonCloudEventConverter
import tools.jackson.module.kotlin.jacksonObjectMapper

val objectMapper = jacksonObjectMapper()
val cloudEventSource = URI.create("urn:company:domain")
val cloudEventConverter = JacksonCloudEventConverter<MyDomainEvent>(objectMapper, cloudEventSource)

You can also configure how different attributes of the domain event should be represented in the cloud event by using the builder. In production, you almost certainly want to change the way the JacksonCloudEventConverter generates the cloud event type from the domain event. By default, the cloud event type will be generated from the fully-qualified class name of the domain event class type. I.e. if you do:

import org.occurrent.application.converter.jackson3.JacksonCloudEventConverter;

CloudEventConverter<MyDomainEvent> cloudEventConverter = new JacksonCloudEventConverter<>(objectMapper, cloudEventSource);
CloudEvent cloudEvent = cloudEventConverter.toCloudEvent(new SomeDomainEvent());

Then cloudEvent.getType() will return com.mycompany.SomeDomainEvent. Typically, you want to decouple the cloud event type from the fully-qualified name of the class. A better, but arguably still not optimal way, would be to make cloudEvent.getType() return SomeDomainEvent instead. The JacksonCloudEventConverter allows us to do this by using the builder:

import org.occurrent.application.converter.jackson3.JacksonCloudEventConverter;

CloudEventConverter<MyDomainEvent> cloudEventConverter = new JacksonCloudEventConverter.Builder<MyDomainEvent>(objectMapper, cloudEventSource)
        .typeMapper(..) // Specify a custom way to map the domain event to a cloud event and vice versa
        .build();

But when using Jackson, we can’t just configure the type mapper to return the “simple name” of the domain event class instead of the fully-qualified name. This is because there’s no generic way to derive the fully-qualified name from just the simple name. The fully-qualified name is needed in order for Jackson to map the cloud event back into a domain event. In order to work-around this you could implement your own type mapper (that you pass to the builder above) or create an instance of ReflectionCloudEventTypeMapper that knows how to convert the “simple name” cloud event type back into the domain event class. There are a couple of ways, the most simple one is probably this:

CloudEventTypeMapper<MyDomainEvent> typeMapper = ReflectionCloudEventTypeMapper.simple(MyDomainEvent.class);

This will create an instance of ReflectionCloudEventTypeMapper that uses the simple name of the domain event as cloud event type. But the crucial thing is that when deriving the domain event type from the cloud event, the ReflectionCloudEventTypeMapper will prepend the package name of supplied domain event type (MyDomainEvent) to the cloud event type, thus reconstructing the fully-qualified name of the class. For this to work, all domain events must reside in exactly the same package as MyDomainEvent.

Another approach would be to supply a higher-order function that knows how to map the cloud event type back into a domain event class.

CloudEventTypeMapper<MyDomainEvent> typeMapper = ReflectionCloudEventTypeMapper.simple(cloudEventType -> ...);

Again, this will create an instance of ReflectionCloudEventTypeMapper that uses the simple name of the domain event as cloud event type, but you are responsible to, somehow, map the cloud event type (cloudEventType) back into a domain event class.

If you don’t want to use reflection or don’t want to couple the class name to the event name (which is recommended) you can roll your own custom CloudEventTypeMapper by implementing the org.occurrent.application.converter.typemapper.CloudEventTypeMapper interface.

As of version 0.20.5 the builder can also truncate the cloud event time to a given precision with timePrecision(ChronoUnit):

CloudEventConverter<MyDomainEvent> cloudEventConverter = new JacksonCloudEventConverter.Builder<MyDomainEvent>(objectMapper, cloudEventSource)
        .timePrecision(ChronoUnit.MILLIS)
        .build();

val cloudEventConverter = JacksonCloudEventConverter.Builder<MyDomainEvent>(objectMapper, cloudEventSource)
        .timePrecision(ChronoUnit.MILLIS)
        .build()

This matters when the event store uses TimeRepresentation.DATE, which cannot store the nanoseconds that Instant.now() and OffsetDateTime.now() carry on modern JVMs, so an append would otherwise fail. With the Spring Boot starter you set it through the occurrent.cloud-event-converter.time-precision property (a ChronoUnit, for example millis). When that property is unset and the event store time-representation is DATE, the converter defaults to truncating to MILLIS, so the common case needs no configuration. RFC_3339_STRING keeps full precision.

If you are migrating an existing application and need to stay on the old Jackson 2 API for a while, org.occurrent:occurrent-cloudevent-converter-jackson is still available as a compatibility lane. However, the recommended direction for new applications and updated documentation is Jackson 3.

Jackson 2 Compatibility

If you’re maintaining an existing application that still uses the Jackson 2 API, you can continue to use the Jackson 2 compatibility lane. Depend on org.occurrent:occurrent-cloudevent-converter-jackson:

<dependency>
    <groupId>org.occurrent</groupId>
    <artifactId>occurrent-cloudevent-converter-jackson</artifactId>
    <version>0.31.0</version>
</dependency>

Then instantiate it like this:

import com.fasterxml.jackson.databind.ObjectMapper;
import org.occurrent.application.converter.jackson.JacksonCloudEventConverter;

ObjectMapper objectMapper = new ObjectMapper();
URI cloudEventSource = URI.create("urn:company:domain");
JacksonCloudEventConverter<MyDomainEvent> cloudEventConverter = new JacksonCloudEventConverter<>(objectMapper, cloudEventSource);

import com.fasterxml.jackson.module.kotlin.jacksonObjectMapper
import org.occurrent.application.converter.jackson.JacksonCloudEventConverter

val objectMapper = jacksonObjectMapper()
val cloudEventSource = URI.create("urn:company:domain")
val cloudEventConverter = JacksonCloudEventConverter<MyDomainEvent>(objectMapper, cloudEventSource)

Jackson 2 still works and is intended for existing applications that are not yet ready to move to the Jackson 3 API. For new applications, prefer the Jackson 3 converter documented above.

Custom CloudEvent Converter

To create a custom cloud event converter first depend on:

<dependency>
    <groupId>org.occurrent</groupId>
    <artifactId>occurrent-cloudevent-converter-api</artifactId>
    <version>0.31.0</version>
</dependency>

compile 'org.occurrent:occurrent-cloudevent-converter-api:0.31.0'

libraryDependencies += "org.occurrent" % "occurrent-cloudevent-converter-api" % "0.31.0"

@Grab(group='org.occurrent', module='occurrent-cloudevent-converter-api', version='0.31.0') 

[org.occurrent/occurrent-cloudevent-converter-api "0.31.0"]

'org.occurrent:occurrent-cloudevent-converter-api:jar:0.31.0'

<dependency org="org.occurrent" name="occurrent-cloudevent-converter-api" rev="0.31.0" />

Let’s have a look at a naive example of how we can create a custom converter that converts domain events to cloud events (and vice versa). This cloud event converter can then be used with the generic application service (the application service implementation provided by Occurrent) and other Occurrent components that requires a CloudEventConverter. Note that instead of using the code below you might as well use the Jackson CloudEvent Converter, this is just an example showing how you could roll your own.

import tools.jackson.databind.ObjectMapper;
import io.cloudevents.CloudEvent;
import io.cloudevents.core.builder.CloudEventBuilder;
import org.occurrent.application.converter.CloudEventConverter;

import java.io.IOException;
import java.net.URI;

import static java.time.ZoneOffset.UTC;
import static org.occurrent.functional.CheckedFunction.unchecked;
import static org.occurrent.time.TimeConversion.toLocalDateTime;

public class MyCloudEventConverter implements CloudEventConverter<DomainEvent> {

    private final ObjectMapper objectMapper;

    public MyCloudEventConverter(ObjectMapper objectMapper) {
        this.objectMapper = objectMapper;
    }

    @Override
    public CloudEvent toCloudEvent(DomainEvent e) {
        try {
            return CloudEventBuilder.v1()
                    .withId(e.getEventId())
                    .withSource(URI.create("urn:myapplication:streamtype"))
                    .withType(getCloudEventType(e))
                    .withTime(LocalDateTime.ofInstant(e.getDate().toInstant(), UTC).atOffset(UTC)
                                           .truncatedTo(ChronoUnit.MILLIS))
                    .withSubject(e.getName())
                    .withDataContentType("application/json")
                    .withData(objectMapper.writeValueAsBytes(e))
                    .build();
        } catch (JsonProcessingException jsonProcessingException) {
            throw new RuntimeException(jsonProcessingException);
        }
    }

    @Override
    public DomainEvent toDomainEvent(CloudEvent cloudEvent) {
        try {
            return (DomainEvent) objectMapper.readValue(cloudEvent.getData().toBytes(), Class.forName(cloudEvent.getType()));
        } catch (IOException | ClassNotFoundException e) {
            throw new RuntimeException(e);
        }
    }
    
    @Override
    public String getCloudEventType(Class<? extends T> type) {
        return type.getName();
    }
}        

To see what the attributes mean in the context of event sourcing refer to the CloudEvents documentation. You can also have a look at GenericApplicationServiceTest.java for an actual code example.

Note that if the data content type in the CloudEvent is specified as “application/json” (or a json compatible content-type) then Occurrent will automatically store it as Bson in a MongoDB event store. The reason for this so that you’re able to query the data, either by the EventStoreQueries API, or manually using MongoDB queries. In order to do this, the byte[] passed to withData, will be converted into a org.bson.Document that is later written to the database. This is not optimal from a performance perspective. A more performant option would be to make use of the io.cloudevents.core.data.PojoCloudEventData class. This class implements the io.cloudevents.CloudEventData interface and allows passing a pre-baked Map or org.bson.Document instance to it. Then no additional conversion will need to take place! Here’s an example:

import tools.jackson.databind.ObjectMapper;
import io.cloudevents.CloudEvent;
import io.cloudevents.core.builder.CloudEventBuilder;
import io.cloudevents.core.data.PojoCloudEventData;
import org.bson.Document;

import java.io.IOException;
import java.net.URI;

import static java.time.ZoneOffset.UTC;
import static org.occurrent.functional.CheckedFunction.unchecked;
import static org.occurrent.time.TimeConversion.toLocalDateTime;
import static java.time.temporal.ChronoUnit.MILLIS;

public class MyCloudEventConverter implements CloudEventConverter<DomainEvent> {
    
    private final ObjectMapper objectMapper;
    
    public MyCloudEventConverter(ObjectMapper objectMapper) {
        this.objectMapper = objectMapper;
    } 

    @Override
    public CloudEvent toCloudEvent(DomainEvent e) {  
            // Convert the data in the domain event into a Document 
            Map<String, Object> eventData = convertDataInDomainEventToMap(e);
            return CloudEventBuilder.v1()
                    .withId(e.getEventId())
                    .withSource(URI.create("http://name"))
                    .withType(getCloudEventType(e))
                    .withTime(LocalDateTime.ofInstant(e.getDate().toInstant(), UTC).atOffset(UTC)
                                           .truncatedTo(MILLIS))
                    .withSubject(e.getName())
                    .withDataContentType("application/json")  
                    // Use the "eventData" map to create an instance of PojoCloudEventData.
                    // If an event store implementation doesn't know how to handle "Map" data,
                    // it'll call the higher-order function that converts the map into byte[] 
                    // (objectMapper::writeValueAsBytes), that it _has_ to understand.
                    // But since all Occurrent event stores currently knows how to handle maps, 
                    // the objectMapper::writeValueAsBytes method will never be called.
                    .withData(PojoCloudEventData.wrap(eventData, objectMapper::writeValueAsBytes))
                    .build();
    }
    
    @Override
    public DomainEvent toDomainEvent(CloudEvent cloudEvent) {
        CloudEventData cloudEventData = cloudEvent.getData();
        if (cloudEventData instanceof PojoCloudEventData && cloudEventData.getValue() instanceof Map) {
            Map<String, Object> eventData = ((PojoCloudEventData<Map<String, Object>>) cloudEventData).getValue();
            return convertToDomainEvent(cloudEvent, eventData);
        } else {
            return objectMapper.readValue(cloudEventData.toBytes(), DomainEvent.class); // try-catch omitted
        }
    }
    
    @Override
    public String getCloudEventType(Class<? extends T> type) {
        return type.getSimpleName();
    }
    
    private static Map<String, Object> convertDataInDomainEventToDocument(DomainEvent e) {
        // Convert the domain event into a Map                
        Map<String, Object> data = new HashMap<String, Object>();
        if (e instanceof GameStarted) {
           data.put("type", "GameStarted"); 
           // Put the rest of the values
        } else if (...) {
            // More events
        }
        return map;
    }

    private static DomainEvent convertToDomainEvent(CloudEvent cloudEvent, Map<String, Object> data) {
        // Re-construct the domain event instance from the cloud event and data
        switch ((String) data.get("type")) {
            case "GameStarted" -> // Convert map to GameStartedEvent 
                break;
            ...
        }
    }
}        

Application Service

Occurrent provides a generic application service that is a good starting point for most use cases. First add the module as a dependency to your project:

<dependency>
    <groupId>org.occurrent</groupId>
    <artifactId>occurrent-application-service-blocking</artifactId>
    <version>0.31.0</version>
</dependency>

compile 'org.occurrent:occurrent-application-service-blocking:0.31.0'

libraryDependencies += "org.occurrent" % "occurrent-application-service-blocking" % "0.31.0"

@Grab(group='org.occurrent', module='occurrent-application-service-blocking', version='0.31.0') 

[org.occurrent/occurrent-application-service-blocking "0.31.0"]

'org.occurrent:occurrent-application-service-blocking:jar:0.31.0'

<dependency org="org.occurrent" name="occurrent-application-service-blocking" rev="0.31.0" />

This module provides an interface, org.occurrent.application.service.blocking.ApplicationService, and a default implementation, org.occurrent.application.service.blocking.generic.GenericApplicationService. The GenericApplicationService takes an EventStore and a org.occurrent.application.converter.CloudEventConverter implementation as parameters. The latter is used to convert domain events to and from cloud events when loaded/written to the event store. There’s a default implementation that you may decide to use called, org.occurrent.application.converter.implementation.GenericCloudEventConverter available in the org.occurrent:occurrent-cloudevent-converter-generic module. You can see an example in the next section.

As of version 0.11.0, the GenericApplicationService also takes a RetryStrategy as an optional third parameter.
By default, the retry strategy uses exponential backoff starting with 100 ms and progressively go up to max 2 seconds wait time between each retry, if a WriteConditionNotFulfilledException is caught (see write condition docs). It will, again by default, only retry 5 times before giving up, rethrowing the original exception. You can override the default strategy by calling new GenericApplicationService(eventStore, cloudEventConverter, retryStrategy). Use new GenericApplicationService(eventStore, cloudEventConverter, RetryStrategy.none()) to disable retry. This is also useful if you want to use another retry library.

That retry runs wherever execute runs, so if you wrap the call in your own @Transactional it retries inside your transaction, where it cannot succeed. See Retry and Transactions.

Using the Application Service

Now you can instantiate the (blocking) GenericApplicationService:

EventStore eventStore = ..
CloudEventConverter<DomainEvent> cloudEventConverter = ..
ApplicationService<DomainEvent> applicationService = new GenericApplicationService<>(eventStore, cloudEventConverter);

val eventStore : EventStore = ..
val cloudEventConverter : CloudEventConverter<DomainEvent> = ..
val applicationService : ApplicationService<DomainEvent> = GenericApplicationService(eventStore, cloudEventConverter)

You’re now ready to use the generic application service in your application.

As an example let’s say you have a domain model with a method defined like this:

public class WordGuessingGame {
    public static List<DomainEvent> guessWord(List<DomainEvent> events, String guess) {
        // Implementation
    }    
}

You can call it using the application service:

applicationService.execute(gameId, events -> WordGuessingGame.guessWord(events, guess));

applicationService.execute(gameId) { events ->
    WordGuessingGame.guessWord(events, guess)
}

When you need more control over the execution, for example filtered stream reads or synchronous side effects, use ExecuteOptions. This is now the preferred API for these concerns.

applicationService.execute(
        gameId,
        ExecuteOptions.<DomainEvent>options()
                .filter(StreamReadFilter.type(GameWasStarted.class.getName()))
                .sideEffect(newEvents -> newEvents.forEach(this::publish)),
        events -> WordGuessingGame.guessWord(events, guess)
);

applicationService.execute(
    gameId,
    filter(StreamReadFilter.type(GameWasStarted::class.java.name)).sideEffect(
        { event: GameWasStarted -> publish(event) }
    )
) { events ->
    WordGuessingGame.guessWord(events, guess)
}

For EventStore capability details and supported implementations, see Stream Filtering.

Stream Filtering and Execute Options

At the ApplicationService level, stream filtering is exposed through ExecuteOptions. This is the preferred way to say both:

• which events should be read before command execution
• which synchronous side effects should run after a successful write

As of 0.20.0 the ApplicationService supports filtered reads and side effects by using the ExecuteOptions object.

org.occurrent.application.service.blocking.ExecuteOptions is the preferred way to configure blocking ApplicationService execution when you need:

• filtered stream reads before command execution
• synchronous post-write side effects
• both at the same time

A Java example:

WriteResult result = applicationService.execute(
        gameId,
        ExecuteOptions.<DomainEvent>options()
                .filter(StreamReadFilter.type(GameWasStarted.class.getName()))
                .sideEffect(newEvents -> newEvents.forEach(this::publish)),
        events -> WordGuessingGame.guessWord(events, guess)
);

val result = applicationService.execute(
    gameId,
    options().filter(StreamReadFilter.type(GameWasStarted::class.java.name)).sideEffect(
        { event: GameWasStarted -> publish(event) }
    )
) { events ->
    WordGuessingGame.guessWord(events, guess)
}

For EventStore support details, filtering semantics, and direct EventStore examples, see Stream Filtering.

Synchronous Side Effects

Use ExecuteOptions.sideEffect(...) for synchronous side effects going forward. This is now the preferred API for new code.

The lower-level SideEffect.executeSideEffect(...) helper (formerly PolicySideEffect.executePolicy(...)) still exists, but it is no longer the recommended primary approach. If you are documenting or writing new synchronous side-effect code, prefer ExecuteOptions.sideEffect(...).

A side effect is not the same as a synchronous subscription. A side effect is a closure you pass at each call site, whereas a synchronous subscription is declared once and reacts to every matching write, the same way an asynchronous subscription does. Reach for a synchronous subscription when the reaction should be declared once and decoupled from the call, or should commit atomically with the write through a TransactionExecutor.

Java Examples

applicationService.execute(
        gameId,
        ExecuteOptions.<DomainEvent>options()
                .sideEffect(newEvents -> newEvents
                        .filter(event -> event instanceof GameWasStarted)
                        .findFirst()
                        .map(GameWasStarted.class::cast)
                        .ifPresent(registerOngoingGame::registerGameAsOngoingWhenGameWasStarted)),
        events -> WordGuessingGame.guessWord(events, guess)
);

applicationService.execute(
    gameId,
    sideEffect { event: GameWasStarted ->
        registerOngoingGame.registerGameAsOngoingWhenGameWasStarted(event)
    }
) { events ->
    WordGuessingGame.guessWord(events, guess)
}

Kotlin Examples

For Kotlin, the domain function is a (List<DomainEvent>) -> List<DomainEvent> lambda passed straight to execute. You can start from either options() or the top-level helper functions:

applicationService.execute(
    gameId,
    sideEffect(
        { event: GameWasStarted -> registerOngoingGame.registerGameAsOngoingWhenGameWasStarted(event) }
    )
) { events ->
    WordGuessingGame.guessWord(events, guess)
}

applicationService.execute(
    gameId,
    options().filter(ExecuteFilters.type<GameWasStarted>()).sideEffect(
        { event: GameWasStarted -> registerOngoingGame.registerGameAsOngoingWhenGameWasStarted(event) }
    )
) { events ->
    WordGuessingGame.guessWord(events, guess)
}

Both Kotlin forms above invoke the callback once per matching event. When you want the whole batch of matching events in a single call instead, use sideEffectOnList, which hands you a List of them (it filters to the event type the same way):

applicationService.execute(
    gameId,
    options().sideEffectOnList { startedGames: List<GameWasStarted> ->
        registerOngoingGame.registerAll(startedGames)
    }
) { events ->
    WordGuessingGame.guessWord(events, guess)
}

It is sideEffectOnList rather than sideEffect because ExecuteOptions already has a Java sideEffect(Consumer<List<E>>) method, and a Kotlin extension of the same name would be shadowed by it. In Java, call that sideEffect(...) method directly with the whole list.

When Not to Use Filtering

Filtering is a read optimization, not a correctness feature. Do not filter away events that are needed to enforce invariants, rebuild state correctly, or make valid domain decisions. If the command depends on the whole stream, read the whole stream.

Application Service Side-Effects

The GenericApplicationService supports executing side-effects after the events returned from the domain model have been written to the event store. This is useful if you need to update a view synchronously after a successful write. To perform side-effects asynchronously, use a subscription instead.

For synchronous side effects, use ExecuteOptions.sideEffect(...) going forward. This is now the preferred API.

As an example, consider that you want to synchronously register a game as ongoing when it is started. It may be defined like this:

public class RegisterOngoingGame {
    private final DatabaseApi someDatabaseApi;

    public RegisterOngoingGame(DatabaseApi someDatabaseApi) {
        this.someDatabaseApi = someDatabaseApi;
    }

    public void registerGameAsOngoingWhenGameWasStarted(GameWasStarted event) {
        // Add the id of the game started event to a set to handle duplicates and idempotency.
        someDatabaseApi.addToSet("ongoingGames", Map.of("gameId", event.gameId(), "date", event.getDate()));
    }
}

class RegisterOngoingGame(private val someDatabaseApi : DatabaseApi) {
    fun registerGameAsOngoingWhenGameWasStarted(event : GameWasStarted) {
        // Add the id of the game started event to a set to handle duplicates and idempotency.
        someDatabaseApi.addToSet("ongoingGames", Map.of("gameId", event.gameId(), "date", event.getDate()));
    }
}

Now that we have the code that registers ongoing games, we can call it from the application service like this:

RegisterOngoingGame registerOngoingGame = ..
applicationService.execute(
        gameId,
        ExecuteOptions.<DomainEvent>options()
                .sideEffect(newEvents -> newEvents
                        .filter(event -> event instanceof GameWasStarted)
                        .findFirst()
                        .map(GameWasStarted.class::cast)
                        .ifPresent(registerOngoingGame::registerGameAsOngoingWhenGameWasStarted)),
        events -> WordGuessingGame.guessWord(events, guess)
);

val registerOngoingGame : RegisterOngoingGame = ..
applicationService.execute(
    gameId,
    sideEffect { event: GameWasStarted ->
        registerOngoingGame.registerGameAsOngoingWhenGameWasStarted(event)
    }
) { events ->
    WordGuessingGame.guessWord(events, guess)
}

Voila! Now registerGameAsOngoingWhenGameWasStarted will be called after the events returned from WordGuessingGame.guessWord(..) are written to the event store.

The lower-level SideEffect.executeSideEffect(...) helper (formerly PolicySideEffect.executePolicy(...)) still exists, but it is no longer the preferred primary approach for new code. If you already use it, you can keep doing so, but new examples and new synchronous side-effect code should prefer ExecuteOptions.sideEffect(...).

Application Service Transactional Side-Effects

In the example above, writing the events to the event store and executing synchronous side effects is not an atomic operation. If your app crashes after one side effect has run, but before the rest of the work completes, you will need to handle idempotency. But if your side effects write data to the same database as the event store, you can make use of transactions to write everything atomically. This is very easy if you’re using a Spring EventStore. What you need to do is to wrap the ApplicationService provided by Occurrent in your own application service, something like this:

@Service
public class CustomApplicationServiceImpl {
	private final GenericApplicationService<DomainEvent> occurrentApplicationService;

	public CustomApplicationService(GenericApplicationService<DomainEvent> occurrentApplicationService) {
		this.occurrentApplicationService = occurrentApplicationService;
	}

	@Transactional
    public WriteResult execute(String gameId,
                               ExecuteOptions<DomainEvent> executeOptions,
                               Function<List<DomainEvent>, List<DomainEvent>> functionThatCallsDomainModel) {
		return occurrentApplicationService.execute(gameId, executeOptions, functionThatCallsDomainModel);
    }
}

@Service
class CustomApplicationServiceImpl(private val occurrentApplicationService: GenericApplicationService<DomainEvent>) {

    @Transactional
    fun execute(
        gameId: String,
        executeOptions: ExecuteOptions<DomainEvent>,
        functionThatCallsDomainModel: Function<List<DomainEvent>, List<DomainEvent>>
    ): WriteResult {
        return occurrentApplicationService.execute(gameId, executeOptions, functionThatCallsDomainModel)
    }
}

Given that you’ve defined a MongoTransactionManager in your Spring Boot configuration (and use this when creating your event store instance) the side effects and events are written atomically in the same transaction!

Application Service Kotlin Extensions

If you’re using Kotlin, the domain function you pass to execute is a (List<E>) -> List<E> lambda that binds directly to the Java execute(..., Function<List<E>, List<E>>) member.

Occurrent provides Kotlin extensions for the surrounding options in the application service module:

<dependency>
    <groupId>org.occurrent</groupId>
    <artifactId>occurrent-application-service-blocking</artifactId>
    <version>0.31.0</version>
</dependency>

compile 'org.occurrent:occurrent-application-service-blocking:0.31.0'

libraryDependencies += "org.occurrent" % "occurrent-application-service-blocking" % "0.31.0"

@Grab(group='org.occurrent', module='occurrent-application-service-blocking', version='0.31.0') 

[org.occurrent/occurrent-application-service-blocking "0.31.0"]

'org.occurrent:occurrent-application-service-blocking:jar:0.31.0'

<dependency org="org.occurrent" name="occurrent-application-service-blocking" rev="0.31.0" />

Use these:

• options() when you want to start an ExecuteOptions chain explicitly
• filter(...) when you want to start directly with a StreamReadFilter or ExecuteFilter
• sideEffect(...) when you want to configure synchronous side effects
• ExecuteFilters.type(...) / ExecuteFilters.includeTypes(...) / ExecuteFilters.excludeTypes(...) when you want typed filters based on domain event classes
• sideEffectOnList(...) when the side effect itself wants a filtered collection view

For example:

applicationService.execute(gameId) { events ->
    WordGuessingGame.guessWord(events, guess)
}

applicationService.execute(
    gameId,
    sideEffect { event: GameWasStarted ->
        registerOngoingGame.registerGameAsOngoingWhenGameWasStarted(event)
    }
) { events ->
    WordGuessingGame.guessWord(events, guess)
}

applicationService.execute(
    gameId,
    options().filter(ExecuteFilters.type<GameWasStarted>()).sideEffect(
        { event: GameWasStarted -> registerOngoingGame.registerGameAsOngoingWhenGameWasStarted(event) }
    )
) { events ->
    WordGuessingGame.guessWord(events, guess)
}

For synchronous side effects, prefer sideEffect(...) or options().sideEffect(...) rather than the older policy-style helpers.

Command Dispatch

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 lambda dispatcher gets. Override it when your commands all target one stream or one decider and you can write them together, for example inside a single transaction.

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, it is worth overriding dispatchAll to make the batch atomic.

Overriding it does not make dispatch exactly-once, and Occurrent does not promise that. It closes the window for the one dispatcher that can, and the contract stays at-least-once either way.

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

compile 'org.occurrent:occurrent-command-dispatch:0.31.0'

libraryDependencies += "org.occurrent" % "occurrent-command-dispatch" % "0.31.0"

@Grab(group='org.occurrent', module='occurrent-command-dispatch', version='0.31.0') 

[org.occurrent/occurrent-command-dispatch "0.31.0"]

'org.occurrent:occurrent-command-dispatch:jar:0.31.0'

<dependency org="org.occurrent" name="occurrent-command-dispatch" rev="0.31.0" />

ApplicationService is the write engine underneath. It takes a stream id and a decider, reads the stream, decides what to append, and appends it. 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

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(@TargetStreamId String orderId, String productId, int quantity) {
}

Decider<PlaceOrder, OrderState, OrderEvent> placeOrderDecider = ...;
DeciderApplicationService<OrderEvent> applicationService = ...;
StreamIdResolver<PlaceOrder> streamIdResolver = new AnnotationStreamIdResolver<>();

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

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

data class PlaceOrder(
    @get:TargetStreamId val orderId: String,
    val productId: String,
    val quantity: Int
)

val placeOrderDecider: Decider<PlaceOrder, OrderState, OrderEvent> = ...
val applicationService: DeciderApplicationService<OrderEvent> = ...
val streamIdResolver: StreamIdResolver<PlaceOrder> = AnnotationStreamIdResolver()

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

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

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.31.0</version>
</dependency>

compile 'org.occurrent:occurrent-command-dispatch-dcb:0.31.0'

libraryDependencies += "org.occurrent" % "occurrent-command-dispatch-dcb" % "0.31.0"

@Grab(group='org.occurrent', module='occurrent-command-dispatch-dcb', version='0.31.0') 

[org.occurrent/occurrent-command-dispatch-dcb "0.31.0"]

'org.occurrent:occurrent-command-dispatch-dcb:jar:0.31.0'

<dependency org="org.occurrent" name="occurrent-command-dispatch-dcb" rev="0.31.0" />

DcbDecider<EnrollStudent, EnrollmentState, DomainEvent> enrollmentDecider = ...;
DcbDeciderApplicationService<DomainEvent> applicationService = ...;

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

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

val enrollmentDcbDecider: DcbDecider<EnrollStudent, EnrollmentState, DomainEvent> = ...
val applicationService: DcbDeciderApplicationService<DomainEvent> = ...

val dispatcher: CommandDispatcher<EnrollStudent> = DcbCommandDispatchers.decider(applicationService, enrollmentDcbDecider)

dispatcher.dispatch(EnrollStudent(courseId, studentId))

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()