TopolearnFlows are the standard way to stream data in Kotlin: database updates, UI state, websockets and paging.
A Flow<T> is an asynchronous stream of values built on coroutines. Where a suspend function returns one value, a flow emits many over time.
val numbers: Flow<Int> = flow { for (i in 1..3) { delay(100); emit(i) } }Flows are cold: the block doesn't run until someone calls a terminal operator like collect { }, toList(), first() or fold, and each collector gets its own execution.
Intermediate operators look like collection operations but are lazy and suspend-friendly: map, filter, take, transform, onEach, distinctUntilChanged, debounce, combine, zip, flatMapLatest. catch { } handles upstream errors; onCompletion runs at the end.
Context: a flow runs in the collector's coroutine. To move upstream work to another dispatcher, use flowOn(Dispatchers.IO); never call withContext around emit. buffer() and conflate() decouple slow collectors from fast producers.
Hot flows exist independently of collectors:
StateFlow<T>always holds a current value (.value), emits only distinct updates, and new collectors get the latest value immediately. It's the standard holder for UI state (MutableStateFlowinside, exposed as read-onlyStateFlow).SharedFlow<T>broadcasts events to all current subscribers, with configurable replay.
Convert cold to hot with stateIn / shareIn in a scope.
Example
import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*
fun ticker(n: Int): Flow<Int> = flow {
for (i in 1..n) { delay(50); emit(i) }
}
fun main() = runBlocking {
ticker(5)
.filter { it % 2 == 1 }
.map { it * it }
.onEach { print("[$it] ") }
.collect() // [1] [9] [25]
println()
println(ticker(3).toList()) // [1, 2, 3]
val failing = flow { emit(1); error("boom") }
failing.catch { emit(-1) }.collect { print("$it ") } // 1 -1
println()
val state = MutableStateFlow(0)
val watcher = launch { state.collect { println("state=$it") } }
delay(10); state.value = 1; delay(10)
state.value = 1 // same value: not re-emitted
state.value = 2; delay(10)
watcher.cancel() // state=0 / state=1 / state=2
}Remember
- flow { emit(x) } is cold: it runs per collector on collect
- Operators like map/filter/catch are lazy and suspend-friendly
- flowOn changes the upstream dispatcher; don't use withContext around emit
- StateFlow holds current state; SharedFlow broadcasts events
Watch out for
- Collecting a flow in a scope that is never cancelled, leaking a collector forever.