How do awaitBody, awaitSingle, and related await* extensions bridge Reactor Mono/Flux into coroutine code, and what are their empty/multi-value semantics?

middlemust knowfreq 45%

answer

awaitSingle throws on empty; awaitSingleOrNull returns null
Flux: awaitFirst / awaitLast / asFlow
awaitBody vs awaitBodyOrNull for HTTP payloads
subscribes once; cancellation propagates
mono{}/flux{}/asFlux = reverse direction

basics

~20 s

They are suspending extension functions (from kotlinx-coroutines-reactor) that subscribe to a Mono/Flux and suspend the coroutine until a value arrives, then return it. awaitSingle() expects exactly one value; awaitBody<T>() reads and deserializes a request/response body inside a suspend function.

solid answer

~40 s

To call reactive APIs from coroutines you convert publishers to suspend calls. kotlinx-coroutines-reactor provides Mono<T>.awaitSingle() (suspends, returns the value; throws NoSuchElementException if the Mono is empty) and awaitSingleOrNull() (returns null for an empty Mono). For Flux<T> there are awaitFirst(), awaitFirstOrNull(), awaitFirstOrDefault(), awaitLast(), awaitSingle(), plus .asFlow() to consume it as a Flow. On the Spring side, ServerRequest.awaitBody<T>()/awaitBodyOrNull<T>() and WebClient's awaitBody<T>()/awaitBodyOrNull<T>() and awaitExchange { } deserialize bodies as suspend calls, built on the same bridge. Each await* subscribes to the publisher once and honors cancellation: if the coroutine is cancelled, the underlying subscription is cancelled. Choose the OrNull variants whenever emptiness is legitimate to avoid exceptions.

To call reactive APIs from coroutines you convert publishers to suspend calls. `kotlinx-coroutines-reactor` provides `Mono<T>.awaitSingle()` (suspends, returns the value; throws `NoSuchElementException` if the Mono is empty) and `awaitSingleOrNull()` (returns `null` for an empty Mono). For `Flux<T>` there are `awaitFirst()`, `awaitFirstOrNull()`, `awaitFirstOrDefault()`, `awaitLast()`, `awaitSingle()`, plus `.asFlow()` to consume it as a `Flow`. On the Spring side, `ServerRequest.awaitBody<T>()`/`awaitBodyOrNull<T>()` and `WebClient`'s `awaitBody<T>()`/`awaitBodyOrNull<T>()` and `awaitExchange { }` deserialize bodies as suspend calls, built on the same bridge. Each await* subscribes to the publisher once and honors cancellation: if the coroutine is cancelled, the underlying subscription is cancelled. Choose the OrNull variants whenever emptiness is legitimate to avoid exceptions.

code

kotlin · 15 lines

kotlin

suspend fun example(webClient: WebClient, repo: ReactiveUserRepository) {
    // Mono -> value; empty Mono would THROW here
    val forced: UserDto = webClient.get().uri("/users/1")
        .retrieve().awaitBody()

    // Mono may be empty -> use OrNull
    val maybe: User? = repo.findById(42L).awaitSingleOrNull()

    // Flux -> Flow, consumed with backpressure & cancellation
    val names: Flow<String> = repo.findAll().asFlow().map { it.name }
    names.collect { println(it) }

    // Flux first element with a safe fallback
    val firstOrNull: User? = repo.findAll().awaitFirstOrNull()
}

go deeper

for a junior

Know await* extensions turn a Mono/Flux into a suspend call and that OrNull variants exist for empty sources.

for a middle

Distinguish awaitSingle vs awaitSingleOrNull vs awaitFirst/awaitLast, and use awaitBody/bodyToFlow correctly on WebClient and ServerRequest.

for a senior

Explain single-subscription and cancellation propagation, cold re-subscription, and the reverse mono{}/flux{}/asFlux adapters WebFlux uses internally.

for a principal

Reason about error/empty modeling at boundaries, timeout+cancellation semantics with withTimeout, and consistent OrNull conventions across a large reactive+coroutine codebase.

