Change Streams

What Morphium Offers

MongoDB Change Streams notify listening applications in (near) realtime whenever a document in a watched collection is inserted, updated, replaced, or deleted — no polling required. Morphium exposes this via ChangeStreamMonitor, which manages the underlying watch() aggregation pipeline, resume tokens, and reconnection for you.

The Challenge

Polling a collection for changes wastes cycles and adds latency. The MongoDB driver's raw change stream API requires managing cursors, resume tokens across reconnects, and translating BSON change documents by hand. ChangeStreamMonitor handles the cursor lifecycle and hands you plain ChangeStreamEvent objects instead.

Morphium Features Used

Prerequisites & Key Concepts

Replica set required. Change streams are built on MongoDB's replication oplog, so they require a replica set (or sharded cluster) — a standalone MongoDB instance cannot open one. Dev Services starts a single-node replica set by default, so this works out of the box locally.
Resume tokens let a change stream reconnect after a network blip or application restart and continue exactly where it left off, without missing or duplicating events. ChangeStreamMonitor tracks this automatically.
No entity-side annotation. Unlike most Morphium features shown elsewhere in this showcase, change streams are entirely configured on the watching side. Any collection can be watched, regardless of how its entity is mapped.
Not supported on Azure CosmosDB. CosmosDB's MongoDB API does not implement change streams the same way; this feature is MongoDB/PoppyDB-only.

About Change Streams

A change stream is a standing MongoDB aggregation pipeline ($changeStream) that the server keeps open and pushes new events into as they happen, rather than the client repeatedly asking "did anything change?". This makes realtime features — live dashboards, cache invalidation, audit trails, cross-service notifications — possible without a separate message broker, as long as all consumers are already talking to the same MongoDB deployment.

Each event carries an operationType (insert, update, replace, delete, ...), the changed document's key, and — for inserts/updates/replaces, if requested — the full resulting document. Update events additionally carry an updateDescription with exactly which fields changed and which were removed, so consumers don't have to diff documents themselves.

Entity Source Code

LiveNote.java Java

import de.caluga.morphium.annotations.CreationTime;
import de.caluga.morphium.annotations.Entity;
import de.caluga.morphium.annotations.Id;
import de.caluga.morphium.annotations.LastChange;
import de.caluga.morphium.driver.MorphiumId;

@Entity(collectionName = "live_notes")1
public class LiveNote {

    @Id
    private MorphiumId id;

    private String title;
    private String content;
    private String author;

    @CreationTime2
    private LocalDateTime createdAt;

    @LastChange3
    private LocalDateTime updatedAt;
}

1 No change-stream-specific annotation is required — this is an ordinary Morphium entity, watched from the outside.

2 @CreationTime gives a reliable "created at" reference independent of the change stream's own clock.

3 @LastChange is a convenient cross-check against the event's own clusterTime.

Watching for Changes

LiveNoteChangeStreamService.java — starting the monitor Java

import de.caluga.morphium.changestream.ChangeStreamMonitor;
import de.caluga.morphium.changestream.ChangeStreamEvent;

ChangeStreamMonitor monitor = new ChangeStreamMonitor(morphium, "live_notes");1
monitor.addListener(evt -> {2
    ChangeEventView view = ChangeEventView.from(evt);
    broadcastToSubscribers(view); // forwarded to every connected SSE client
    return true; // keep listening
});
morphium.queueTask(monitor);3

1 The monitor is scoped to a single collection name — here live_notes, matching LiveNote's @Entity(collectionName=...).

2 The listener receives every insert/update/replace/delete on that collection as a ChangeStreamEvent, converted here into the demo's flattened ChangeEventView.

3 The monitor runs on Morphium's background task executor for the lifetime of the application (started on StartupEvent, stopped on ShutdownEvent in this showcase).

Delivering events to the browser (Server-Sent Events) Java

@GET
@Path("/events")
@Produces(MediaType.SERVER_SENT_EVENTS)
public void streamEvents(@Context SseEventSink sink, @Context Sse sse) {1
    for (ChangeEventView evt : changeStreamService.recentHistory()) {2
        sink.send(sse.newEventBuilder().name("change").data(evt).build());
    }
    changeStreamService.subscribe(evt ->
        sink.send(sse.newEventBuilder().name("change").data(evt).build()));3
}

1 Each connecting browser tab opens its own SseEventSink — this is the same JAX-RS SSE mechanism used by the Messaging feature.

2 Recent history is replayed first, so a tab opened after some activity already has context.

3 The stream is driven purely by what MongoDB's change stream reports — it would react identically to writes made by a completely different process connected to the same database.