Spring Data R2DBC con Corrutinas

Spring Data R2DBC es la contraparte reactiva de Spring Data JDBC. En lugar del driver bloqueante de JDBC, habla con tu base de datos mediante R2DBC (Reactive Relational Database Connectivity), de modo que una consulta nunca deja un hilo parado mientras espera la red o el disco. La fontaneria reactiva por debajo es Project Reactor (Mono y Flux), pero casi nunca tienes que tocar esos tipos directamente. Al extender CoroutineCrudRepository, Spring expone toda la superficie del repositorio como corrutinas de Kotlin: los resultados unicos se vuelven suspend fun y los resultados de multiples filas se vuelven Flow<T>. El resultado es E/S no bloqueante que escribes y lees como si fuera codigo secuencial ordinario.

El modelo mental es simple una vez que interiorizas una regla: un metodo que devuelve como mucho un valor es un suspend fun, y un metodo que devuelve muchos valores es un Flow<T>. Asi, findById(id) es suspend fun findById(id): User? (devuelve una unica fila anulable), mientras que findAll() es fun findAll(): Flow<User> (sin la palabra suspend: llamarlo simplemente te entrega un Flow frio que recolectas despues). count() y existsById() son suspend porque producen exactamente un escalar. save() y deleteById() tambien son suspend. Este contraste, suspend para el uno y Flow para los muchos, es lo mas importante que recordar de la API.

CoroutineCrudRepository te da el CRUD basico, pero la verdadera productividad viene de los metodos de consulta derivados. Spring analiza el nombre del metodo y escribe el SQL por ti: findByEmail(email: String): User? genera SELECT ... WHERE email = :email, y como puede devolver cero o una fila lo declaras suspend con un retorno anulable. Un buscador que puede coincidir con muchas filas (findByActiveTrue() o findByLastNameOrderByCreatedAtDesc(name)) devuelve Flow<User> en su lugar. Manten la eleccion entre suspend y Flow alineada con la cardinalidad del metodo y el compilador y Spring estaran de acuerdo contigo.

Cuando un nombre derivado seria incomodo o necesitas un join, una proyeccion o un predicado personalizado, baja a @Query con SQL explicito. R2DBC usa SQL plano (no JPQL), con parametros nombrados enlazados mediante @Param. Aplican las mismas reglas de tipo de retorno: anota el metodo suspend fun ...: T? para una unica fila, o devuelve Flow<T> para un stream. Ten en cuenta que con R2DBC normalmente escribes los nombres literales de tabla y columna, ya que no hay una capa de mapeo ORM completa haciendolo por ti: lo que ves en la cadena se parece a lo que recibe la base de datos.

Flow es genuinamente perezoso y consciente de la contrapresion, lo que lo hace ideal para grandes conjuntos de resultados. Nada se obtiene hasta que empiezas a recolectar, y puedes transformar el stream con los operadores habituales (map, filter, take, onEach) antes de que una sola fila cruce el cable de verdad. Para consumir un Flow lo recolectas (repo.findAll().collect { ... }), lo pliegas (toList(), first(), single()) o lo devuelves directamente desde un controlador de Spring WebFlux, donde el framework transmite cada elemento emitido al cliente. Devolver el Flow en lugar de materializarlo con toList() preserva el streaming de extremo a extremo y mantiene la memoria plana.

Todo esto solo funciona dentro de una corrutina. En Spring WebFlux un manejador de controlador suspend o uno que devuelve Flow<T> ya corre dentro de una corrutina gestionada por el framework, asi que simplemente llamas a tus metodos suspend del repositorio directamente. Fuera de una peticion (un CommandLineRunner, un test, un trabajo programado) entras con runBlocking { } o lanzando dentro de un CoroutineScope. Combina el repositorio con TransactionalOperator (o el @Transactional compatible con corrutinas en un metodo de servicio suspend) para que multiples escrituras se confirmen atomicamente; las transacciones reactivas se propagan a traves del contexto de la corrutina y no por thread-locals.

Algunas notas practicas mantienen todo fluido. R2DBC no tiene carga perezosa, ni dirty checking, ni grafos de relaciones automaticos: save() inserta o actualiza el unico agregado que le entregas, asi que modelas las relaciones de forma explicita (columnas de clave foranea) y las cargas con consultas adicionales. Las entidades suelen ser data classes inmutables con un campo @Id; cuando el id es null Spring emite un INSERT y devuelve la copia poblada, asi que usa siempre el valor que devuelve save() y no el que pasaste. Y como cada llamada es suspendente, nunca bloquees dentro de estos metodos (sin Thread.sleep, sin JDBC): una sola llamada bloqueante puede atascar el pequeno pool de hilos del event-loop que comparte toda la pila reactiva.

En conjunto, la pila es pequena y predecible: una data class @Table inmutable, un CoroutineCrudRepository con metodos derivados mas algun @Query ocasional, y un servicio cuyas funciones son suspend o devuelven Flow. Obtienes persistencia reactiva y no bloqueante con la legibilidad del Kotlin en linea recta: la maquinaria de corrutinas oculta Reactor por completo mientras conservas el control total sobre tu SQL y tus transacciones.

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