Flows

This section documents Kotlin 2.4.x on the JVM, as published at kotlinlang.org, which is the reference these pages are written and verified against.

This content was generated with the assistance of AI and should be verified against kotlinlang.org before being relied on in production.

This section’s bibliography lists the reference material consulted while preparing these pages.

Where a suspend function produces exactly one value asynchronously, Flow<T> produces zero or more values over time — Kotlin’s coroutine-based answer to a reactive stream (comparable in spirit to Project Reactor’s Flux, but built entirely on suspend functions rather than a separate reactive-streams runtime).

Cold Flow Basics

A Flow is cold: its producing code does not run at all until something actually collects it, and it runs again, from scratch, for every new collector:

fun numbers(): Flow<Int> = flow {          // the "flow { }" builder -- nothing runs yet
    println("starting to emit")
    for (i in 1..3) {
        delay(100)
        emit(i)                              // suspends until the collector is ready for the next value
    }
}

suspend fun main() {
    val stream = numbers()                    // still nothing has run
    println("about to collect")
    stream.collect { value -> println("got $value") }   // NOW "starting to emit" prints, then values arrive
}

This is the direct analogue of `Sequence’s laziness (Collections and Sequences) applied to values that arrive over time rather than all at once.

sequenceDiagram participant Caller participant Flow as numbers(): Flow Caller->>Flow: numbers() — builder called Note over Flow: nothing runs yet — cold Caller->>Flow: .collect { ... } activate Flow Flow-->>Caller: emit(1) Flow-->>Caller: emit(2) Flow-->>Caller: emit(3) deactivate Flow Note over Caller,Flow: producing code only ran because collect() was called

Flow Builders and Common Operators

Besides flow { }, flowOf(1, 2, 3) and (1..3).asFlow() build a Flow from fixed values or an existing collection/range. Intermediate operators like map/filter mirror their Sequence/collection counterparts, and are themselves suspending-aware:

flowOf(1, 2, 3, 4, 5)
    .filter { it % 2 == 0 }
    .map { it * it }
    .collect { println(it) }        // 4, then 16

// each intermediate step can itself suspend
flow { emit(fetchUserId()) }
    .map { id -> fetchUser(id) }     // "map"'s transform is itself a suspend lambda
    .collect { user -> render(user) }

collect is the primary terminal operator — the point where the whole upstream pipeline actually runs; a Flow with no terminal operator called on it does nothing at all, same as an unconsumed Sequence.

StateFlow vs. SharedFlow

Both are hot flows — unlike flow { }, they run independently of whether anything is collecting, and multiple collectors share the same running producer instead of each triggering it from scratch:

Type Use

StateFlow<T>

always holds a current value (.value), conflates rapid updates (a slow collector only ever sees the latest value, not every intermediate one), and always has at least one value — the natural fit for UI state (the ViewModel pattern on Kotlin for Android uses it extensively).

SharedFlow<T>

a more general broadcast stream with configurable replay and no requirement to hold a "current" value — the right fit for one-off events (a snackbar message, a navigation command) that should not be replayed as "state" the way StateFlow would.

 
class CounterViewModel {
    private val _count = MutableStateFlow(0)
    val count: StateFlow<Int> = _count.asStateFlow()    // read-only view exposed publicly

    fun increment() {
        _count.value += 1                                 // every collector sees the new value
    }
}

Buffering

By default, a slow collector makes the upstream producer suspend until it catches up. buffer() decouples them with an internal channel, letting the producer run ahead; conflate() goes further and drops intermediate values entirely, keeping only the latest when the collector is behind — the same conflation StateFlow applies automatically:

flow {
    for (i in 1..100) {
        delay(10)          // fast producer
        emit(i)
    }
}
.buffer()                  // producer no longer waits on a slow collector, up to the buffer's capacity
.collect { slowCollector(it) }

See Also

References