Flow: the Basics Explained
KOTLIN › Flow
You already know a suspend fun returns a single value, once:
suspend fun loadUser(): User // one User, then it's done
But lots of things aren't one value, they're a **stream of values that arrive over time**: location updates as you move, rows coming back from a database, search results as the user types. A single return can't model that.
A **Flow** is Kotlin's type for exactly this: an asynchronous stream that produces zero, one, or many values over time. The clearest mental model is *a List you receive one item at a time* rather than all at once, or *a Sequence that's allowed to pause and wait* between items.
val numbers: Flow<Int> = flow {
emit(1) // hand out a value
emit(2)
emit(3)
}
numbers.collect { n -> println(n) } // 1, 2, 3
emit sends a value out; collect receives each one. The rest of this lesson takes each half apart properly. The one idea to hold onto: a Flow is a **stream**, not a single result.
Every Flow has two halves that work as a pair.
The **producer** is the flow { } block. Inside it you call **emit(value)** to hand out values, one at a time, in order:
val ticker = flow {
emit("first")
delay(1000) // wait a second
emit("second")
}
The **collector** is whoever calls **collect { }**. Its lambda runs once for every value the producer emits:
ticker.collect { value ->
println(value) // "first", then a second later, "second"
}
Here's the key: producer and collector are wired together. Each emit sends one value straight to the collector's block, waits for it to finish, and only then moves on to the next line. Nothing is buffered or fired-and-forgotten by default, it's a controlled, one-at-a-time handoff.
The reason Flow is so pleasant to write: everything inside it is **suspending**, so you write straight-line code instead of callbacks.
Inside a flow { } you can call any suspend function, a network request, a database read, delay, right before you emit the result:
val prices = flow {
while (true) {
val price = api.fetchPrice() // suspend network call
emit(price)
delay(5000) // poll every 5s
}
}
No callback interfaces, no listener registration, no manual threading. Compare that to the old world of setOnPriceChangedListener { ... } with nested callbacks: Flow turns it into a simple loop. And because collect is itself a suspend function, the whole thing is **cancellable**, if the coroutine collecting this flow is cancelled, the loop stops cleanly.
Creating a Flow runs **nothing**. val f = flow { ... } just builds a description of the stream, like writing down a recipe without cooking it. The producer block doesn't execute until a **terminal operator** starts it, and the most common one is collect.
val f = flow { println("running!"); emit(1) }
// nothing has printed yet
f.collect { } // NOW "running!" prints and 1 is emitted
Because collect is a suspend function, it has to run inside a coroutine, you'll almost always see it in a launch { }, a viewModelScope, or another suspend function:
viewModelScope.launch {
f.collect { value -> updateUi(value) }
}
Hold onto that: nothing runs until something actually drives the flow. The next part covers exactly which calls count as driving it, and which only describe it.
Once you have a flow, you'll usually shape it before collecting, filtering values out or transforming them. That's what operators like map, filter, and onEach are for. But those are **intermediate operators**, and they're just as lazy as the flow itself. Calling .map { } doesn't loop over anything or run any code, it just wraps the flow in a new description: when someone eventually collects, transform each value like this.
val doubled = flow {
emit(1)
emit(2)
emit(3)
}.map { it * 2 } // just wraps, runs nothing yet
doubled.collect { println(it) } // NOW it runs: 2, 4, 6
Nothing in that chain actually runs until a **terminal operator**, collect, toList, or first, drives it from the start. That's the split worth remembering: intermediate operators (map, filter, onEach) describe a transformation and stay lazy; terminal operators (collect, toList, first) are what actually starts the producer and pulls values through the whole chain.
One more foundational fact, and it's the bridge to the next lesson. Because a Flow is just a recipe that runs when collected, **each collector triggers its own independent run** of the producer:
val f = flow {
println("producer starting")
emit(1)
}
launch { f.collect { } } // "producer starting" prints
launch { f.collect { } } // "producer starting" prints AGAIN
Two collectors, two separate runs, nothing shared between them by default. This property has a name: a Flow like this is **cold** (it comes alive only when collected, and re-runs per collector), as opposed to a **hot** stream that's always running and shared. You don't need the hot side yet, just hold onto "cold = runs on demand, once per collector."
That's the whole foundation. A Flow is a stream you build with flow { emit(...) }, shape with lazy intermediate operators like map and filter, and actually run with a terminal operator like collect { } inside a coroutine. It does nothing until collected, and it's cold. If an interviewer asks what a Flow is, that's your answer: an asynchronous stream, cold and lazy by default, one producer run per collector. The **Flow & Operators** lesson picks up right here, everything you can do to that stream: transform it, switch threads, handle errors, combine two of them.
Flows are **cancellable by default**, but only at certain points. The emit() function inside a flow builder automatically checks whether the collector's coroutine is still active. If the coroutine has been cancelled, emit() throws CancellationException and the flow stops:
val numbers = flow {
for (i in 1..1_000_000) {
emit(i) // checks for cancellation on every emit
}
}
val job = scope.launch {
numbers.collect { value ->
if (value == 3) cancel() // cancel the collecting coroutine
}
}
// Flow stops after emitting 3 -- emit(4) sees the cancellation
But not every operator checks cancellation. Intermediate operators like map, filter, and onEach do **not** automatically check -- they just transform values as they pass through. If you have a long chain of operators between emit and collect, cancellation is still detected at emit and collect, but not in between.
For operators that produce values without calling emit (like IntRange.asFlow()), you can add an explicit check with the .cancellable() operator:
(1..1_000_000).asFlow()
.cancellable() // inserts a cancellation check
.filter { it % 2 == 0 }
.collect { println(it) }
Inside a custom flow { } builder, you can also call currentCoroutineContext().ensureActive() to check cancellation manually at any point, not just at emit().
Here is a practical summary of the three tools for flow cancellation:
**1. emit() -- automatic.** Inside a flow { } builder, every emit() call checks the collector's coroutine. This is why most flows cancel promptly without any extra work.
**2. .cancellable() -- opt-in for non-builder flows.** Flows created from asFlow(), flowOf(), or other sources that don't use emit() internally skip the automatic check. Chain .cancellable() to restore it:
flowOf(1, 2, 3)
.cancellable() // now checks cancellation between elements
.collect { ... }
**3. ensureActive() -- manual check anywhere.** Inside a flow { } builder you can call currentCoroutineContext().ensureActive() at any point, not just at emit. Useful when you have a long computation between emits:
flow {
for (chunk in hugeDataset.chunked(1000)) {
val processed = heavyTransform(chunk)
currentCoroutineContext().ensureActive() // check mid-computation
emit(processed)
}
}
The mental model: emit() is your free cancellation checkpoint. If you're not calling emit() frequently enough (heavy computation between emits, or a non-builder flow), add .cancellable() or ensureActive() to keep the flow responsive to cancellation.
flow { } is the builder you will write most, but there are four others and knowing when each is simpler is worth a moment.
flowOf(1, 2, 3) // a fixed set of known values
listOf("a", "b").asFlow() // an existing collection, sequence, or range
(1..100).asFlow() // a range
emptyFlow<Int>() // emits nothing and completes
flow { emit(api.load()) } // anything that needs to suspend
The rule is simple: **if nothing suspends, you do not need flow { }.** Writing flow { listOf(1,2,3).forEach { emit(it) } } when flowOf(1, 2, 3) says the same thing is a small tell.
asFlow() has one caveat worth carrying forward: because it emits without suspending, it provides no cancellation check, so a long asFlow() chain doing heavy work per item ignores cancellation until you add cancellable().
There are also two **channel-backed** builders for the cases these cannot handle:
callbackFlow { } // values arrive from a callback that cannot suspend
channelFlow { } // several coroutines emit concurrently into one flow
Both exist because the plain flow { } builder requires emit to be called from its own coroutine, which a listener callback or a second child coroutine cannot do. They have their own topic later; for now, just know the plain builder is not the only one.
collect is the terminal operator you meet first, and it is not the only one. A terminal operator is simply one that **runs** the flow, so each is a suspend function that returns a value rather than another Flow.
flow.collect { } // run it, do something per value
flow.toList() // collect everything into a List
flow.first() // take one value, then cancel the upstream
flow.firstOrNull() // same, but null if the flow was empty
flow.single() // exactly one value: throws if there are zero or two
flow.count() // how many values
flow.reduce { a, b -> } // fold using the first value as the seed
flow.fold(0) { a, b -> }// fold with an explicit initial value
Two are worth knowing precisely.
**first() cancels the upstream** once it has a value. That is what makes it safe on an infinite or hot flow, and it is the idiomatic way to read a current value from something you do not want to keep collecting:
val settings = dataStore.data.first() // one read, then done
**toList() on a hot flow never returns.** A StateFlow or a SharedFlow has no end, so collecting it into a list suspends forever. This is a real bug people hit in tests, where the fix is take(n).toList() or a Turbine assertion instead.
val all = viewModel.state.toList() // hangs: a StateFlow never completes
Underneath all of it, Flow is one of the smallest interfaces in the standard library:
public interface Flow<out T> {
public suspend fun collect(collector: FlowCollector<T>)
}
public fun interface FlowCollector<in T> {
public suspend fun emit(value: T)
}
That is the whole thing. Two interfaces, one method each, both suspending.
Everything else is built on top. An intermediate operator is a function that returns a new Flow whose collect calls the previous flow's collect with a wrapped collector:
// Roughly what map is
fun <T, R> Flow<T>.map(transform: suspend (T) -> R): Flow<R> = flow {
collect { value -> emit(transform(value)) }
}
Read that twice and the whole library stops being a list of names to memorise. map collects the upstream, transforms each value, and emits it into whatever is collecting **it**. Operators compose because each is a flow wrapping a flow.
It also explains two things you have already met. **Coldness** is not a feature that had to be implemented: the producer block simply lives inside collect, so calling collect twice runs it twice. And **backpressure** is not a feature either: emit is suspending, so a slow collector suspends the producer automatically.
The interview version: a Flow is a suspending function that takes a callback. Everything else is convenience on top.
Sequence and Flow look alike and differ on exactly one axis: **suspension.**
// Sequence: lazy, but blocking
val s: Sequence<Int> = sequence {
yield(1)
Thread.sleep(100) // blocks the thread
yield(2)
}
// Flow: lazy, and suspending
val f: Flow<Int> = flow {
emit(1)
delay(100) // suspends, thread is free
emit(2)
}
Both are cold and lazy, both compute values on demand rather than up front, and both have nearly identical operator names. The difference is that a Sequence produces values on the calling thread and any wait inside it blocks that thread, whereas a Flow can suspend.
The rule that follows: **use Sequence for lazy CPU work, use Flow when anything waits.** Parsing a large file line by line without loading it all into memory is a Sequence. The same parse where each line triggers a network lookup is a Flow.
The compiler enforces this from one side: you cannot call a suspend function inside sequence { }, because yield is a restricted suspension point. That error is the language telling you which of the two you actually needed.
sequence {
yield(api.fetch()) // does not compile: suspend call not allowed here
}
Finally, the design question this topic exists to answer: **should this function return a value, or a Flow?**
suspend fun getUser(id: String): User // one value, once
fun observeUser(id: String): Flow<User> // a value now and whenever it changes
Return a plain suspend function when the caller asks a question once and the answer does not change underneath them: a form submission, a one-off lookup, an upload.
Return a Flow when there is genuinely **more than one meaningful value over time**: a Room query that should update the UI as the table changes, a location stream, search results arriving progressively, a WebSocket.
The failure mode in each direction is worth naming, because interviewers probe both.
**Flow where a value would do** is the more common one. fun getUser(id): Flow<User> that emits once and completes gives the caller a stream to collect, a lifecycle to manage, and an operator chain to reason about, all to deliver one value. It is a tell of someone applying a pattern rather than choosing one.
**A value where a Flow was needed** shows up as manual refreshing: a screen that calls getUsers() again after every insert, because nothing tells it the data changed. The moment you find yourself re-fetching to stay in sync, the source should have been observable.
The one nuance: a Flow that emits many values and then **completes**, such as paginated results, is perfectly legitimate. Flow does not imply infinite, it implies more than one.