Change Streams de MongoDB como un Flow de Kotlin

Un change stream de MongoDB es un cursor del lado del servidor que emite un documento cada vez que los datos de una coleccion (o base de datos, o todo el despliegue) se insertan, actualizan, reemplazan o eliminan. Internamente sigue el oplog del replica set, por lo que los change streams requieren un replica set o un cluster fragmentado — incluso una configuracion de un solo nodo debe arrancarse como replica set. Como la fuente es un cursor abierto que produce valores a lo largo del tiempo, encaja casi a la perfeccion con un Flow de Kotlin: un flujo frio y asincrono de valores que solo trabaja cuando alguien lo recolecta.

Con Spring Data MongoDB Reactive, el punto de entrada idiomatico es ReactiveMongoTemplate.changeStream(...). Devuelve un Flux<ChangeStreamEvent<T>> de Reactor, y el puente kotlinx-coroutines-reactive te permite convertir ese Publisher en un Flow con .asFlow(). Construyes la peticion con ChangeStreamOptions, donde puedes adjuntar un pipeline de agregacion para filtrar en el servidor (por ejemplo, solo operaciones 'insert' y 'update') y elegir cuanto del documento recibes. Para las actualizaciones, solicita FullDocument.UPDATE_LOOKUP para que cada evento traiga el documento actual completo en lugar de solo los campos cambiados.

Cada emision es un ChangeStreamEvent<T>. Los dos miembros que mas usaras son event.operationType (un enum OperationType: INSERT, UPDATE, REPLACE, DELETE, INVALIDATE, etc.) y event.body, el objeto de dominio deserializado — que es null en los borrados y en las actualizaciones cuando no solicitaste un lookup del documento completo. El evento crudo del driver tambien esta disponible via event.raw, que expone el resume token, el cluster time y el documentKey (el _id del documento afectado), que es el unico campo que siempre obtienes incluso en un borrado.

Esta nulabilidad es justo la razon por la que mapNotNull es el operador estrella aqui. En lugar de recolectar eventos crudos y esparcir comprobaciones de null por tu manejador, mapeas cada ChangeStreamEvent a una senal de dominio limpia y descartas en silencio cualquier cosa que no produzca un valor util. mapNotNull mantiene fuera del pipeline los borrados sin cuerpo y cualquier evento que decidas ignorar, de modo que los recolectores solo ven elementos bien formados. Combinalo con onEach para registrar y catch para resiliencia.

El hilo importa porque la deserializacion y cualquier transformacion por evento no deberian ejecutarse en los hilos del event-loop reactivo que el driver usa para bombear el stream. flowOn(Dispatchers.Default) traslada el trabajo de aguas arriba (todo lo declarado encima de la llamada a flowOn) a un dispatcher de trabajadores, dejando libres los hilos de red para seguir leyendo de MongoDB. Recuerda que flowOn solo afecta a los operadores aguas arriba de el; el recolector terminal sigue ejecutandose en el contexto de la corrutina que lanzo la recoleccion, asi que coloca flowOn deliberadamente.

La contrapresion es la parte sutil de construir un feed en vivo. Un change stream puede dispararse — una importacion masiva o una escritura con fan-out puede producir miles de eventos mas rapido de lo que un consumidor lento (un cliente SSE, un websocket, una llamada HTTP) puede manejar. El Flow simple es secuencial y suspende al productor hasta que el recolector esta listo, lo que da contrapresion natural pero puede atascar las lecturas del oplog. Cuando necesitas desacoplar las velocidades de productor y consumidor, inserta un buffer(...) con capacidad explicita y una estrategia de desbordamiento, o usa conflate() cuando solo importa el ultimo estado por elemento y los valores intermedios pueden descartarse.

Elige la politica de desbordamiento segun tu dominio. BufferOverflow.SUSPEND (la predeterminada) es segura pero aplica contrapresion a la fuente; DROP_OLDEST es ideal para un feed de 'ultimo estado del dashboard' donde los eventos viejos no valen nada; DROP_LATEST protege una cola de tamano fijo prefiriendo el trabajo mas antiguo en curso. Para una UI que muestra el valor actual de una entidad, conflate() suele ser la opcion mas limpia — es esencialmente un muestreador de tasa ilimitada que siempre reenvia la emision mas fresca. Dimensiona siempre los buffers conscientemente: un buffer ilimitado simplemente cambia un lector atascado por un crecimiento ilimitado de memoria.

Por ultimo, trata el stream como algo de larga vida y falible. Cortes de red, cambios de primario y timeouts por inactividad terminaran el cursor. El codigo de produccion persiste el ultimo resume token y lo pasa via ChangeStreamOptions.resumeAfter(...) / resumeToken(...) para que un reinicio continue exactamente donde lo dejo en lugar de reproducir o perder eventos. Envuelve el flow con retryWhen para errores transitorios y recuerda que todo el pipeline es amigable con la concurrencia estructurada: cancela la corrutina recolectora (por ejemplo cuando un cliente SSE se desconecta) y el cursor subyacente de MongoDB se cierra por ti.

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