Flow ↔ Reactor: Bridging Coroutines and WebFlux

Spring WebFlux is built on Project Reactor, whose core types are Mono (zero or one item) and Flux (zero to many items). Kotlin coroutines model the same ideas with suspend functions (one value) and Flow (a cold stream of many values). Because both worlds describe asynchronous, backpressure-aware streams, the kotlinx-coroutines-reactor and kotlinx-coroutines-reactive libraries give you lossless, two-way conversions so you can write idiomatic suspend/Flow code while still talking to the reactive Spring stack underneath.

The most common direction is consuming Reactor from coroutines. Calling Flux.asFlow() turns any Flux<T> into a Flow<T>, while a Mono<T> becomes a suspend-friendly value via .awaitSingle(), .awaitSingleOrNull(), or .awaitFirstOrNull(). These adapters preserve cancellation and backpressure: when the collecting coroutine is cancelled, the underlying Reactor subscription is disposed; and because Flow is cold, nothing runs until you actually collect. This lets a reactive repository (for example a Spring Data R2DBC ReactiveCrudRepository that returns Flux) be consumed inside a regular suspend handler as if it were a sequence.

The reverse direction matters when a framework expects a reactive type. Flow.asPublisher() produces a generic Reactive Streams Publisher<T>, and Flow.asFlux() (from kotlinx-coroutines-reactor) produces a Reactor Flux<T> specifically. A single suspend result can be wrapped with mono { ... } to build a Mono whose body is a coroutine. Use these when an API only accepts Publisher/Flux/Mono — for instance wiring a Flow into a reactive Kafka sender, a WebClient body, or any operator that has no coroutine equivalent.

An important subtlety is context propagation. asFlux() and asPublisher() accept (or capture) a CoroutineContext, because Reactive Streams subscribers may request items from threads that have no coroutine context attached. When converting a Flow back to a publisher you usually let the operator capture the surrounding context, but be aware that the resulting publisher can be subscribed to from anywhere, so avoid relying on a thread-local that only exists at construction time. Conversely, asFlow() runs your collector in whatever coroutine context you collect it in, not on Reactor's scheduler.

Collecting a Flow inside a coroutine handler is the everyday pattern. In a suspend controller method you can call repository.findActive().asFlow() and then use familiar operators — map, filter, take, onEach, buffer — and terminal collectors like toList(), first(), or collect { }. This keeps transformation logic in plain Kotlin instead of the Reactor operator DSL, which is easier to read, test, and debug, while the suspension and cancellation semantics still flow through to the database driver.

Returning a Flow directly is where Spring's coroutine support shines. Since Spring Framework 5.2, a WebFlux @RestController handler may be a suspend function and may return a Flow<T> directly; Spring internally adapts it back to a Flux for the HTTP response. If you set produces = MediaType.TEXT_EVENT_STREAM_VALUE (or APPLICATION_NDJSON_VALUE), each element is flushed to the client as it is emitted, giving you true server-side streaming without ever writing a single Reactor type in your controller.

Putting it together, a clean WebFlux service can be coroutine-first end to end: the repository returns Flux, you adapt with asFlow(), transform with coroutine operators, and return the Flow straight from a suspend handler. Spring closes the loop by converting it back to a Flux for the wire. You only reach for asFlux()/asPublisher()/mono { } at the seams where a non-coroutine reactive API forces the issue — and even those conversions are one-liners.

A few practical guardrails: prefer awaitSingleOrNull() over awaitSingle() when a Mono may be empty (the latter throws NoSuchElementException), remember that Flow is cold so re-collecting re-runs the upstream, and add buffer() or conflate() if a fast producer outpaces a slow collector. Keep the conversion boundary thin — convert once at the edge, do your business logic in Flow, and let Spring handle the final adaptation. These libraries are read-and-copy dependencies: add kotlinx-coroutines-reactor to your Gradle build, they are not runnable in a browser sandbox.

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