Bridging Coroutines and Reactor

Spring WebFlux and the broader reactive ecosystem are built on Project Reactor's two core types: a `Mono<T>` that emits zero or one item, and a `Flux<T>` that emits zero to many. Kotlin coroutines speak a different dialect entirely: `suspend` functions that read like sequential code, and `Flow<T>` for asynchronous streams. The `kotlinx-coroutines-reactor` (and `kotlinx-coroutines-reactive`) modules are the official, bidirectional bridge between these two worlds, letting you consume a reactive `Publisher` from inside a coroutine and, conversely, expose coroutine code as a `Publisher` that Reactor can subscribe to.

Going from Reactor into coroutines is the most common direction, because so many Spring APIs (WebClient, the reactive repositories, R2DBC) hand you a `Mono` or `Flux`. The bridge provides suspending await operators that subscribe under the hood and resume your coroutine with the result. For a `Mono`, use `awaitSingle()` when you expect exactly one value, `awaitSingleOrNull()` when the `Mono` may be empty, and the single-element operators `awaitFirst()` / `awaitFirstOrNull()` / `awaitLast()` when working with a `Flux` and you only care about one element. These functions are honest about Reactor's contracts: `awaitSingle()` throws `NoSuchElementException` on an empty source and `IllegalArgumentException` if more than one element arrives, so pick the operator that matches the cardinality you actually expect.

A crucial subtlety is the empty `Mono`. Reactor models 'no value' as a completed-without-emission signal, not as `null`, so calling `awaitSingle()` on an empty `Mono` raises `NoSuchElementException` rather than returning `null`. Whenever a value is genuinely optional — a lookup that may miss, a `deleteById` that returns `Mono<Void>` — reach for `awaitSingleOrNull()` (or `awaitFirstOrNull()` on a `Flux`). This maps Reactor's empty completion onto Kotlin's nullable types cleanly, so your `?:`, `?.let`, and smart-casts behave exactly as a Kotlin developer expects.

Spring's reactive `WebClient` adds its own convenience on top of this: `awaitBody<T>()` and `awaitBodyOrNull<T>()`. They are extension functions that internally call `bodyToMono<T>()` and then await it, collapsing the usual `.retrieve().bodyToMono(User::class.java).awaitSingle()` chain into a single reified call. Because the type is reified you write `awaitBody<User>()` with no `.class` argument, and the generic information survives erasure for deserialization. For streaming responses you can instead call `bodyToFlow<T>()` to get a Kotlin `Flow` and collect it idiomatically.

The reverse direction — exposing suspend code as a `Publisher` — is handled by the coroutine builders `mono { }` and `flux { }`. `mono { block }` runs your suspending block and produces a `Mono<T>` that emits the block's return value (or completes empty if the block returns `null`); `flux { block }` runs the block with a `ProducerScope<T>` receiver whose `send()` channel-style API lets you emit many values from suspend code. These are the tools you need when a framework demands a reactive return type but you want to author the logic with coroutines — for example a Spring Security `ReactiveAuthenticationManager`, a `WebFilter`, or any contract typed as `Mono`/`Flux` that you cannot change to `suspend`.

Both builders take an optional `CoroutineContext` argument (defaulting to `EmptyCoroutineContext`, which means the subscriber's context is used). Critically, `mono { }` and `flux { }` are cold and lazy: the block does not run until something subscribes, and it re-runs on every subscription — matching Reactor's cold-publisher semantics rather than launching eagerly like `launch` or `async`. Cancellation is wired through as well: if the downstream subscriber cancels, the coroutine is cancelled and your `suspend` calls observe it cooperatively, so you get correct resource cleanup for free.

In a typical Spring Boot 3 controller you rarely need these builders at all, because Spring WebFlux understands `suspend` functions and `Flow<T>` natively — a `suspend fun` handler returning `User` is automatically adapted to `Mono<User>`, and a `Flow<User>` to `Flux<User>`. The bridge earns its keep at the boundaries: inside a suspend handler you call third-party reactive APIs and `awaitSingle()` their results, and you only fall back to `mono { }` / `flux { }` when an interface signature outside your control is hard-typed to a `Publisher`.

A few practical guidelines tie it together. Prefer `Flow` over `Flux` for stream logic you own, converting at the edges with `.asFlow()` and `.asFlux()`. Choose the `*OrNull` variant the moment emptiness is a legal outcome, and let the exception-throwing variant guard genuine invariants. Add the `org.jetbrains.kotlinx:kotlinx-coroutines-reactor` dependency (it transitively brings `-reactive`) so both the await operators and the builders are on your classpath. Done well, the bridge lets the reactive plumbing stay invisible while you write straight-line, debuggable, sequential-looking Kotlin.

import kotlinx.coroutines.reactor.awaitSingle
import kotlinx.coroutines.reactor.awaitSingleOrNull
import reactor.core.publisher.Mono

// Imagine these come from a reactive repository / WebClient
fun findUser(id: Long): Mono<User> = Mono.just(User(id, "Ada"))
fun findMissing(): Mono<User> = Mono.empty()

suspend fun loadUsers() {
    // Exactly one expected -> awaitSingle()
    val user: User = findUser(1).awaitSingle()
    println("Loaded " + user.name)

    // May be empty -> awaitSingleOrNull() returns null instead of throwing
    val maybe: User? = findMissing().awaitSingleOrNull()
    println(maybe?.name ?: "no user found")
}

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

import org.springframework.web.reactive.function.client.WebClient
import org.springframework.web.reactive.function.client.awaitBody
import org.springframework.web.bind.annotation.*

@RestController
class ProfileController(private val client: WebClient) {

    @GetMapping("/profile/{id}")
    suspend fun profile(@PathVariable id: Long): Profile {
        // reified body type, no .class argument, internally awaits the Mono
        return client.get()
            .uri("/users/" + id)
            .retrieve()
            .awaitBody<Profile>()
    }
}

data class Profile(val id: Long, val handle: String)

import kotlinx.coroutines.reactor.mono
import kotlinx.coroutines.delay
import reactor.core.publisher.Mono

// Pretend this signature is dictated by a framework interface you cannot change.
fun authenticate(token: String): Mono<Principal> = mono {
    // suspend code lives here; it runs only when subscribed
    delay(50) // e.g. a suspend call to a token service
    if (token == "valid") Principal("ada") else null // null -> empty Mono
}

data class Principal(val name: String)

import kotlinx.coroutines.runBlocking
import kotlinx.coroutines.flow.flow
import kotlinx.coroutines.flow.first
import kotlinx.coroutines.flow.toList
import kotlinx.coroutines.flow.map

fun main() = runBlocking {
    // A Flow models the same stream concept Flux does, but with suspend operators.
    val numbers = flow {
        for (n in 1..3) emit(n)
    }

    val first = numbers.first()        // analogous to Flux.awaitFirst()
    val all = numbers.map { it * 10 }.toList()

    println("first = " + first)        // first = 1
    println("all = " + all)            // all = [10, 20, 30]
}