Reactive MongoDB with Coroutines

Spring Data MongoDB ships two stacks. The classic blocking stack (MongoTemplate, MongoRepository) parks a thread on every database round trip; the reactive stack does not. Under the hood the reactive driver speaks to MongoDB over non-blocking I/O and surfaces results as Project Reactor types — Mono<T> for zero-or-one and Flux<T> for zero-or-many. To turn it on you add the spring-boot-starter-data-mongodb-reactive dependency and let Spring Boot autoconfigure a ReactiveMongoTemplate and reactive repositories. The payoff is throughput: a handful of event-loop threads can service thousands of concurrent queries, which is exactly the model WebFlux is built around.

In Kotlin you almost never want to consume Mono and Flux directly. The kotlinx-coroutines-reactor bridge lets you await a Mono with .awaitSingle()/.awaitSingleOrNull() and collect a Flux as a Kotlin Flow with .asFlow(). Spring Data goes one step further: CoroutineCrudRepository exposes its CRUD methods as suspend functions and returns Flow for collections, so your data layer reads like ordinary sequential Kotlin while staying fully non-blocking. The rule of thumb is simple — a single result is a suspend fun returning T?, a stream of results is a fun returning Flow<T>, and you never block to get either.

Documents map cleanly onto Kotlin data classes. Annotate the class with @Document("collection") and mark the primary key with @Id; an immutable val id: String? = null is idiomatic because MongoDB generates the ObjectId on insert and Spring populates it on the returned copy. Keep the rest of the fields as vals — data classes give you equals, hashCode, copy and destructuring for free, and the driver instantiates them through the primary constructor, so nullable defaults double as a lightweight schema-evolution strategy. These Spring Data annotations target the field, so on a primary-constructor property you write them with no use-site prefix. Use @Field to rename a property in storage and @Indexed to declare an index.

Defining a repository is mostly declaration. Extend CoroutineCrudRepository<Customer, String> and you immediately get suspend save, findById, deleteById and a Flow-returning findAll. Derived query methods work the same way as in blocking Spring Data — method names like findByLastName or findByAgeGreaterThan are parsed into queries — except the single-result variants are suspend and the multi-result variants return Flow. For anything the name parser cannot express, drop to @Query with a raw MongoDB JSON filter and positional parameters. One Kotlin gotcha: a bare $ starts a string template, so Mongo operators must be escaped — write @Query("{ 'age': { \$gt: ?0 } }"), not a bare $gt.

When you need full control, inject ReactiveMongoTemplate and build queries with the Criteria DSL. Query(Criteria.where("age").gte(18).and("active").isEqualTo(true)) composes a type-safe filter, and template.find(query, Customer::class.java) returns a Flux you convert with .asFlow(). The template is also where you reach for update-in-place operations, aggregation pipelines, upserts and findAndModify — things repositories do not surface. A common pattern is to use CoroutineCrudRepository for the 80% of plain CRUD and fall back to the template for the bespoke 20%.

Choosing return types deliberately keeps your API honest. Return T? from a lookup that may miss, return Flow<T> from anything that yields many rows, and return a suspend Unit (or the saved entity) from a write. Because Flow is cold, the query does not execute until something collects it — so a function that returns Flow<Customer> has not yet touched the database when it returns. Collect with .toList() when you genuinely need the whole set in memory, or keep it as a Flow and stream it straight out of a WebFlux controller for end-to-end backpressure.

Two cross-cutting concerns deserve a mention. First, cancellation: coroutine cancellation propagates into the reactive driver, so when a client disconnects or you wrap a call in withTimeout, the in-flight query is actually cancelled rather than left running. Second, transactions: MongoDB supports multi-document transactions on replica sets, and Spring exposes them reactively through TransactionalOperator (or @Transactional on a suspend method when reactive transaction management is configured). Keep transactions short, and remember that the reactive and blocking Mongo starters must not both be on the classpath — pick one stack per service.

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