skip to content

Why does emitting from a launched coroutine inside flow { } throw an exception, and how does channelFlow solve it?

level: middleimportance: must knowfreq 50%

answer

  1. Flow invariant = context preservation
  2. emit() checks emission context == collection context
  3. IllegalStateException: Flow invariant is violated
  4. channelFlow's send() is thread-safe
  5. Change dispatcher: flowOn, not launch

basics

~20 s

flow { } requires that all emit() calls happen in the same coroutine, so emitting from a new coroutine breaks that rule and throws. channelFlow uses a channel and a thread-safe send(), so any coroutine can produce values.

solid answer

~50 s

The plain flow { } builder enforces context preservation (the "flow invariant"): emit() must be invoked from the exact coroutine that runs the collector's block, so the emission context stays consistent. If you launch a child coroutine and call emit() from it, the framework detects the context mismatch and throws IllegalStateException: 'Flow invariant is violated... emissions from different coroutine context'. This rule keeps cancellation and context behavior predictable for the common sequential case. channelFlow sidesteps it by introducing a backing Channel: producers call send(), which is a thread-safe suspend operation, and the channel funnels values into the single collector context. So concurrency is allowed for production while the collector still sees a serial, well-ordered stream. The same applies to callbackFlow. If you only need to change dispatcher for part of a flow { }, use flowOn instead of launching.

code

kotlin · 13 lines
kotlin
// Illegal: violates the flow invariant
val bad = flow {
    coroutineScope { launch { emit(1) } } // throws IllegalStateException
}

// Legal: channelFlow allows concurrent send()
val good = channelFlow {
    launch { send(1) }
    launch { send(2) }
}

// Just want a different dispatcher? Use flowOn, no channel:
val onIo = flow { emit(load()) }.flowOn(Dispatchers.IO)

go deeper

for a junior

Recognizes that emitting from a launched coroutine in flow { } fails and that channelFlow is the fix.

for a middle

Explains the context-preservation invariant precisely and names IllegalStateException plus flowOn as the dispatcher-only alternative.

for a senior

Connects the invariant to structured concurrency/cancellation guarantees and notes channelFlow underlies merge/flatMapMerge.

for a principal

Reasons about when to pay the channel cost vs. restructuring, and the determinism trade-offs the invariant protects.

## The flow invariant (context preservation) The plain `flow { }` builder guarantees **context preservation**: the producer block must emit from the *same* coroutine context that ultimately runs the collector. This is why you change the upstream dispatcher with the **`flowOn`** operator, not by switching context inside the block. The rule keeps semantics simple: emission and collection share one coroutine, so structured concurrency, cancellation, and exception propagation are deterministic. If you write: ```kotlin flow { launch { emit(1) } // different coroutine! } ``` Kotlin throws at runtime: ``` IllegalStateException: Flow invariant is violated: Flow was collected in [context A], but emission happened in [context B]. Please refer to 'flow' documentation or use 'flowOn' instead ``` `emit()` internally checks `currentCoroutineContext()` against the captured collection context; a mismatch means the emission came from another coroutine, which is forbidden. ## How channelFlow fixes it `channelFlow` provides a `ProducerScope` that is **both** a `CoroutineScope` and a `SendChannel`. Emission goes through `send()`: - `send()` is **thread-safe and concurrency-safe** — it synchronizes on the backing channel, so any number of coroutines may call it. - The channel **decouples** producers from the collector: producers can run on any dispatcher; the channel delivers values into the collection context serially. ```kotlin fun merged(a: Flow<Int>, b: Flow<Int>): Flow<Int> = channelFlow { launch { a.collect { send(it) } } launch { b.collect { send(it) } } } ``` This is exactly how operators like `merge` and `flatMapMerge` are implemented under the hood. ## When NOT to reach for channelFlow - If you only want the **upstream on a different dispatcher**, use `flowOn(Dispatchers.IO)` — no channel needed. - If emission is naturally **sequential**, plain `flow { }` is cheaper (no channel allocation, no extra coroutine hop). ## Summary of the trade `flow { }` = sequential, single-context, lightweight. `channelFlow` = concurrent, channel-mediated, with a small synchronization/buffer cost — the price for legal multi-coroutine emission.

  • How would you change only the dispatcher of an upstream flow { } without violating the invariant?
    Apply the flowOn(dispatcher) operator downstream of the flow { } block. It shifts upstream execution to that dispatcher while preserving context-correct emission; no channel or launch is required.
  • Does channelFlow weaken cancellation guarantees?
    No — channelFlow runs the block in a child scope under structured concurrency. Cancelling the collector cancels the producer block and its children, and the channel is closed.

saying these in an interview costs you the question

  • Claiming you can launch+emit in flow { } if you use coroutineScope
  • Confusing flowOn with channelFlow as solutions to the same problem
  • Saying the exception is a compile error (it's runtime)
  • Not knowing send() is the channelFlow emit equivalent
  • Thinking the invariant is about thread safety rather than context

context

resolving links…