/kotlin-flows

name: kotlin-flows
description: Use when working with Flow, StateFlow, SharedFlow, or Channel in Kotlin — including cold vs hot stream decisions, operator chains, lifecycle-safe collection, UI state management, callback bridging, or Channel migration in Android or KMP projects.

Kotlin Flows

This reference covers the Flow traps and semantic edge cases — Channel vs SharedFlow, callback bridging, retry/error handling, side effects in transforms, and KMP boundaries — not the basics of cold/hot streams, the operator table, or lifecycle-safe collection.

Channel vs SharedFlow — the semantics that bite

| Found | Action | |---|---| | `BroadcastChannel` | Migrate → `SharedFlow` (deprecated) | | `ConflatedBroadcastChannel` | Migrate → `StateFlow` (deprecated) | | `Channel` for single-consumer fire-once events (nav, snackbars, one-shot effects) | **Keep — correct.** `Channel(BUFFERED).receiveAsFlow()` | | `Channel` broadcast to multiple collectors | Migrate → `SharedFlow` (see below) | | `Channel` as producer-consumer queue | Keep — correct |

**`Channel.receiveAsFlow()` is fan-out, NOT broadcast.** With multiple collectors, each event reaches **one** collector (the framework picks which), not all of them. If every collector must see every event, you need `SharedFlow`. This is the trap that's easy to fall into when reaching for `Channel` to multicast.

For single-consumer one-shot events, `Channel(BUFFERED).receiveAsFlow()` is the default because `send` suspends until consumed (the event is queued, never dropped), whereas `SharedFlow(replay = 0)` drops the emission if no collector is active at the instant of emission. Expose the `Flow`, never the `Channel`, and collect with `collect` inside `LaunchedEffect` — **not** `collectAsStateWithLifecycle`, which retains the last event as state and re-fires it on every recomposition / config change.

private val _events = Channel<UiEvent>(Channel.BUFFERED)
val events: Flow<UiEvent> = _events.receiveAsFlow()

// in the UI
LaunchedEffect(Unit) {
    viewModel.events.collect { event -> /* navigate / snackbar — consumed once */ }
}

**Exactly-once is really at-most-once-with-very-high-probability.** A `receiveAsFlow()` collector cancelled *after* `receive()` succeeds but *before* its block processes the element loses that element. Rare (most cancellations land between receives), but for payment/persistence-critical signals, store the outcome in durable state (a `pendingResult` field on `UiState`, cleared by the UI after consumption) — see `compose/references/state-management.md` "Durable state + acknowledgement".

SharedFlow: `launch { emit() }` over `tryEmit`

Default to `launch { _events.emit(e) }` — it suspends until the collector is ready, so the effect is never silently lost. `tryEmit()` on a default `MutableSharedFlow()` (no buffer) **silently drops** when no subscriber is ready; adding `extraBufferCapacity = 1` only moves the cliff (a second rapid emission while the buffer is full returns `false` and is dropped with no error). Use `tryEmit` only for miss-tolerable effects (tooltip, sound).

Exposing a read-only flow — explicit backing fields (Kotlin 2.4+)

The `_x` (mutable) / `x` (read-only) two-property idiom collapses into a single property with explicit backing fields — stable in Kotlin **2.4**, experimental in 2.3:

val uiState: StateFlow<UiState>
    field = MutableStateFlow(UiState())   // inside the class: uiState.update { … }; callers see StateFlow

Applies wherever a `StateFlow` / `SharedFlow` is exposed from a mutable backing (`MutableStateFlow` *is* a `StateFlow`, so the default getter type-checks). It does **not** fit a `Channel` exposed as a `Flow` — a `Channel` isn't a `Flow`, so keep the explicit `_events` / `receiveAsFlow()` pair above. On Kotlin < 2.4 (check the version catalog), keep the classic `private val _x` + `val x = _x.asStateFlow()`.

Callback bridging

// Single value: suspendCancellableCoroutine — always prefer it (propagates cancellation)
suspend fun authenticate(token: String): User = suspendCancellableCoroutine { cont ->
    val call = authApi.authenticate(token) { user, error ->
        if (user != null) cont.resume(user) else cont.resumeWithException(error ?: Exception())
    }
    cont.invokeOnCancellation { call.cancel() }
}
// suspendCoroutine only when the API has no cancellation concept at all.

// Stream: callbackFlow — awaitClose is MANDATORY (omitting it leaks the callback and never completes)
fun LocationManager.locationUpdates(provider: String): Flow<Location> = callbackFlow {
    val listener = LocationListener { trySend(it) }
    requestLocationUpdates(provider, 1000L, 0f, listener)
    awaitClose { removeUpdates(listener) }
}

For a third-party SDK with no deregistration API, set a flag inside `awaitClose` to signal shutdown and document the limitation.

retry / retryWhen — always guard the attempt count

repository.getItems()
    .retryWhen { cause, attempt ->
        if (cause is IOException && attempt < 3) { delay((attempt + 1) * 1_000L); true } else false
    }
    .collect { ... }

`attempt` is the 0-based retry index. Without the `attempt <` count check the flow retries **forever** on a persistent error. (`retry(3) { it is IOException }` is the simple form.)

Error handling — `.catch` scope and the cancellation trap

`.catch` intercepts **upstream** errors only (not exceptions thrown inside `collect {}`), and — unlike `try/catch` — it does not swallow `CancellationException`. So prefer the operator form, and for collector-side work catch **specific** types.

// ✓ upstream errors, cancellation-safe
repository.getItems().onEach { process(it) }.catch { logger.error(it) }.collect()

// ✗ catch (e: Exception) inside collect/any suspend fn swallows CancellationException and breaks
//   structured concurrency — the single most common coroutine bug.
//   If a broad catch is unavoidable: if (e is CancellationExcepti