Flow ↔ Reactor: Puente entre Coroutines y WebFlux

Spring WebFlux está construido sobre Project Reactor, cuyos tipos centrales son Mono (cero o un elemento) y Flux (cero a muchos elementos). Las coroutines de Kotlin modelan las mismas ideas con funciones suspend (un valor) y Flow (un flujo frío de muchos valores). Como ambos mundos describen streams asíncronos y conscientes de la contrapresión, las librerías kotlinx-coroutines-reactor y kotlinx-coroutines-reactive ofrecen conversiones bidireccionales sin pérdida, para que escribas código idiomático suspend/Flow mientras hablas con el stack reactivo de Spring por debajo.

La dirección más común es consumir Reactor desde coroutines. Llamar a Flux.asFlow() convierte cualquier Flux<T> en un Flow<T>, mientras que un Mono<T> se vuelve un valor compatible con suspend mediante .awaitSingle(), .awaitSingleOrNull() o .awaitFirstOrNull(). Estos adaptadores preservan la cancelación y la contrapresión: cuando la coroutine que colecta se cancela, la suscripción de Reactor subyacente se descarta; y como Flow es frío, nada se ejecuta hasta que realmente colectas. Esto permite consumir un repositorio reactivo (por ejemplo un ReactiveCrudRepository de Spring Data R2DBC que devuelve Flux) dentro de un handler suspend normal como si fuera una secuencia.

La dirección inversa importa cuando un framework espera un tipo reactivo. Flow.asPublisher() produce un Publisher<T> genérico de Reactive Streams, y Flow.asFlux() (de kotlinx-coroutines-reactor) produce específicamente un Flux<T> de Reactor. Un único resultado suspend puede envolverse con mono { ... } para construir un Mono cuyo cuerpo es una coroutine. Usa esto cuando una API solo acepta Publisher/Flux/Mono, por ejemplo al conectar un Flow a un emisor reactivo de Kafka, al cuerpo de un WebClient, o a cualquier operador sin equivalente en coroutines.

Una sutileza importante es la propagación del contexto. asFlux() y asPublisher() aceptan (o capturan) un CoroutineContext, porque los suscriptores de Reactive Streams pueden solicitar elementos desde hilos que no tienen contexto de coroutine asociado. Al convertir un Flow de vuelta a un publisher normalmente dejas que el operador capture el contexto circundante, pero ten en cuenta que el publisher resultante puede ser suscrito desde cualquier lugar, así que evita depender de un thread-local que solo existe en el momento de la construcción. A la inversa, asFlow() ejecuta tu colector en el contexto de coroutine en el que colectas, no en el scheduler de Reactor.

Colectar un Flow dentro de un handler de coroutine es el patrón cotidiano. En un método de controlador suspend puedes llamar repository.findActive().asFlow() y luego usar operadores familiares —map, filter, take, onEach, buffer— y colectores terminales como toList(), first() o collect { }. Esto mantiene la lógica de transformación en Kotlin plano en lugar del DSL de operadores de Reactor, lo cual es más fácil de leer, probar y depurar, mientras la semántica de suspensión y cancelación sigue propagándose hasta el driver de la base de datos.

Devolver un Flow directamente es donde brilla el soporte de coroutines de Spring. Desde Spring Framework 5.2, un handler de @RestController de WebFlux puede ser una función suspend y puede devolver un Flow<T> directamente; Spring lo adapta internamente de vuelta a un Flux para la respuesta HTTP. Si configuras produces = MediaType.TEXT_EVENT_STREAM_VALUE (o APPLICATION_NDJSON_VALUE), cada elemento se envía al cliente conforme se emite, dándote streaming real del lado del servidor sin escribir nunca un solo tipo de Reactor en tu controlador.

Juntándolo todo, un servicio WebFlux limpio puede ser coroutine-first de extremo a extremo: el repositorio devuelve Flux, adaptas con asFlow(), transformas con operadores de coroutine y devuelves el Flow directamente desde un handler suspend. Spring cierra el círculo convirtiéndolo de vuelta a un Flux para el cable. Solo recurres a asFlux()/asPublisher()/mono { } en las costuras donde una API reactiva no-coroutine lo exige, y hasta esas conversiones son de una sola línea.

Algunas precauciones prácticas: prefiere awaitSingleOrNull() sobre awaitSingle() cuando un Mono pueda estar vacío (el último lanza NoSuchElementException), recuerda que Flow es frío así que volver a colectar vuelve a ejecutar el upstream, y añade buffer() o conflate() si un productor rápido supera a un colector lento. Mantén la frontera de conversión delgada: convierte una vez en el borde, haz tu lógica de negocio en Flow y deja que Spring maneje la adaptación final. Estas librerías son dependencias de leer-y-copiar: añade kotlinx-coroutines-reactor a tu build de Gradle, no son ejecutables en un sandbox de navegador.

import kotlinx.coroutines.flow.*
import kotlinx.coroutines.reactive.awaitSingleOrNull
import reactor.core.publisher.Flux
import reactor.core.publisher.Mono

interface UserRepo {
    fun findActive(): Flux<User>      // Spring Data R2DBC style
    fun findById(id: Long): Mono<User>
}

class UserService(private val repo: UserRepo) {
    // Flux -> Flow, then plain Kotlin operators
    fun activeNames(): Flow<String> =
        repo.findActive()
            .asFlow()
            .filter { it.enabled }
            .map { it.name }

    // Mono -> suspend value (null-safe; awaitSingle() would throw if empty)
    suspend fun nameOf(id: Long): String? =
        repo.findById(id).awaitSingleOrNull()?.name
}

data class User(val id: Long, val name: String, val enabled: Boolean)

import kotlinx.coroutines.flow.*
import org.springframework.http.MediaType
import org.springframework.web.bind.annotation.*

@RestController
@RequestMapping("/api/users")
class UserController(private val service: UserService) {

    // Server-Sent Events: streamed item by item
    @GetMapping("/stream", produces = [MediaType.TEXT_EVENT_STREAM_VALUE])
    fun stream(): Flow<String> = service.activeNames()

    // suspend handler returning a single value
    @GetMapping("/{id}")
    suspend fun one(@PathVariable id: Long): String =
        service.nameOf(id) ?: "unknown"
}

import kotlinx.coroutines.flow.*
import kotlinx.coroutines.reactor.asFlux
import kotlinx.coroutines.reactor.mono
import kotlinx.coroutines.reactive.asPublisher
import kotlinx.coroutines.delay
import org.reactivestreams.Publisher
import reactor.core.publisher.Flux
import reactor.core.publisher.Mono

val source: Flow<Int> = flowOf(1, 2, 3)

// Flow -> Reactor Flux (Reactor-specific)
val asFlux: Flux<Int> = source.asFlux()

// Flow -> generic Reactive Streams Publisher
val asPub: Publisher<Int> = source.asPublisher()

// single suspend result -> Mono via a coroutine builder
fun loadToken(): Mono<String> = mono {
    delay(50)
    "token-" + (1..3).sum()
}

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

fun numbers(): Flow<Int> = flow {
    for (i in 1..5) emit(i)   // cold: re-runs on every collect
}

fun main() = runBlocking {
    val evens = numbers()
        .filter { it % 2 == 0 }
        .map { it * 10 }
        .toList()
    println(evens)            // [20, 40]

    // collecting again re-runs the upstream
    numbers().take(2).collect { println("got " + it) }
}