MongoDB Change Streams as a Kotlin Flow

A MongoDB change stream is a server-side cursor that emits a document every time data in a collection (or database, or whole deployment) is inserted, updated, replaced, or deleted. Under the hood it tails the replica set's oplog, so change streams require a replica set or sharded cluster — even a single-node setup must be started as a replica set. Because the source is an open-ended cursor that produces values over time, it maps almost perfectly onto a Kotlin Flow: a cold, asynchronous stream of values that only does work once someone collects it.

With Spring Data MongoDB Reactive, the idiomatic entry point is ReactiveMongoTemplate.changeStream(...). It returns a Reactor Flux<ChangeStreamEvent<T>>, and the kotlinx-coroutines-reactive bridge lets you convert that Publisher to a Flow with .asFlow(). You build the request with ChangeStreamOptions, where you can attach an aggregation pipeline to filter server-side (for example, only 'insert' and 'update' operations) and choose how much of the document you receive. For updates, request FullDocument.UPDATE_LOOKUP so each event carries the complete current document rather than just the changed fields.

Each emission is a ChangeStreamEvent<T>. The two members you reach for most are event.operationType (an OperationType enum: INSERT, UPDATE, REPLACE, DELETE, INVALIDATE, and so on) and event.body, the deserialized domain object — which is null for deletes and for updates when you did not request a full-document lookup. The raw driver event is also available via event.raw, exposing the resume token, the cluster time, and the documentKey (the _id of the affected document), which is the one field you always get even on a delete.

This nullability is exactly why mapNotNull is the workhorse operator here. Rather than collecting raw events and scattering null checks through your handler, you map each ChangeStreamEvent to a clean domain signal and silently drop anything that does not yield a usable value. mapNotNull keeps the deletes-without-bodies and any events you choose to ignore out of the downstream pipeline, so collectors only ever see fully-formed items. Pair it with onEach for logging and catch for resilience.

Threading matters because deserialization and any per-event transformation should not run on the reactive event-loop threads that the driver uses to pump the stream. flowOn(Dispatchers.Default) moves the upstream work (everything declared above the flowOn call) onto a worker dispatcher, leaving the network threads free to keep reading from MongoDB. Remember that flowOn only affects the operators upstream of it; the terminal collector still runs in whatever coroutine context launched the collection, so place flowOn deliberately.

Backpressure is the subtle part of building a live feed. A change stream can burst — a bulk import or a fan-out write can produce thousands of events faster than a slow consumer (an SSE client, a websocket, a downstream HTTP call) can handle them. Plain Flow is sequential and suspends the producer until the collector is ready, which gives you natural backpressure but can stall reads from the oplog. When you need to decouple producer and consumer speeds, insert a buffer(...) with an explicit capacity and an overflow strategy, or use conflate() when only the latest state per item matters and intermediate values can be dropped.

Choose the overflow policy to match your domain. BufferOverflow.SUSPEND (the default) is safe but back-pressures the source; DROP_OLDEST is ideal for a 'latest dashboard state' feed where stale events are worthless; DROP_LATEST protects a fixed-size queue while preferring older, in-flight work. For a UI that shows the current value of an entity, conflate() is often the cleanest choice — it is essentially an unbounded-rate sampler that always forwards the freshest emission. Always size buffers consciously: an unbounded buffer simply trades a stalled reader for unbounded memory growth.

Finally, treat the stream as long-lived and fallible. Network blips, primary step-downs, and idle timeouts will terminate the cursor. Production code persists the last resume token and passes it via ChangeStreamOptions.resumeAfter(...) / resumeToken(...) so a restart continues exactly where it left off instead of replaying or losing events. Wrap the flow with retryWhen for transient errors and remember the whole pipeline is structured-concurrency friendly: cancel the collecting coroutine (for example when an SSE client disconnects) and the underlying MongoDB cursor is closed for you.

import com.mongodb.client.model.changestream.OperationType
import com.mongodb.client.model.changestream.FullDocument
import org.springframework.data.mongodb.core.ReactiveMongoTemplate
import org.springframework.data.mongodb.core.ChangeStreamOptions
import org.springframework.data.mongodb.core.aggregation.Aggregation
import org.springframework.data.mongodb.core.aggregation.Aggregation.match
import org.springframework.data.mongodb.core.query.Criteria.where
import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.flow.*
import kotlinx.coroutines.reactive.asFlow

sealed interface OrderEvent {
    data class Upserted(val order: Order) : OrderEvent
    data class Removed(val id: String) : OrderEvent
}

class OrderFeed(private val template: ReactiveMongoTemplate) {

    fun stream(): Flow<OrderEvent> {
        val options = ChangeStreamOptions.builder()
            .fullDocumentLookup(FullDocument.UPDATE_LOOKUP)
            .filter(Aggregation.newAggregation(
                match(where("operationType").`in`("insert", "update", "replace", "delete"))
            ))
            .build()

        return template.changeStream("orders", options, Order::class.java)
            .asFlow()
            .mapNotNull { event ->
                when (event.operationType) {
                    OperationType.INSERT,
                    OperationType.UPDATE,
                    OperationType.REPLACE -> event.body?.let { OrderEvent.Upserted(it) }
                    OperationType.DELETE -> event.raw?.documentKey
                        ?.getObjectId("_id")?.value?.toHexString()
                        ?.let { OrderEvent.Removed(it) }
                    else -> null // INVALIDATE, DROP, etc. are ignored
                }
            }
            .flowOn(Dispatchers.Default)
    }
}

import org.springframework.http.MediaType
import org.springframework.http.codec.ServerSentEvent
import org.springframework.web.bind.annotation.GetMapping
import org.springframework.web.bind.annotation.RestController
import kotlinx.coroutines.channels.BufferOverflow
import kotlinx.coroutines.flow.*

@RestController
class OrderSseController(private val feed: OrderFeed) {

    @GetMapping("/orders/live", produces = [MediaType.TEXT_EVENT_STREAM_VALUE])
    fun live(): Flow<ServerSentEvent<OrderEvent>> =
        feed.stream()
            .buffer(capacity = 256, onBufferOverflow = BufferOverflow.DROP_OLDEST)
            .onEach { /* metrics, logging */ }
            .map { event -> ServerSentEvent.builder(event).event("order").build() }
            .catch { e -> emit(ServerSentEvent.builder<OrderEvent>().event("error").comment(e.message).build()) }
}

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*

fun main() = runBlocking {
    val rawEvents: Flow<Pair<String, Int?>> = flow {
        emit("insert" to 1)
        emit("update" to 2)
        emit("delete" to null) // no body, like a Mongo delete
        emit("update" to 3)
    }

    rawEvents
        .mapNotNull { (op, body) -> body?.let { "$op -> order #$it" } }
        .flowOn(Dispatchers.Default)
        .conflate()
        .collect { signal ->
            delay(50) // simulate a slow consumer
            println(signal)
        }
}