Flow Operators Explained
KOTLIN › Flow
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.
map and filter are the operators everyone reaches for, but transform is the one they are both built on, and knowing it is a small credibility signal.
map emits exactly one value per input. filter emits zero or one. transform emits **however many you like**, including none:
flowOf(1, 2, 3)
.transform { n ->
emit(n) // pass it through
emit(n * 10) // and emit an extra one
}
// 1, 10, 2, 20, 3, 30
That makes transform the right tool whenever the input-to-output relationship is not one-to-one:
// Insert a Loading state before each result
repo.search(query).transform { result ->
emit(UiState.Loading)
emit(UiState.Loaded(result))
}
Two neighbours worth distinguishing from it. **onEach** runs a side effect and always passes the value through unchanged, so it is for logging and analytics, not for shaping data. **mapNotNull** is map followed by filter, dropping nulls in one step.
.onEach { log("saw $it") } // value continues untouched
.mapNotNull { it.toIntOrNull() } // non-numeric entries disappear
The tell of someone who has only used map is a chain that fights the one-to-one constraint: a map producing a list followed by a flattenConcat, where a single transform would have said it directly.
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.
debounce and sample are the pair people mix up, and an interviewer asking for one and receiving the other notices.
**debounce(300) waits for silence.** It emits a value only once 300ms have passed with no new emission. If values keep arriving faster than that, nothing is ever emitted.
**sample(300) takes a snapshot on a timer.** Every 300ms it emits whatever the latest value was, regardless of whether things have gone quiet.
// User typing: emit only when they pause
searchQuery.debounce(300).flatMapLatest { api.search(it) }
// Location updates arriving continuously: emit at most every 5s
locationUpdates.sample(5_000).collect { updateMap(it) }
The distinction in one line: **debounce is for bursts that end, sample is for streams that do not.**
Use debounce on a search box and you get one request when the user stops typing, which is what you want. Use sample there and you fire a request every 300ms while they are still typing, which is what you were trying to avoid. Conversely, use debounce on a continuous sensor feed and you may emit **nothing at all**, because the quiet window never arrives.
// Bug: an accelerometer never goes quiet, so this emits nothing
accelerometer.debounce(1000).collect { ... }
A related detail: debounce also takes a lambda, so the timeout can depend on the value. A short query can wait longer than a long one, on the theory that the user is still typing.
The Latest family is where Flow's cancellation story becomes an operator, and there are more members than most people can name.
.mapLatest { } // cancel the in-progress transform when a new value arrives
.transformLatest { } // same, but may emit any number of values
.flatMapLatest { } // cancel the in-progress inner flow
.collectLatest { } // terminal: cancel the in-progress collector block
All four share one rule: **a new upstream value cancels whatever the previous one was still doing.** They differ only in what that work is.
// Each keystroke abandons the in-flight request and starts a new one
searchQuery
.debounce(300)
.flatMapLatest { query -> repo.search(query) } // returns a Flow
.collect { render(it) }
// The suspending equivalent when the work is a single value, not a flow
searchQuery
.debounce(300)
.mapLatest { query -> repo.searchOnce(query) } // returns a value
.collect { render(it) }
The trap is that cancellation only helps if the work is **actually cancellable**. A suspending network call is; a tight CPU loop with no suspension points is not, so mapLatest around one will simply run every transform to completion and give you no benefit at all.
.mapLatest { heavyCpuLoop(it) } // nothing is ever cancelled: no suspension points
And a detail worth knowing: these operators are still marked @ExperimentalCoroutinesApi in current releases, which surprises people given how routine flatMapLatest is in production code.
Three operators shape the ends of a stream rather than its values, and they turn a flow into a complete UI state machine.
repo.users()
.onStart { emit(UiState.Loading) } // before the first upstream value
.onEmpty { emit(UiState.NoResults) } // upstream finished without emitting
.onCompletion { cause -> hideSpinner() } // every ending, whatever it was
**onStart** runs before the upstream is collected, and it can emit, which is what makes the loading-state idiom work rather than needing a separate flow.
**onEmpty** fires only when the upstream completed having emitted nothing at all. It is the clean way to express "no results" without tracking a boolean in the collector.
**onCompletion** runs on **every** ending: values exhausted, upstream threw, or collection cancelled. Its cause parameter is null on success and non-null otherwise, which makes it symmetric with onStart and the right home for cleanup.
.onCompletion { cause ->
when (cause) {
null -> log("finished normally")
is CancellationException -> log("cancelled")
else -> log("failed: $cause")
}
}
The distinction to hold on to: **onCompletion observes, catch handles.** onCompletion seeing a non-null cause does not stop that exception propagating; the flow still fails. If you want to recover, you need catch. Using onCompletion where you meant catch gives you a log line and a crash.
Two operators control how much of a stream you take, and one of them has a consequence people find surprising.
.take(5) // first 5 values, then cancel the upstream
.takeWhile { it < 10 } // values until the predicate fails, then cancel
.drop(2) // skip the first 2, emit the rest
.dropWhile { it < 0 } // skip until the predicate first fails
The surprise is that **take cancels the upstream.** Once it has its quota it does not merely stop emitting downstream, it cancels the producer, which for an infinite flow is the only way collection could ever finish:
// Terminates, even though the producer loops forever
flow { var i = 0; while (true) emit(i++) }
.take(3)
.collect { println(it) } // 0, 1, 2, then done
It does this by throwing an internal AbortFlowException, which the operator catches itself. That matters because a badly written catch upstream of a take can swallow it and break termination, and it is why you should never catch Throwable broadly inside a flow builder.
first() behaves the same way as a terminal operator: it takes one value and cancels the rest, which is how you read a single value from a hot flow without collecting forever:
val current = viewModel.state.first() // one value, then done
withIndex() and distinctUntilChangedBy { } round out the group: the first pairs each value with its position, the second deduplicates on a selector rather than on the whole value.
Finally, the operator that changes how you launch a collection rather than what it emits.
collect is a suspend function, so it blocks the rest of its coroutine until the flow finishes. For a hot flow that never finishes, that means everything after it is unreachable:
viewModelScope.launch {
repo.updates().collect { handle(it) }
setupSomethingElse() // never runs
}
launchIn fixes it by starting the collection in its own coroutine and returning a Job immediately:
repo.updates()
.onEach { handle(it) }
.launchIn(viewModelScope) // returns a Job, does not suspend
setupSomethingElse() // runs straight away
launchIn is exactly scope.launch { collect() }, nothing more. The idiom that comes with it is moving the work into onEach, since launchIn takes no lambda of its own. That reads well when the whole pipeline is declarative:
repo.updates()
.filter { it.isRelevant }
.onEach { _state.value = it }
.catch { _state.value = Error(it) }
.launchIn(viewModelScope)
The mistake worth avoiding is reaching for it to collect several flows and losing track of the jobs. Each launchIn starts an independent coroutine, so if two of them must be cancelled together, launch them inside one parent instead of relying on the scope alone.
The way this gets examined is rarely "what does zip do". It is a problem stated in words, expecting the operator name back.
| The problem | The operator | |---|---| | Fire a search only once the user stops typing | debounce(300) | | Emit at most once per interval from a continuous feed | sample(1000) | | Cancel the in-flight request when a new query arrives | flatMapLatest | | Recompute whenever **either** input changes | combine | | Pair items up one-for-one, waiting for both | zip | | Stop doing redundant work when the same value repeats | distinctUntilChanged() | | Emit a Loading state before the real values | onStart { emit(Loading) } | | Say "no results" when nothing arrived | onEmpty { } | | Hide the spinner however the flow ended | onCompletion { } | | Read one value from a hot flow and stop | first() | | Turn one input into several outputs | transform { } | | Start collecting without suspending the caller | launchIn(scope) |
The two pairs worth rehearsing because they are the ones candidates swap:
- **combine vs zip**: combine fires whenever either side emits, using the latest from both. zip waits for one from each and pairs them, so it runs at the speed of the slower flow and finishes when the shorter one does. - **debounce vs sample**: waiting for silence versus emitting on a timer. A burst that ends versus a stream that does not.
Naming the right operator for a stated problem is what separates having read the API list from having used it.