Why Your Flow Refuses to Hop Threads: Context Preservation, flowOn, and channelFlow in Kotlin

ule behind that crash, why Kotlin enforces it at runtime, the three legitimate ways to get work onto another dispatcher, and the production failure modes that still bite experienced teams. What you will take away What context preservation means, stated precisely rather than as folklore. The one-line fix that covers most cases, and why it beats flowOn. How flowOn and channelFlow actually work, including the hidden buffer.

A test that locks the invariant in so a refactor can’t quietly break it. A cold flow { } is a suspend lambda that the collector runs inside its own coroutine. Every call to emit is a direct, suspending call into the collector's downstream code. The rule is: emit must be called from the same coroutine context in which collect is running.

“Same context” means the relevant elements match: the dispatcher, plus a compatible Job hierarchy. It is not only about threads, which is why a nested withContext that swaps in a different dispatcher is rejected, while a withContext that doesn't change the effective context passes. The runtime check lives in the collector wrapper, and it fires on every emit. It is not a debug-only assertion.

Downstream operators and your own collect lambda are written as ordinary sequential code. That design depends on three guarantees: Sequential delivery. One value is fully processed before the next arrives. A predictable execution context.

A collect { } on Dispatchers.Main can touch views without wrapping every line in withContext(Dispatchers.Main). Safe unsynchronised state. An operator such as scan or onEach can keep a plain var because it is never entered concurrently. If emit could be called from arbitrary dispatchers, all three break at once.

Allowing it would push synchronisation costs onto every operator author and every collector. Kotlin chose the opposite trade: make the producer side responsible, and fail fast when it cheats. A small clarification of a claim you will see repeated: the invariant is not “all emissions happen on one thread.” Dispatchers can resume a coroutine on different threads between suspension points. The guarantee is about the coroutine context, and about emissions being sequential.

