Skip to content
OccurrentOccurrent

MongoDB event stores

Modules: occurrent-eventstore-mongodb-native, occurrent-eventstore-mongodb-spring-blocking, occurrent-eventstore-mongodb-spring-reactor

There are three different MongoDB EventStore implementations to choose from:

EventStore with MongoDB Native Driver

Section titled “EventStore with MongoDB Native Driver”

What is it?

Section titled “What is it?”

An EventStore implementation that uses the “native” Java MongoDB synchronous driver (see website) to read and write CloudEvents to MongoDB.

When to use?

Section titled “When to use?”

Use when you don’t need Spring support and want to use MongoDB as the underlying datastore.

Dependencies

Section titled “Dependencies”
<dependency>
    <groupId>org.occurrent</groupId>
    <artifactId>occurrent-eventstore-mongodb-native</artifactId>
    <version>0.34.0</version>
</dependency>

Getting Started

Section titled “Getting Started”

Once you’ve imported the dependencies you create a new instance of org.occurrent.eventstore.mongodb.nativedriver.MongoEventStore. It takes four arguments, a MongoClient, the “database” and “event collection “that the EventStore will use to store events as well as an org.occurrent.eventstore.mongodb.nativedriver.EventStoreConfig.

For example:

MongoClient mongoClient = MongoClients.create("mongodb://localhost:27017");
// How the CloudEvent "time" property will be serialized in MongoDB! !!Important!!
EventStoreConfig config = new EventStoreConfig(TimeRepresentation.RFC_3339_STRING);
String mongoDatabase = "database";
String mongoEventCollection = "events";


MongoEventStore eventStore = new MongoEventStore(mongoClient, mongoDatabase, mongoEventCollection, config);

This implementation also supports filtered stream reads via org.occurrent.eventstore.api.blocking.ReadEventStreamWithFilter. See Stream Filtering for when and how to use it.

Now you can start reading and writing events to the EventStore:

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")
    );
}

Examples

Section titled “Examples”
NameDescription
Number Guessing GameA simple game implemented using a pure domain model and stores events in MongoDB using MongoEventStore. It also generates integration events and publishes these to RabbitMQ.
UnoA port of FsUno, a classic card game. Stores events in MongoDB using MongoEventStore.

EventStore with Spring MongoTemplate (Blocking)

Section titled “EventStore with Spring MongoTemplate (Blocking)”

What is it?

Section titled “What is it?”

An implementation that uses Spring’s MongoTemplate to read and write events to/from MongoDB.

When to use?

Section titled “When to use?”

If you’re already using Spring and you don’t need reactive support then this is a good choice. You can make use of the @Transactional annotation to write events and views in the same transaction (but make sure you understand what you’re doing before attempting this).

Dependencies

Section titled “Dependencies”
<dependency>
    <groupId>org.occurrent</groupId>
    <artifactId>occurrent-eventstore-mongodb-spring-blocking</artifactId>
    <version>0.34.0</version>
</dependency>

Getting Started

Section titled “Getting Started”

Once you’ve imported the dependencies you create a new instance of org.occurrent.eventstore.mongodb.spring.blocking.SpringMongoEventStore. It takes two arguments, a MongoTemplate and an org.occurrent.eventstore.mongodb.spring.blocking.EventStoreConfig.

For example:

MongoClient mongoClient = MongoClients.create("mongodb://localhost:27017");
MongoTransactionManager mongoTransactionManager = new MongoTransactionManager(new SimpleMongoClientDatabaseFactory(mongoClient, "database"));


MongoTemplate mongoTemplate = new MongoTemplate(mongoClient, "database");
EventStoreConfig eventStoreConfig = new EventStoreConfig.Builder()
                                                         // The collection where all events will be stored
                                                        .eventStoreCollectionName("events")
                                                        .transactionConfig(mongoTransactionManager)
                                                        // How the CloudEvent "time" property will be serialized in MongoDB! !!Important!!
                                                        .timeRepresentation(TimeRepresentation.RFC_3339_STRING)
                                                        .build();


