Spring Data R2DBC with Coroutines

Spring Data R2DBC is the reactive counterpart to Spring Data JDBC. Instead of the blocking JDBC driver, it talks to your database over R2DBC (Reactive Relational Database Connectivity), so a query never parks a thread while it waits for the network or the disk. The reactive plumbing underneath is Project Reactor (Mono and Flux), but you almost never have to touch those types directly. By extending CoroutineCrudRepository, Spring exposes the whole repository surface as Kotlin coroutines: single results become suspend fun, and multi-row results become Flow<T>. The result is non-blocking I/O that you write and read as if it were ordinary sequential code.

The mental model is simple once you internalize one rule: a method that returns at most one value is a suspend fun, and a method that returns many values is a Flow<T>. So findById(id) is suspend fun findById(id): User? (it returns a single nullable row), while findAll() is fun findAll(): Flow<User> (no suspend keyword — calling it just hands you a cold Flow you collect later). count() and existsById() are suspend because they yield exactly one scalar. save() and deleteById() are suspend too. This contrast — suspend for the one, Flow for the many — is the single most important thing to remember about the API.

CoroutineCrudRepository gives you the CRUD basics, but the real productivity comes from derived query methods. Spring parses the method name and writes the SQL for you: findByEmail(email: String): User? generates SELECT ... WHERE email = :email, and because it can return zero or one row you declare it suspend with a nullable return. A finder that can match many rows — findByActiveTrue() or findByLastNameOrderByCreatedAtDesc(name) — returns Flow<User> instead. Keep the suspend-vs-Flow choice aligned with the method's cardinality and the compiler and Spring will agree with you.

When a derived name would be awkward or you need a join, projection, or custom predicate, drop down to @Query with explicit SQL. R2DBC uses plain SQL (not JPQL), with named parameters bound via @Param. The same return-type rules hold: annotate the method suspend fun ...: T? for a single row, or return Flow<T> for a stream. Note that with R2DBC you typically write the literal table and column names, since there is no full ORM mapping layer doing it for you — what you see in the string is close to what the database receives.

Flow is genuinely lazy and back-pressure aware, which makes it ideal for large result sets. Nothing is fetched until you start collecting, and you can transform the stream with the usual operators — map, filter, take, onEach — before a single row crosses the wire in earnest. To consume a Flow you collect it (repo.findAll().collect { ... }), fold it (toList(), first(), single()), or return it straight out of a Spring WebFlux controller, where the framework streams each emitted item to the client. Returning the Flow rather than materializing it with toList() preserves streaming end to end and keeps memory flat.

All of this only works inside a coroutine. In Spring WebFlux a suspend controller handler or one returning Flow<T> already runs in a coroutine the framework manages, so you simply call your suspend repository methods directly. Outside a request — a CommandLineRunner, a test, a scheduled job — you bridge in with runBlocking { } or by launching inside a CoroutineScope. Pair the repository with TransactionalOperator (or the coroutine-friendly @Transactional on a suspend service method) so that multiple writes commit atomically; reactive transactions are propagated through the coroutine context rather than thread-locals.

A few practical notes keep things smooth. R2DBC has no lazy loading, no dirty checking, and no automatic relationship graphs — save() either inserts or updates the single aggregate you hand it, so you model relationships explicitly (foreign-key columns) and load them with extra queries. Entities are usually immutable data classes with an @Id field; when the id is null Spring issues an INSERT and returns the populated copy, so always use the value save() returns rather than the one you passed in. And because every call is suspending, never block inside these methods (no Thread.sleep, no JDBC) — one blocking call can stall the small event-loop thread pool that the whole reactive stack shares.

Put together, the stack is small and predictable: an immutable @Table data class, a CoroutineCrudRepository with derived methods plus the occasional @Query, and a service whose functions are suspend or return Flow. You get reactive, non-blocking persistence with the readability of straight-line Kotlin — the coroutine machinery hides Reactor entirely while you keep full control over your SQL and your transactions.

import org.springframework.data.annotation.Id
import org.springframework.data.relational.core.mapping.Table
import org.springframework.data.repository.kotlin.CoroutineCrudRepository
import kotlinx.coroutines.flow.Flow

@Table("users")
data class User(
    @Id val id: Long? = null,   // null -> Spring does an INSERT
    val email: String,
    val active: Boolean = true,
)

interface UserRepository : CoroutineCrudRepository<User, Long> {
    // one row at most -> suspend, nullable return
    suspend fun findByEmail(email: String): User?

    // many rows -> Flow, no suspend keyword
    fun findByActiveTrue(): Flow<User>
}

import org.springframework.data.r2dbc.repository.Query
import org.springframework.data.repository.query.Param
import org.springframework.data.repository.kotlin.CoroutineCrudRepository
import kotlinx.coroutines.flow.Flow

interface AccountRepository : CoroutineCrudRepository<User, Long> {

    @Query("SELECT * FROM users WHERE email = :email")
    suspend fun lookup(@Param("email") email: String): User?

    @Query("SELECT * FROM users WHERE active = true ORDER BY id DESC LIMIT :max")
    fun recentActive(@Param("max") max: Int): Flow<User>
}

import org.springframework.stereotype.Service
import org.springframework.transaction.annotation.Transactional
import kotlinx.coroutines.flow.Flow
import kotlinx.coroutines.flow.map

@Service
class UserService(private val repo: UserRepository) {

    suspend fun register(email: String): User =
        repo.save(User(email = email))   // use the RETURNED copy (id populated)

    suspend fun findOne(email: String): User? = repo.findByEmail(email)

    // returns a Flow -> streaming preserved all the way to the caller
    fun activeEmails(): Flow<String> =
        repo.findByActiveTrue().map { it.email }

    @Transactional
    suspend fun deactivate(id: Long) {
        val user = repo.findById(id) ?: return
        repo.save(user.copy(active = false))
    }
}

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

// like 'suspend fun findById' -> returns at most one value
suspend fun findOne(id: Int): String? =
    listOf("ann", "bob", "cy").getOrNull(id)

// like a Flow finder -> a cold stream of many values
fun findActive(): Flow<String> = flow {
    listOf("ann", "bob", "cy").forEach { emit(it) }
}

fun main() = runBlocking {
    println(findOne(1))                 // bob
    println(findOne(9))                 // null
    val names = findActive()
        .map { it.uppercase() }
        .toList()                       // collect the stream
    println(names)                      // [ANN, BOB, CY]
}