Here is a deliberately ordinary repository for a parcel-tracking app. The broken version is the one many of us write on the first attempt: class ParcelRepository( private val api: ParcelApi, private val io: CoroutineDispatcher = Dispatchers.IO, ) { // Violates the invariant: emit() is called while the context is `io`. fun trackBroken(id: String): Flow<ParcelStatus> = flow { withContext(io) { emit(api.fetchStatus(id)) } } } Collect it from a viewModelScope coroutine (which runs on Dispatchers.Main.immediate) and the first emission throws. The message begins with Flow invariant is violated and then lists the collector's context and the emitter's context side by side, which is the quickest way to see what differed.

Pin the behaviour down with a test. Injecting the dispatcher is what makes this testable: @Test fun `emitting from a different dispatcher is rejected`() = runTest { val otherDispatcher = StandardTestDispatcher(testScheduler) val broken = flow { withContext(otherDispatcher) { emit("late") } } assertFailsWith<IllegalStateException> { broken.first() } } Most of the time you don’t need a different flow builder at all. You need to separate where the work runs from where the value is emitted: class ParcelRepository( private val api: ParcelApi, private val io: CoroutineDispatcher = Dispatchers.IO, ) { fun track(id: String): Flow<ParcelStatus> = flow { val status = withContext(io) { api.fetchStatus(id) } emit(status) // back in the collector's context } } withContext returns before emit runs, so the emission happens in the collector's context. The context switch is a detail of one suspend call, not a property of the whole stream.

Better still, push the dispatcher switch into the data source so every caller benefits: class ParcelRemoteSource( private val service: ParcelService, private val io: CoroutineDispatcher = Dispatchers.IO, ) : ParcelApi { override suspend fun fetchStatus(id: String): ParcelStatus = withContext(io) { service.status(id).toDomain() } } // The repository no longer needs a dispatcher, because `api` is main-safe. class ParcelRepository(private val api: ParcelApi) { fun track(id: String): Flow<ParcelStatus> = flow { emit(api.fetchStatus(id)) } } This is the convention behind “suspend functions should be main-safe”: callers never have to know which dispatcher a function wants. flowOn for a stream of blocking or CPU-heavy work When the producer is a loop, such as polling, reading a file line by line, or decoding frames, wrapping every iteration in withContext is noisy. flowOn moves the entire upstream instead: class ParcelRepository( private val api: ParcelApi, private val io: CoroutineDispatcher = Dispatchers.IO, ) { // normalised() stands for any pure, CPU-bound extension on ParcelStatus.

fun poll(id: String, every: Duration = 5.seconds): Flow<ParcelStatus> = flow { while (currentCoroutineContext().isActive) { emit(api.fetchStatus(id)) delay(every) } } .map { it.normalised() } // runs on io .flowOn(io) // boundary .distinctUntilChanged() // runs in the collector's context } Two details deserve attention. It only affects what is above it. Operators placed after flowOn keep running in the collector's context. A flowOn(io) added at the end of a chain does not rescue a heavy map that was placed below an earlier boundary.

It is not free. To cross the boundary, flowOn runs the upstream in a separate coroutine and hands values over through a channel. With default settings that channel uses Channel.BUFFERED, which is 64 items on the JVM and therefore on Android (tunable through the kotlinx.coroutines.channels.defaultBuffer system property). The producer can therefore run ahead of the collector, which changes your back-pressure story compared with a plain flow.

See failure mode 4 below. If the upstream is already running on the dispatcher you request, the operator can avoid the extra hop, but you should not rely on that when reasoning about buffering. channelFlow for concurrent or multi-context producers Sometimes the problem genuinely is “several things produce values and they live on different dispatchers.” That is the job of channelFlow. Here send goes through a Channel, which is safe to call from any coroutine in the producer scope: class ParcelRepository( private val api: ParcelApi, private val io: CoroutineDispatcher = Dispatchers.IO, ) { fun trackMany(ids: List<String>): Flow<ParcelStatus> = channelFlow { val gate = Semaphore(permits = 4) // bound the fan-out for (id in ids) { launch { gate.withPermit { withContext(io) { send(api.fetchStatus(id)) } } } } } } The semaphore caps how many requests are in flight, but this still creates one coroutine per id.

Coroutines are cheap, so that is fine for hundreds of items. For very large lists, prefer flatMapMerge (shown below), which keeps only concurrency collectors active at a time. The legality of send inside withContext(io) is the key contrast with emit. The collector is not being called by the producer; it is draining a channel from its own coroutine.

The invariant still holds. It simply holds on the receiving side. Notes that matter in review: The block returns only after all of its children finish, and then the channel closes. That is why launching children inside the channelFlow scope matters.

Results arrive in completion order, not list order. If you need the original order, tag each result with its index and sort downstream, or use a different structure. For a plain “map each element concurrently” requirement, ids.asFlow().flatMapMerge(concurrency = 4) { id -> flow { emit(api.fetchStatus(id)) } } (inside the same repository) says the same thing with less ceremony. callbackFlow is channelFlow with an obligation Bridging a listener API follows the same channel mechanics, plus a rule you must satisfy: fun connectivity(manager: ConnectivityManager): Flow<Boolean> = callbackFlow { val callback = object : ConnectivityManager.NetworkCallback() { override fun onAvailable(network: Network) { trySend(true) } override fun onLost(network: Network) { trySend(false) } } manager.registerDefaultNetworkCallback(callback) awaitClose { manager.unregisterNetworkCallback(callback) } }.distinctUntilChanged() Two platform notes: registerDefaultNetworkCallback requires API 24 (Android 7.0), so on a lower minSdk register a NetworkRequest with registerNetworkCallback instead or guard the call with a Build.VERSION.SDK_INT check, and the manifest needs the ACCESS_NETWORK_STATE permission.

Without awaitClose, the block would finish immediately, the channel would close, and Kotlin throws to tell you the listener would leak. A short heuristic to go with the matrix: Can the suspend call itself switch dispatchers? Do that, and keep a plain flow. Is it a long-running sequential producer?

Use flowOn. Do multiple coroutines produce, or must values come from several contexts? Use channelFlow. Is a non-coroutine callback the producer?

Use callbackFlow. emit inside withContext Covered above. The tell-tale detail in code review is emit appearing inside the braces of a withContext, launch, or async block in a flow { }. emit from a launched child This is the same violation, but it hides behind a fork, so it passes your happy-path test: // Inside ParcelRepository, so `api` resolves.

// Compiles. Throws the moment the child reaches emit(). fun trackManyBroken(ids: List<String>): Flow<ParcelStatus> = flow { coroutineScope { ids.forEach { id -> launch { emit(api.fetchStatus(id)) } } } } The child coroutine has a different Job, so the collector rejects the call. The fix is channelFlow with send, or flatMapMerge.

flowOn in the wrong place // The heavy parse still runs on the collector's context. repo.rawLines() .flowOn(Dispatchers.IO) .map { parseHeavy(it) } // below the boundary Move flowOn so it sits directly after the expensive operators. When in doubt, log Thread.currentThread().name from inside each stage once, and read the output. Because flowOn introduces a channel, a fast producer and a slow collector no longer run in lock-step.

For "latest state wins" streams such as sensor readings or UI state, you rarely want a queue of stale values: sensor.readings() .flowOn(Dispatchers.Default) .conflate() // keep only the newest value when the collector is busy For streams where every item matters, set the capacity deliberately instead of inheriting the default: .buffer(capacity = 16, onBufferOverflow = BufferOverflow.SUSPEND). callbackFlow that never closes properly Two variants. Forgetting awaitClose causes the exception described earlier. Putting awaitClose { } in place but never unregistering inside it leaks the listener for the life of the process, with no crash to warn you.

The unregister call belongs in that trailing block. emit flow { try { emit(compute()) } catch (e: Exception) { // also catches downstream failures emit(fallback()) // now emitting after a downstream error } } A try/catch around emit also intercepts exceptions thrown by the collector. The flow contract (exception transparency) says the producer must not handle them. Catch around the upstream work, or use the catch operator, which only sees upstream exceptions.

[ ] No emit call sits lexically inside withContext, launch, or async in a flow { }. [ ] Suspend data-source functions switch dispatchers themselves, with the dispatcher injected. [ ] Every flowOn is placed directly after the operators it is meant to move. [ ] Each flowOn has an intentional buffering policy (conflate, buffer, or a documented default).

[ ] Each channelFlow bounds its concurrency and does not launch outside its own scope. [ ] Each callbackFlow ends with awaitClose and unregisters the listener there. [ ] A test asserts the happy path with an injected test dispatcher. The restriction on flow { } is not an arbitrary limitation.

It is what lets map, filter, collect, and your own operators be written as plain sequential Kotlin. Most of the time the right answer is to move the dispatcher switch down into the suspend function. When the shape of the problem is truly concurrent, channelFlow exists because a channel gives you the thread safety that the plain builder deliberately refuses to provide. If “we’ll add tests later” is a sentence your team says a little too often, I’d recommend “Mastering Android Testing” — a 35-chapter deep dive into why Android testing is uniquely hard (Activity/Fragment coupling, async complexity, tooling churn) and exactly how to fix it, layer by layer.

It walks through architecting for testability, JUnit/Truth/MockK fundamentals, ViewModel and coroutine/Flow testing without flakiness, Compose UI testing and the semantics tree, Espresso and instrumented tests, Room/Retrofit data-layer testing with a full sync-conflict-resolution case study, Robolectric’s real limits, CI pipelines, flaky test triage, and how to build a testing culture that actually survives contact with a deadline. E-book: Available on Google Play Kindle Edition: Available on Amazon Topmate: Available on Topmate Also available in Paperback & Hardcover. Originally published on Medium.