Flow: Asynchronous Streams

A suspend function returns one value. A Flow emits many values over time — search results as the user types, live prices, progress updates, rows streamed from a database.

Building and collecting a flow

import kotlinx.coroutines.delay
import kotlinx.coroutines.flow.Flow
import kotlinx.coroutines.flow.filter
import kotlinx.coroutines.flow.flow
import kotlinx.coroutines.flow.map
import kotlinx.coroutines.flow.toList
import kotlinx.coroutines.runBlocking

fun countdown(from: Int): Flow<Int> = flow {
    for (i in from downTo 1) {
        delay(100)
        emit(i)
    }
}

fun main() = runBlocking {
    countdown(3)
        .map { it * 10 }
        .filter { it > 10 }
        .collect { println(it) }

    println(countdown(5).toList())
}

30
20
[5, 4, 3, 2, 1]

• flow { } builds a flow; emit sends a value.
• Operators like map and filter work like their collection counterparts, but on values as they arrive.
• collect is a terminal operator: it starts the flow and handles each value.

Flows are cold: nothing runs until someone collects, and each collector gets its own fresh run — which is why countdown(5) starts again from 5.

Useful operators

import kotlinx.coroutines.flow.catch
import kotlinx.coroutines.flow.flow
import kotlinx.coroutines.flow.onCompletion
import kotlinx.coroutines.flow.take
import kotlinx.coroutines.flow.flowOf
import kotlinx.coroutines.flow.zip
import kotlinx.coroutines.runBlocking

fun main() = runBlocking {
    flowOf(1, 2, 3, 4, 5).take(2).collect { println("take: $it") }

    flowOf("a", "b").zip(flowOf(1, 2)) { letter, n -> "$letter$n" }
        .collect { println("zip: $it") }

    flow {
        emit(1)
        throw IllegalStateException("sensor failed")
    }
        .catch { e -> println("caught: ${e.message}") }
        .onCompletion { println("done") }
        .collect { println("value: $it") }
}

take: 1
take: 2
zip: a1
zip: b2
value: 1
caught: sensor failed
done

Other everyday operators: debounce (wait for a pause — ideal for search boxes), distinctUntilChanged, flatMapLatest (cancel the previous search when a new query arrives), combine, and flowOn (run the upstream work on another dispatcher).

Where flows come from

You'll mostly consume flows produced by libraries: Room database queries can return a Flow that re-emits whenever the data changes, and DataStore exposes preferences as a flow.

Common mistakes

• Forgetting the terminal operator. A chain of map and filter with no collect (or toList, first, ...) does nothing.
• Using withContext inside flow { } to change threads — it's not allowed there. Use the flowOn operator instead.