A Flow represents sequential values that become available over time. Use it when a suspending function would otherwise wait to return a complete batch, but consumers can act on each value as soon as it is ready—for progressive loading, event streams, subscription-style APIs, database work, or HTTP-driven processing. Flows build on Kotlin coroutines and resemble Reactive Streams abstractions such as RxJava and Project Reactor. section 16.1 section 16.0
Emit and collect values incrementally
Create a cold Flow with flow { ... }, publish each value with emit, and start execution with collect: fun createValues(): Flow<Int> = flow { emit(1) delay(1.seconds) emit(2) delay(1.seconds) emit(3) } runBlocking { createValues().collect { value -> log(value) } } Unlike returning List<Int>, this logs 1 before later values have been computed. emit suspends until the collector has processed the value, so producer and collector proceed in an intertwined sequence. section 16.1.1 section 16.2.2
Listing 16.2 Creating and collecting a flow
import kotlinx.coroutines.delay
import kotlinx.coroutines.runBlocking
import kotlinx.coroutines.flow.*
import kotlin.time.Duration.Companion.milliseconds
fun createValues(): Flow<Int> {
return flow {
emit(1) #1
delay(1000.milliseconds)
emit(2)
delay(1000.milliseconds)
emit(3)
delay(1000.milliseconds)
}
}
fun main() = runBlocking {
val myFlowOfValues = createValues()
myFlowOfValues.collect { log(it) } #2
}
// 29 [main @coroutine#1] 1
// 1100 [main @coroutine#1] 2
// 2156 [main @coroutine#1] 3
Compare a Flow producer with collection: emitted values reach collect at their individual timestamps rather than after all computation finishes. section 16.1.1
A flow created with flow is cold: constructing it returns an inert Flow<T>, without starting its builder. A terminal operation such as collect triggers the builder; therefore a regular, non-suspending function can return a flow whose body calls suspending operations. The same flow can also be collected repeatedly, but each collection reruns its code and repeats side effects such as network requests. section 16.2.1 section 16.2.2
Choose the flow category by who controls production:
- Use a cold flow for a service-like operation, such as a network request or database read. Each collector initiates its own execution and receives all emissions. section 16.1.2 section 16.3.4
- Use SharedFlow for broadcast-style values produced independently of subscribers. Keep MutableSharedFlow private and expose SharedFlow to let consumers subscribe without emitting. section 16.3 section 16.3.1
- Use StateFlow to communicate changing state. It emits only when its value changes according to equality, while shared flows broadcast values; convert a cold flow with shareIn or stateIn when appropriate. section 16.4
Flow pipelines stay lazy through intermediate operators. Operators such as map, filter, and onEach transform an upstream flow into a downstream flow, but do not run until collection. flowOn changes the coroutine context only for the upstream portion of a pipeline; collection otherwise runs in the context where collect was invoked. section 17.2 section 17.2.7
Build a cold custom operator
A custom intermediate operator collects its upstream flow inside a new flow builder and emits downstream values. This moving average remains cold because upstream collection starts only when a caller collects its result: fun Flow<Double>.averageOfLast(n: Int): Flow<Double> = flow { val numbers = mutableListOf<Double>() collect { if (numbers.size >= n) numbers.removeFirst() numbers.add(it) emit(numbers.average()) } } section 17.3
Under the hood, the basic cold-flow contract is small: Flow defines suspending collect, and FlowCollector defines suspending emit. Calling collect runs the builder; emit invokes the collector lambda; when that lambda finishes, execution returns to the builder. This lightweight model usually runs within one coroutine unless a different mechanism is introduced. section 16.2.4
