sdlc.dev

Kotlin path · Coroutines from the inside · Topic 13 of 13

Flow: cold streams, operators and context

Not started

A suspend function returns one value. A Flow<T> returns many, over time: the asynchronous version of a Sequence. Like a sequence it is **cold**: building one does nothing, and each terminal operation (collect, first, toList) runs the producer again from the start. Unlike a sequence, the producer can suspend between values, to wait for a network page or a database change.

Why it matters: Flow is how Kotlin code returns many values over time. Knowing that it is cold, sequential and context-preserving explains where to put flowOn, buffer and catch.

  1. 1Cold by default

    A flow { } block runs from the start for each collector, and only while it is being collected.

  2. 2Emit waits for collect

    Producer and collector run in the same coroutine by default, so emit suspends until the collector has handled the value: built-in backpressure.

  3. 3Context is preserved

    A flow runs in the collector's context; flowOn changes the context of the operators above it, never below.

  4. 4Hot flows are different

    StateFlow and SharedFlow exist independently of collectors and share one stream between many.

1

How it works

The explanation, with examples you can run.

Cold and sequential

import kotlinx.coroutines.*import kotlinx.coroutines.flow.* fun ticks(): Flow<Int> = flow {    println("producer starts")    for (i in 1..3) {        delay(100)        emit(i)                       // suspends until the collector is done with i    }} fun main() = runBlocking {    val doubled = ticks().map { it * 2 }      // nothing printed yet    doubled.collect { println("got $it") }    // producer starts, got 2, got 4, got 6    println(doubled.first())                  // producer starts again, prints 2}

By default, producer and collector run in one coroutine, taking turns: emit calls straight into the collector and returns when the collector's block has finished. A slow collector therefore slows the producer down, which is backpressure for free. Operators such as map, filter, take and onEach are cheap wrappers around the next stage; take(n) cancels the producer once it has enough.

Context preservation and flowOn

A flow runs in the context of the code that collects it. You may not switch context and emit from there (withContext(IO) { emit(x) } throws). To move the producing side to another dispatcher, use flowOn:

val rows: Flow<Row> = flow { readRowsBlocking().forEach { emit(it) } }    .map(::parse)    .flowOn(Dispatchers.IO)           // flow { } and map run on IO// .collect { render(it) }            // still runs in the collector's context

flowOn changes only what is above it. Under the hood it starts the upstream in a separate coroutine and passes values through a channel, so the two sides now run concurrently.

Buffering and dropping

When producer and consumer run at different speeds you can decide what happens:

  • buffer(n): let the producer run up to n values ahead.
  • conflate(): keep only the latest value if the collector is busy (good for UI state).
  • collectLatest { }: cancel the handling of a value when a newer one arrives.
  • debounce, sample: time-based thinning for search boxes and sensors.

Errors and completion

Exceptions in the producer or operators travel downstream to the collector. catch { } handles exceptions from upstream only, and can emit a fallback; onCompletion { } runs at the end, with or without an error. Wrap the collect call in a normal try for errors in the collector itself. Cancellation of the collecting coroutine cancels the whole chain.

Hot flows

StateFlow (always has a current value, conflated, replays it to new collectors) and SharedFlow (broadcast to any number of collectors, configurable replay and buffer) are hot: they exist and hold values whether or not anyone is collecting. stateIn and shareIn turn a cold flow into a hot one inside a scope. For values from callbacks, callbackFlow { } with awaitClose { } bridges listener APIs; for several producers feeding one consumer, channelFlow or a Channel is the tool.

2

Check yourself

3 questions. Get them all right to finish this topic.

  1. 1.val f = flow { println("start"); emit(1) }. Nothing collects f yet. What has been printed?
  2. 2.In flow.map { parse(it) }.flowOn(Dispatchers.Default).collect { show(it) }, where does show run?
  3. 3.Why does calling emit from inside withContext(Dispatchers.IO) { } in a flow { } builder fail?

0 of 3 right so far. Get all three to finish this topic.

Go deeper