SpringMongoEventStore eventStore = new SpringMongoEventStore(mongoTemplate, eventStoreConfig);

This implementation also supports filtered stream reads via org.occurrent.eventstore.api.blocking.ReadEventStreamWithFilter. See Stream Filtering for when and how to use it.

Now you can start reading and writing events to the EventStore:

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")
    );
}

Examples

Section titled “Examples”
NameDescription
Number Guessing GameA simple game implemented using a pure domain model and stores events in MongoDB using SpringMongoEventStore and Spring Boot. It also generates integration events and publishes these to RabbitMQ.
Word Guessing GameSimilar to the “Number Guessing Game” but more advanced, leveraging several Occurrent features such as CQRS, queries, and transactional projections. Implemented using a pure domain model and stores events in MongoDB using SpringMongoEventStore and Spring Boot.
UnoA port of FsUno, a classic card game. Implemented using a pure domain model and stores events in MongoDB using SpringMongoEventStore and Spring Boot.
Subscription ViewAn example showing how to create a subscription that listens to certain events stored in the EventStore and updates a view/projection from these events.
Transactional ViewAn example showing how to combine writing events to the SpringMongoEventStore and update a view transactionally using the @Transactional annotation.
Custom Aggregation ViewExample demonstrating that you can query the SpringMongoEventStore using custom MongoDB aggregations.

EventStore with Spring ReactiveMongoTemplate (Reactive)

Section titled “EventStore with Spring ReactiveMongoTemplate (Reactive)”

What is it?

Section titled “What is it?”

An implementation that uses Spring’s ReactiveMongoTemplate to read and write events to/from MongoDB.

When to use?

Section titled “When to use?”

If you’re already using Spring and want to use the reactive driver (project reactor) then this is a good choice. It uses the ReactiveMongoTemplate to write events to MongoDB. You can make use of the @Transactional annotation to write events and views in the same tx (but make sure that you understand what you’re doing before attempting this).

Dependencies

Section titled “Dependencies”
<dependency>
    <groupId>org.occurrent</groupId>
    <artifactId>occurrent-eventstore-mongodb-spring-reactor</artifactId>
    <version>0.34.0</version>
</dependency>

Getting Started

Section titled “Getting Started”

Once you’ve imported the dependencies you create a new instance of org.occurrent.eventstore.mongodb.spring.reactor.ReactorMongoEventStore. It takes two arguments, a ReactiveMongoTemplate and an org.occurrent.eventstore.mongodb.spring.reactor.EventStoreConfig.

For example:

MongoClient mongoClient = MongoClients.create("mongodb://localhost:27017");
MongoTransactionManager mongoTransactionManager = new MongoTransactionManager(new SimpleMongoClientDatabaseFactory(mongoClient, "database"));


MongoTemplate mongoTemplate = new MongoTemplate(mongoClient, "database");
EventStoreConfig eventStoreConfig = new EventStoreConfig.Builder()
                                                         // The collection where all events will be stored
                                                        .eventStoreCollectionName("events")
                                                        .transactionConfig(mongoTransactionManager)
                                                        // How the CloudEvent "time" property will be serialized in MongoDB! !!Important!!
                                                        .timeRepresentation(TimeRepresentation.RFC_3339_STRING)
                                                        .build();


ReactorMongoEventStore eventStore = new ReactorMongoEventStore(mongoTemplate, eventStoreConfig);

This implementation also supports filtered stream reads via org.occurrent.eventstore.api.reactor.ReadEventStreamWithFilter. See Stream Filtering for when and how to use it.

Now you can start reading and writing events to the EventStore:

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
Mono<Void> mono = eventStore.write("streamId", Flux.just(event));


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


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

Examples

Section titled “Examples”
NameDescription
Custom Aggregation ViewExample demonstrating that you can query a MongoDB event store using custom MongoDB aggregations. It uses the blocking SpringMongoEventStore. There is no reactive version of this example.