Why bridging exists. Reactor (Mono/Flux) and Kotlin coroutines are two different async models. To call one from the other you need adapters. kotlinx-coroutines-reactor supplies them in both directions; the coroutine→consume-a-publisher direction is the await* family of suspending extension functions. Calling mono.awaitSingle() subscribes to the Mono, suspends the current coroutine, and resumes it with the emitted value (or an exception) — no thread is blocked while waiting.

Mono awaiters.

Mono<T>.awaitSingle(): T — resumes with the single value; if the Mono completes empty it throws NoSuchElementException; a Mono error resumes with that exception.
Mono<T>.awaitSingleOrNull(): T? — resumes with the value or null on empty completion. This is the correct choice for Mono that may be empty (e.g. a findById that returns empty).

Flux awaiters.

Flux<T>.awaitFirst(): T — first element; throws on empty.
Flux<T>.awaitFirstOrNull(): T? and awaitFirstOrDefault(default) — safe empty handling.
Flux<T>.awaitLast(): T — last element (waits for completion).
Flux<T>.awaitSingle(): T — exactly one element expected; throws if zero or more than one.
Flux<T>.asFlow(): Flow<T> — the usual way to consume a multi-value stream: iterate with collect { }. Preserves backpressure and cancellation.

Spring body awaiters. These build on the same idea for HTTP payloads:

Functional endpoints: ServerRequest.awaitBody<T>(), awaitBodyOrNull<T>(), bodyToFlow<T>(), awaitFormData(), awaitMultipartData(), awaitPrincipal().
WebClient: after .retrieve(), use awaitBody<T>(), awaitBodyOrNull<T>(), awaitBodilessEntity(); or awaitExchange { response -> ... } for full control; and bodyToFlow<T>() for streaming.

kotlin

val user: UserDto = webClient.get().uri("/users/{id}", id)
    .retrieve().awaitBody()
val maybe: UserDto? = webClient.get().uri("/users/{id}", id)
    .retrieve().awaitBodyOrNull()
val feed: Flow<Event> = webClient.get().uri("/events")
    .retrieve().bodyToFlow()

The reverse direction (for completeness). To expose coroutine code as Reactor, kotlinx-coroutines-reactor provides the mono { } and flux { } coroutine builders, and Flow<T>.asFlux(). WebFlux uses these internally to adapt your suspend/Flow handlers.

Cancellation & single-subscription. Every await* call subscribes to the publisher exactly once. If the enclosing coroutine is cancelled (client disconnect, timeout via withTimeout), the subscription is cancelled too — structured concurrency and Reactor cancellation are linked. Because Mono/Flux are cold, awaiting the same publisher twice re-subscribes and re-executes it.

Common gotchas.

awaitSingle()/awaitBody() on an empty source throw NoSuchElementException — a frequent bug when a repository legitimately returns empty. Use the OrNull variant.
Flux.awaitSingle() throws if the flux emits more than one element; use awaitFirst() / asFlow() for multi-element streams.
Do not mix .block() into coroutine code — it blocks the event loop; use the await* extensions instead.
These extensions are only in scope with kotlinx-coroutines-reactor on the classpath.

When to use. Use await* whenever you are inside a suspend function and need a single value from a reactive API (WebClient, R2DBC ReactiveCrudRepository, a library returning Mono); use .asFlow()/bodyToFlow() for streams. Prefer the OrNull/OrDefault variants at every boundary where empty is a valid outcome.

