Kotlin Flow | Ultimate Handbook

🌀

Cold Flows

runs per collector · nothing until collect

Coroutines introduces Flow<T> as an asynchronous stream of values; "cold" is the specific property that the producer block doesn't execute at all until a collector actually calls collect, and it runs again — independently, from the very start — for every new collector. This is fundamentally different from a hot stream like a Channel, where values are produced regardless of whether anyone's listening and a late collector simply misses whatever already happened.

  val numbers = flow {
      println("starting")
      emit(1)
      emit(2)
  }
  // Nothing printed yet — defining a flow does no work

  numbers.collect { println(it) }
  // starting
  // 1
  // 2

  numbers.collect { println(it) }
  // starting        ← runs again, from scratch, for this NEW collector
  // 1
  // 2

💡

This mirrors Sequences' lazy evaluation almost exactly, just extended into the async world: nothing happens until a terminal step (there, a terminal operation; here, collect) pulls values through the pipeline. Being cold is what makes a Flow safe to define once and reuse — like a request builder — without the producer's side effects firing before you're ready for them.

🏗

Flow Builders

flow { } · flowOf · asFlow

The general-purpose builder is flow { }, which lets you call emit from arbitrary suspend code — including loops, conditionals, and calls to other suspending functions. For the common cases of a fixed set of values or converting an existing collection, flowOf and .asFlow() are shorter and communicate intent more directly.

flow { }, flowOf, and asFlow Builders

import kotlinx.coroutines.flow.*
import kotlinx.coroutines.delay

// General-purpose: full suspend function body, arbitrary control flow
fun pollStatus(): Flow<String> = flow {
    repeat(3) { attempt ->
        delay(1000)
        emit("attempt $attempt: checking...")
    }
    emit("done")
}

// flowOf: a fixed, known set of values
val fixed: Flow<Int> = flowOf(1, 2, 3)

// asFlow: converting an existing collection or range into a Flow
val fromRange: Flow<Int> = (1..5).asFlow()

⚙️

Flow Operators

map · filter · take · intermediate, lazy

Flow's operators are intentionally named after the same Collections API operations — map, filter, take, onEach — and behave the same conceptually, except each step can itself suspend, and nothing executes until a terminal operation like collect pulls values through the whole chain, same as an intermediate operation on a Sequence.

Chaining operators, lazily map / filter / onEach

import kotlinx.coroutines.flow.*
import kotlinx.coroutines.runBlocking

fun main() = runBlocking {
    (1..10).asFlow()
        .filter { it % 2 == 0 }
        .map { it * it }
        .onEach { println("processing: $it") }  // side effect, passes value through unchanged
        .take(2)
        .collect { println("collected: $it") }
}
// processing: 4     collected: 4
// processing: 16    collected: 16
// take(2) means the source flow stops emitting after this —
// 6, 8, 10 never get filtered/mapped/onEach'd at all

🔀

Context & flowOn

upstream dispatcher · context preservation

A Flow normally runs in whatever coroutine context calls collect — this is called context preservation, and it's a deliberate design choice so a flow's behavior doesn't silently change depending on which dispatcher happens to collect it. flowOn is the explicit, visible way to override the dispatcher for everything upstream of it in the chain, without affecting the collector's own context.

Running the producer on a different dispatcher flowOn

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*

fun readLinesFromDisk(): Flow<String> = flow {
    File("large.txt").forEachLine { emit(it) }  // blocking I/O — belongs on Dispatchers.IO
}
    .flowOn(Dispatchers.IO)
    // Everything above flowOn() runs on Dispatchers.IO;
    // everything below (the collector) runs on ITS OWN dispatcher,
    // unaffected — flowOn only changes the upstream context

fun main() = runBlocking {
    readLinesFromDisk()
        .filter { it.isNotBlank() }
        .collect { println(it) }  // runs on runBlocking's own dispatcher
}

🚨

Exception Handling

catch · try/catch around collect

An exception thrown anywhere upstream in a flow chain propagates downstream and cancels the flow — a plain try/catch wrapped around the whole collect call handles it, same as ordinary suspend code. The dedicated catch operator does the same job but as part of the chain itself, which lets you emit a fallback value or transform the failure without breaking out of the surrounding code structure.

catch operator vs try/catch catch

import kotlinx.coroutines.flow.*
import kotlinx.coroutines.runBlocking

fun riskyFlow(): Flow<Int> = flow {
    emit(1)
    emit(2)
    throw RuntimeException("upstream failure")
}

