MongoDB Reactivo con Corrutinas

Spring Data MongoDB ofrece dos pilas. La clásica y bloqueante (MongoTemplate, MongoRepository) aparca un hilo en cada viaje a la base de datos; la reactiva no. Por debajo, el driver reactivo se comunica con MongoDB mediante E/S no bloqueante y expone los resultados como tipos de Project Reactor: Mono<T> para cero-o-uno y Flux<T> para cero-o-muchos. Para activarla añades la dependencia spring-boot-starter-data-mongodb-reactive y dejas que Spring Boot autoconfigure un ReactiveMongoTemplate y los repositorios reactivos. La recompensa es el rendimiento: un puñado de hilos del bucle de eventos puede atender miles de consultas concurrentes, justo el modelo sobre el que se construye WebFlux.

En Kotlin casi nunca quieres consumir Mono y Flux directamente. El puente kotlinx-coroutines-reactor te permite esperar un Mono con .awaitSingle()/.awaitSingleOrNull() y recolectar un Flux como un Flow de Kotlin con .asFlow(). Spring Data va un paso más allá: CoroutineCrudRepository expone sus métodos CRUD como funciones suspend y devuelve Flow para las colecciones, así que tu capa de datos se lee como Kotlin secuencial normal mientras permanece totalmente no bloqueante. La regla práctica es simple: un único resultado es una suspend fun que devuelve T?, un flujo de resultados es una fun que devuelve Flow<T>, y nunca bloqueas para obtener ninguno.

Los documentos se mapean limpiamente a data classes de Kotlin. Anota la clase con @Document("coleccion") y marca la clave primaria con @Id; un val id: String? = null inmutable es idiomático porque MongoDB genera el ObjectId al insertar y Spring lo rellena en la copia devuelta. Mantén el resto de campos como vals: las data classes te dan equals, hashCode, copy y desestructuración gratis, y el driver las instancia a través del constructor primario, por lo que los valores nulos por defecto sirven como una estrategia ligera de evolución del esquema. Estas anotaciones de Spring Data apuntan al campo, así que en una propiedad del constructor primario las escribes sin prefijo de use-site. Usa @Field para renombrar una propiedad en almacenamiento y @Indexed para declarar un índice.

Definir un repositorio es casi pura declaración. Extiende CoroutineCrudRepository<Customer, String> y obtienes de inmediato save, findById, deleteById suspend y un findAll que devuelve Flow. Los métodos de consulta derivados funcionan igual que en el Spring Data bloqueante — nombres como findByLastName o findByAgeGreaterThan se analizan y convierten en consultas — salvo que las variantes de resultado único son suspend y las de múltiples resultados devuelven Flow. Para lo que el analizador de nombres no pueda expresar, recurre a @Query con un filtro JSON de MongoDB en crudo y parámetros posicionales. Un detalle de Kotlin: un $ suelto inicia una plantilla de string, así que los operadores de Mongo deben escaparse — escribe @Query("{ 'age': { \$gt: ?0 } }"), no un $gt pelado.

Cuando necesitas control total, inyecta ReactiveMongoTemplate y construye consultas con el DSL Criteria. Query(Criteria.where("age").gte(18).and("active").isEqualTo(true)) compone un filtro con tipos seguros, y template.find(query, Customer::class.java) devuelve un Flux que conviertes con .asFlow(). La plantilla también es donde recurres a operaciones de actualización en sitio, pipelines de agregación, upserts y findAndModify — cosas que los repositorios no exponen. Un patrón común es usar CoroutineCrudRepository para el 80% del CRUD simple y caer a la plantilla para el 20% a medida.

Elegir los tipos de retorno con intención mantiene tu API honesta. Devuelve T? desde una búsqueda que puede fallar, devuelve Flow<T> desde cualquier cosa que produzca muchas filas, y devuelve un suspend Unit (o la entidad guardada) desde una escritura. Como Flow es frío, la consulta no se ejecuta hasta que algo la recolecta — así que una función que devuelve Flow<Customer> aún no ha tocado la base de datos cuando retorna. Recolecta con .toList() cuando de verdad necesites todo el conjunto en memoria, o mantenlo como Flow y transmítelo directamente desde un controlador WebFlux para tener backpressure de extremo a extremo.

Dos asuntos transversales merecen mención. Primero, la cancelación: la cancelación de corrutinas se propaga al driver reactivo, así que cuando un cliente se desconecta o envuelves una llamada en withTimeout, la consulta en curso se cancela de verdad en lugar de quedar ejecutándose. Segundo, las transacciones: MongoDB soporta transacciones multi-documento en replica sets, y Spring las expone de forma reactiva mediante TransactionalOperator (o @Transactional en un método suspend cuando la gestión reactiva de transacciones está configurada). Mantén las transacciones cortas, y recuerda que los starters reactivo y bloqueante de Mongo no deben estar ambos en el classpath — elige una pila por servicio.

@Document("customers")
data class Customer(
    @Id val id: String? = null,
    @Indexed val lastName: String,
    val firstName: String,
    val age: Int,
    val active: Boolean = true,
)

interface CustomerRepository : CoroutineCrudRepository<Customer, String> {

    // Derived query, single result -> suspend, nullable
    suspend fun findByLastName(lastName: String): Customer?

    // Derived query, many results -> Flow
    fun findByAgeGreaterThan(age: Int): Flow<Customer>

    // Raw filter when the name parser is not enough ($ escaped for Kotlin)
    @Query("{ 'active': true, 'age': { \$gt: ?0 } }")
    fun findActiveOlderThan(age: Int): Flow<Customer>
}

@Service
class CustomerService(private val repo: CustomerRepository) {

    suspend fun register(firstName: String, lastName: String, age: Int): Customer =
        repo.save(Customer(firstName = firstName, lastName = lastName, age = age))

    suspend fun findByLastNameOrThrow(lastName: String): Customer =
        repo.findByLastName(lastName)
            ?: throw NoSuchElementException("No customer: " + lastName)

    // Returns a cold Flow: no query runs until the caller collects it.
    fun adults(): Flow<Customer> = repo.findByAgeGreaterThan(17)

    suspend fun adultsSnapshot(): List<Customer> = adults().toList()
}

@Service
class CustomerSearch(private val template: ReactiveMongoTemplate) {

    fun search(minAge: Int, onlyActive: Boolean): Flow<Customer> {
        val criteria = Criteria.where("age").gte(minAge)
        if (onlyActive) criteria.and("active").isEqualTo(true)
        return template.find(Query(criteria), Customer::class.java).asFlow()
    }

    suspend fun countAdults(): Long =
        template.count(Query(Criteria.where("age").gte(18)), Customer::class.java)
            .awaitSingle()
}

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

data class Customer(val id: Int, val lastName: String, val age: Int)

class FakeRepo(private val data: List<Customer>) {
    // single result -> suspend, nullable
    suspend fun findByLastName(name: String): Customer? = data.firstOrNull { it.lastName == name }
    // many results -> cold Flow
    fun findByAgeGreaterThan(age: Int): Flow<Customer> = data.asFlow().filter { it.age > age }
}

fun main() = runBlocking {
    val repo = FakeRepo(listOf(Customer(1, "Lee", 30), Customer(2, "Ng", 15), Customer(3, "Cruz", 41)))
    println(repo.findByLastName("Lee"))
    val adults = repo.findByAgeGreaterThan(17).toList()
    println("adults=" + adults.size)
}