**Why bridging exists.** Reactor (`Mono`/`Flux`) and Kotlin coroutines are two different async models. To call one from the other you need adapters. `kotlinx-coroutines-reactor` supplies them in both directions; the coroutine→consume-a-publisher direction is the `await*` family of **suspending extension functions**. Calling `mono.awaitSingle()` subscribes to the `Mono`, suspends the current coroutine, and resumes it with the emitted value (or an exception) — no thread is blocked while waiting. **Mono awaiters.** - `Mono<T>.awaitSingle(): T` — resumes with the single value; if the Mono completes empty it throws `NoSuchElementException`; a Mono error resumes with that exception. - `Mono<T>.awaitSingleOrNull(): T?` — resumes with the value or `null` on empty completion. This is the correct choice for `Mono` that may be empty (e.g. a `findById` that returns empty). **Flux awaiters.** - `Flux<T>.awaitFirst(): T` — first element; throws on empty. - `Flux<T>.awaitFirstOrNull(): T?` and `awaitFirstOrDefault(default)` — safe empty handling. - `Flux<T>.awaitLast(): T` — last element (waits for completion). - `Flux<T>.awaitSingle(): T` — exactly one element expected; throws if zero or more than one. - `Flux<T>.asFlow(): Flow<T>` — the usual way to consume a multi-value stream: iterate with `collect { }`. Preserves backpressure and cancellation. **Spring body awaiters.** These build on the same idea for HTTP payloads: - Functional endpoints: `ServerRequest.awaitBody<T>()`, `awaitBodyOrNull<T>()`, `bodyToFlow<T>()`, `awaitFormData()`, `awaitMultipartData()`, `awaitPrincipal()`. - WebClient: after `.retrieve()`, use `awaitBody<T>()`, `awaitBodyOrNull<T>()`, `awaitBodilessEntity()`; or `awaitExchange { response -> ... }` for full control; and `bodyToFlow<T>()` for streaming. ```kotlin val user: UserDto = webClient.get().uri("/users/{id}", id) .retrieve().awaitBody() val maybe: UserDto? = webClient.get().uri("/users/{id}", id) .retrieve().awaitBodyOrNull() val feed: Flow<Event> = webClient.get().uri("/events") .retrieve().bodyToFlow() ``` **The reverse direction (for completeness).** To expose coroutine code as Reactor, `kotlinx-coroutines-reactor` provides the `mono { }` and `flux { }` coroutine builders, and `Flow<T>.asFlux()`. WebFlux uses these internally to adapt your `suspend`/`Flow` handlers. **Cancellation & single-subscription.** Every `await*` call subscribes to the publisher exactly once. If the enclosing coroutine is cancelled (client disconnect, timeout via `withTimeout`), the subscription is cancelled too — structured concurrency and Reactor cancellation are linked. Because `Mono`/`Flux` are cold, awaiting the same publisher twice re-subscribes and re-executes it. **Common gotchas.** - `awaitSingle()`/`awaitBody()` on an empty source throw `NoSuchElementException` — a frequent bug when a repository legitimately returns empty. Use the `OrNull` variant. - `Flux.awaitSingle()` throws if the flux emits more than one element; use `awaitFirst()` / `asFlow()` for multi-element streams. - Do not mix `.block()` into coroutine code — it blocks the event loop; use the await* extensions instead. - These extensions are only in scope with `kotlinx-coroutines-reactor` on the classpath. **When to use.** Use `await*` whenever you are inside a suspend function and need a single value from a reactive API (WebClient, R2DBC `ReactiveCrudRepository`, a library returning `Mono`); use `.asFlow()`/`bodyToFlow()` for streams. Prefer the `OrNull`/`OrDefault` variants at every boundary where empty is a valid outcome.

Your repo.findById(id).awaitSingle() intermittently throws NoSuchElementException. Why and how do you fix it?
The `Mono` completes empty when the id is absent, and `awaitSingle()` treats empty as an error. Switch to `awaitSingleOrNull()` and handle `null` (e.g. return 404), or map empty to a domain result before awaiting.
How do you consume a Flux of many elements in coroutine code?
Convert it with `flux.asFlow()` and iterate via `collect { }` (or `toList()`), which preserves backpressure and cancellation. `awaitSingle()` would throw because the Flux emits more than one element; `awaitFirst()` only takes the first.

saying these in an interview costs you the question

Using awaitSingle() on sources that can be empty and being surprised by NoSuchElementException.
Using awaitSingle() on a multi-element Flux.
Reaching for .block() inside a suspend function instead of await*.
Believing await* re-uses a cached value across calls (each call re-subscribes cold publishers).

How do awaitBody, awaitSingle, and related await* extensions bridge Reactor Mono/Flux into coroutine code, and what are their empty/multi-value semantics?

§ answer

basics

solid answer

code

§ go deeper

▾level notes4 levels

▾deep dive~2 min

▾common follow-ups2 questions

▾red flags4 flags

§ context

▾related questions6 questions

▾related topics3 topics

answer

go deeper

context