fun main() = runBlocking {
    // catch: part of the chain, can emit a recovery value
    riskyFlow()
        .catch { e -> emit(-1) }   // runs only for exceptions from UPSTREAM of this line
        .collect { println(it) }
    // 1
    // 2
    // -1
}

⚠️

The catch operator only catches exceptions thrown upstream of where it's placed in the chain — an exception thrown inside the collect block itself is not caught by a catch operator positioned before it. Wrap the collector logic in an ordinary try/catch instead if the failure originates there.

📡

StateFlow vs SharedFlow

hot flows · current value · replay

Both are hot flows — unlike a plain flow { }, they exist and can emit independently of whether anyone is collecting, and multiple collectors share the same running stream instead of each triggering their own independent execution. StateFlow<T> always holds exactly one current value, readable synchronously via .value without collecting at all — it's built for observable state. SharedFlow<T> is the more general form: no built-in "current value," configurable replay of past emissions to new collectors, and support for values with no natural "current" reading, like one-off events.

StateFlow: observable current value StateFlow

import kotlinx.coroutines.flow.MutableStateFlow
import kotlinx.coroutines.flow.StateFlow

class CounterViewModel {
    private val _count = MutableStateFlow(0)
    val count: StateFlow<Int> = _count.asStateFlow()  // read-only view, same split idea as MutableList/List

    fun increment() {
        _count.value++
    }
}

val vm = CounterViewModel()
println(vm.count.value)  // 0: read synchronously, no collect needed
vm.increment()
println(vm.count.value)  // 1

// A collector also sees every subsequent update, plus the CURRENT
// value immediately upon starting to collect — that's the "state" part

SharedFlow: general-purpose event stream SharedFlow

import kotlinx.coroutines.flow.MutableSharedFlow

class EventBus {
    // replay = 0: a new collector only sees FUTURE events,
    // not whatever already happened before it subscribed
    private val _events = MutableSharedFlow<String>(replay = 0)
    val events = _events.asSharedFlow()

    suspend fun publish(event: String) {
        _events.emit(event)
    }
}
// Good fit for one-off signals: "show a toast", "navigate to screen" —
// events that don't make sense to model as a persistent "current value"

💡

Default to StateFlow whenever there's a meaningful "current value" to expose — UI state, a loaded resource, a toggle. Reach for SharedFlow for discrete events that don't have a sensible "current" reading, or when you need explicit control over replay behavior for late subscribers.

📥

Collecting

collect · first · toList · launchIn

collect is Flow's primary terminal operation — a suspend function that runs until the flow completes, invoking a lambda for every emitted value. A handful of other terminal operations exist for common patterns: first collects just the first value and cancels, toList gathers every value into a List, and launchIn starts collection inside a given CoroutineScope without needing an enclosing suspend function.

Terminal operations beyond collect first / toList / launchIn

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*

fun main() = runBlocking {
    val numbers = (1..5).asFlow()

    // first: collects one value, then stops — like Sequence's first()
    println(numbers.first())  // 1

    // toList: gathers everything into a List
    println(numbers.toList())  // [1, 2, 3, 4, 5]

    // launchIn: starts collection as a new coroutine in the given scope,
    // instead of suspending the current one — fire-and-forget within a scope
    numbers
        .onEach { println("launched collect: $it") }
        .launchIn(this)  // `this` refers to the runBlocking CoroutineScope
}

📋

Quick Reference

Syntax cheat-sheet

Concept	Syntax	Notes
General builder	flow { emit(x) }	Full suspend body; runs fresh per collector
Fixed values	flowOf(1, 2, 3)	Shorthand for a known, static set of values
From a collection	list.asFlow()	Converts an existing Iterable
Transform / filter	flow.map { } / flow.filter { }	Lazy: no work until collected
Side effect mid-chain	flow.onEach { }	Passes the value through unchanged
Switch upstream dispatcher	flow.flowOn(Dispatchers.IO)	Only affects code above it in the chain
Recover from upstream failure	flow.catch { emit(fallback) }	Only catches exceptions from upstream, not from collect
Observable current value	MutableStateFlow(initial)	Hot; .value readable synchronously
General event stream	MutableSharedFlow(replay = n)	Hot; configurable replay for late subscribers
Collect all values	flow.collect { }	Suspends until the flow completes
First value only	flow.first()	Collects one, then cancels
Gather into a List	flow.toList()	Materializes every emission
Collect in a scope	flow.launchIn(scope)	Starts a new coroutine instead of suspending the caller