Flow & Operators Explained
KOTLIN › Concurrency
Flow sits between coroutines and the UI, and it's the default way modern Android code exposes an asynchronous stream of values, a Room query, a network response, a sequence of location updates. Interviewers use it to check whether you actually understand how a stream behaves, not just which operator name to type, and the first thing to get right is that a Flow is **cold**.
A cold flow runs nothing on its own. The producer block inside flow { } sits inert until something collects it, and it reruns independently, from the top, for every single collector.
val coldFlow = flow {
println("Producer started")
emit(1); emit(2)
}
launch { coldFlow.collect { println("A: $it") } }
launch { coldFlow.collect { println("B: $it") } }
// "Producer started" prints TWICE, once per collector
Nothing is shared and nothing is cached between those two collectors, each one gets its own full run of the block. If you want code to run once, right at the start of every collection, without touching the producer body itself, wrap it in onStart { }, it still fires fresh for each collector for the exact same reason the whole flow does. Cold is the default. Turning a flow hot and shared is a deliberate extra step, and it's worth coming back to once the pipeline-building operators are down.
Flow operators split into two kinds. **Intermediate operators** (map, filter, transform, onEach, flowOn) are lazy: each one wraps the upstream flow and returns a new Flow, but nothing executes yet. **Terminal operators** (collect, toList, first, single) are suspend functions that actually drive collection and make the whole chain run.
val upstream = flowOf(1, 2, 3)
val mapped: Flow<Int> = upstream.map { it * 2 } // nothing runs yet
val result: List<Int> = upstream.toList() // this triggers execution
A chain of ten map or filter calls with no terminal operator does precisely nothing, it's just an inert description of a pipeline until something collects it.
**flowOn changes only the upstream context, and only upstream.** Anything declared *before* flowOn in the chain runs on the dispatcher you give it; everything *after* it, and the collector itself, stays in whatever context called collect.
flow { emit(loadData()) } // runs on IO
.map { it.transform() } // also runs on IO
.flowOn(Dispatchers.IO) // only affects everything ABOVE this line
.collect { value -> // runs in the CALLER's context, e.g. Main
updateUi(value)
}
This is called **context preservation**, and it's why you never need to wrap collect { } in its own withContext to get back to the main thread: flowOn already guarantees the collector sees whatever context it was called from, untouched.
Flow also enforces **exception transparency**: a failure from the producer should propagate downstream the same way a value does, and the rule that guarantees this is that emit may only be called directly from the flow { } builder's own coroutine context, never from inside a withContext block.
// BAD: throws "Flow invariant is violated"
val bad = flow {
withContext(Dispatchers.IO) { emit(loadData()) } // throws IllegalStateException
}
// GOOD: keep emit in the builder's own context, shift the whole producer with flowOn
val good = flow {
emit(loadData())
}.flowOn(Dispatchers.IO)
Wrapping emit in your own try/catch inside the builder breaks the same guarantee for a different reason: it hides producer failures from everything downstream instead of letting them propagate to be handled properly. If you genuinely need to push values from a different coroutine or a callback, flow { } isn't the right builder at all, channelFlow or callbackFlow are built for exactly that and don't carry this restriction.
You recover from an upstream failure with the catch operator, never by wrapping emit in a try/catch yourself, that would violate the rule from a moment ago. Inside catch you can log, rethrow, or emit a fallback value to keep the flow alive.
flow { emit(1); throw RuntimeException("upstream") }
.catch { e -> emit(-1) } // catches the upstream exception, emits a fallback
.collect { value -> /* ... */ }
The limitation to hold onto: catch only sees exceptions thrown by the producer and by operators placed above it in the chain. It has no visibility into the collect lambda, or into anything placed below it.
flow { emit(1) }
.catch { println("never reached for collector errors") }
.collect { throw RuntimeException("collector error") } // NOT caught above
To handle a failure thrown inside collect itself, wrap the entire collect call in an ordinary try/catch. catch and try/catch cover two different halves of the same pipeline, upstream and the collector.
When a producer is fast and the collector is slow, you have three tools, and they behave very differently:
- **buffer**: runs producer and collector concurrently, queueing emissions so nothing is dropped - **conflate**: keeps only the most recent value, silently dropping intermediates the collector couldn't keep up with - **collectLatest**: cancels the in-progress collector block whenever a new value arrives and restarts it fresh
flow { repeat(10) { i -> emit(i); delay(10) } }
.conflate()
.collect { value ->
delay(100) // slow
println(value) // only ever sees the latest, not all 10
}
collectLatest is the one to reach for when the collector block itself does expensive, cancellable work, an in-flight network call, say, and only the newest input actually matters.
combine and zip both merge two flows, but they answer different questions.
**zip** pairs values strictly one-to-one, waiting for a match from both sides and finishing when either flow completes:
nums.zip(strs) { n, s -> "$n$s" } // strict pairing: 1-X, 2-Y
**combine** re-emits every time *either* flow produces a new value, using the latest value from each side, so it fires far more often once both have emitted at least once:
nums.combine(strs) { n, s -> "$n$s" } // fires on every change: 1-X, 2-X, 2-Y
A good mental model: zip is for aligned pairs of data, item N from stream A goes with item N from stream B. combine is for 'recompute whenever any input changes', the shape you want for something like combining a search query flow with a filter-settings flow into one live result.
flatMapConcat, flatMapMerge, and flatMapLatest all flatten a flow of flows, but they order and cancel very differently:
- **flatMapConcat**: collects inner flows one at a time, in upstream order, waiting for each to finish before starting the next - **flatMapMerge**: collects all inner flows concurrently, with no ordering guarantee - **flatMapLatest**: cancels the previous inner flow the moment a new upstream value arrives
searchQueryFlow
.flatMapLatest { query ->
searchApi.search(query) // cancels the previous in-flight search
}
.collect { results -> updateUi(results) }
flatMapLatest is the textbook fit for search-as-you-type: only the newest query's results matter, and cancelling the stale in-flight request avoids wasted work and out-of-order results landing on screen. flatMapConcat, by contrast, is what you want when order genuinely matters more than speed, uploading a queue of files one at a time, say.
A few more operators round out the flow toolbox for situations you'll actually hit on Android. **debounce(300)** waits for a quiet window with no new emissions before letting a value through, the standard fix for a search box: you don't want to fire a network call on every keystroke, only once the user pauses.
searchQueryFlow // emits on every keystroke
.debounce(300) // only emits after 300ms of silence
.flatMapLatest { query -> searchApi.search(query) }
.collect { results -> updateUi(results) }
**distinctUntilChanged** is a much narrower tool: it only drops a value if it's equal to the one immediately before it, so it stops you doing redundant work when the same value repeats back to back. It does nothing to space values out over time the way debounce does.
For failures, **retry(n) { predicate }** re-collects the entire upstream flow from scratch when a thrown exception matches the predicate, useful for a flaky network call. retryWhen gives you the same idea with access to the attempt count, so you can back off longer between each retry.
Not every source of values is already a suspend-friendly Flow, a location provider or a click listener hands you values through a callback instead. callbackFlow { } bridges that gap: inside it you call trySend(value) from the callback to push values downstream, the callback-safe equivalent of emit in a regular flow { } builder.
fun locationUpdates(client: FusedLocationProviderClient): Flow<Location> =
callbackFlow {
val cb = object : LocationCallback() {
override fun onLocationResult(r: LocationResult) {
trySend(r.lastLocation!!)
}
}
client.requestLocationUpdates(request, cb, Looper.getMainLooper())
awaitClose { client.removeLocationUpdates(cb) }
}
The one thing you cannot skip is the final awaitClose { } block. It suspends and keeps the flow alive until the collector cancels, and it's your one guaranteed place to unregister the listener, removeLocationUpdates here, so it doesn't keep firing after nobody's collecting. Leave it out and callbackFlow throws IllegalStateException at runtime, it exists specifically to force you to clean up.
Every operator up to this point, map, flowOn, catch, combine, flatMapLatest, debounce, describes how a cold pipeline is shaped. None of it makes the flow shared. stateIn and shareIn are the step where a cold, per-collector pipeline becomes a hot, shared stream that many collectors can observe without each one re-running the whole chain from scratch.
val state: StateFlow<UiState> = coldDataFlow
.map { UiState(it) }
.stateIn(
scope = viewModelScope,
started = SharingStarted.WhileSubscribed(5_000),
initialValue = UiState.Loading
)
stateIn produces a StateFlow, which always has a current value, that's why it demands an initialValue up front. shareIn produces a SharedFlow instead, which has no required current value and is better suited to one-off events than to state.
If an interviewer asks you to describe Flow, the shape of a strong answer is this: it's cold and lazy by default, intermediate operators build a pipeline without running it, a terminal operator drives that pipeline to completion, flowOn and exception transparency control which context and which operator sees a given value or failure, and operators like conflate, collectLatest, combine, and flatMapLatest exist because producer and collector speed rarely match. Naming the right operator for a stated problem, cancel-the-stale-request, keep-only-the-latest, recompute-on-any-change, is what separates knowing the API from actually having used it.