Kotlin Flows

Handle streams of values over time with Kotlin Flow: building flows, collecting them, transforming with map and filter, combining flows, handling errors, and holding state with StateFlow.

Lesson 23 of 43Advanced17 min read

A suspend function returns one value, eventually. But many things produce a series of values over time: the location of a phone as it moves, search results as the user types, messages arriving in a chat, progress updates from a download, or a database query whose results change. For these, Kotlin has Flow.

A Flow<T> is like a list whose items arrive one at a time, possibly with waits in between. You transform it with familiar operators such as map and filter, and you collect it to receive the values. Flows are built on coroutines, so waiting never blocks a thread, and collection is cancelled automatically when its scope ends.

In this lesson you'll build flows, collect them, transform and combine them, handle errors, move work to the right dispatcher, and use StateFlow, the standard way Android apps hold screen state that the UI observes.

Building and collecting flows

Kotlin
fun countdown(from: Int): Flow<Int> = flow {
    for (i in from downTo 1) {
        emit(i)          // send a value to the collector
        delay(1000)
    }
}

countdown(3).collect { println(it) }
  • The flow { } builder runs its block and emits values.
  • collect is a suspend function that receives each value as it arrives, so it's called from a coroutine.
  • Flows are cold: the code inside doesn't run until someone collects, and it runs again for each collector.
  • Quick builders: flowOf(1, 2, 3) and listOf(...).asFlow().

Operators

Intermediate operators return a new flow and run lazily, as each value passes through:

OperatorDoes
map { }Transforms each value
filter { }Keeps matching values
take(n)Stops after n values
onEach { }Runs a side effect, such as logging
transform { }Emits zero, one or many values per input
debounce(ms)Waits for a pause, ideal for search boxes

Terminal operators start the flow and wait for it: collect, toList(), first(), reduce { } and count().

Errors, dispatchers and combining

  • catch { e -> emit(fallback) } handles exceptions thrown upstream (before the catch in the chain).
  • flowOn(Dispatchers.IO) runs the upstream work on another dispatcher, while collection stays where it was.
  • combine(a, b) { x, y -> ... } emits whenever either flow emits, using the latest value of each, which is perfect for joining a search term with a filter setting.
  • zip(a, b) pairs values one-to-one, waiting for both.

StateFlow and SharedFlow

Ordinary flows are cold and start fresh for each collector. Hot flows exist independently of collectors:

  • StateFlow<T> always holds a current value. New collectors immediately receive the latest value, then every update. Create one with MutableStateFlow(initial) and change it with .value = ... or update { }.
  • SharedFlow<T> broadcasts events to all current collectors, with no current value, which suits one-off events like "show a message".

A common pattern exposes a read-only StateFlow and keeps the MutableStateFlow private, just like read-only lists.

Kotlin Flows Syntax

Flow essentials

import kotlinx.coroutines.flow.*

fun numbers(): Flow<Int> = flow { emit(1); emit(2) }
numbers().map { it * 2 }.filter { it > 2 }.collect { println(it) }
flowOf("a", "b").toList()
flow.catch { emit(default) }.flowOn(Dispatchers.IO)
combine(flowA, flowB) { a, b -> a + b }

private val _state = MutableStateFlow(0)
val state: StateFlow<Int> = _state
_state.value = 5
_state.update { it + 1 }

Kotlin Flows Summary

  • A Flow emits a series of values over time; flow { emit(x) } builds one and collect receives values.
  • Flows are cold: their code runs for each collector, only when collected.
  • Intermediate operators like map, filter, take and onEach build pipelines; collect, toList and first start them.
  • catch handles upstream errors, flowOn changes the upstream dispatcher, combine merges latest values.
  • StateFlow holds current state for observers; SharedFlow broadcasts events.

Kotlin Flows Examples

Each example shows the code, the output it produces and what each part does.

Example 1

Building, transforming and collecting

Kotlin
import kotlinx.coroutines.delay
import kotlinx.coroutines.flow.*
import kotlinx.coroutines.runBlocking

fun downloads(): Flow<Int> = flow {
    for (percent in 0..100 step 25) {
        delay(100)
        emit(percent)
    }
}

fun main() = runBlocking {
    downloads()
        .filter { it > 0 }
        .map { "Downloaded $it%" }
        .collect { println(it) }

    val total = flowOf(4, 8, 15).reduce { a, b -> a + b }
    println("Sum: $total")
}
Output
Downloaded 25%
Downloaded 50%
Downloaded 75%
Downloaded 100%
Sum: 27
Line-by-line explanation
  1. Lines 5 to 10A cold flow that emits progress every 100 ms.
  2. Lines 13 to 16A pipeline: drop 0%, turn numbers into messages, then collect them one at a time.
  3. Line 18reduce is a terminal operator that combines all the values into one.
Example 2

Errors and fallbacks

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

fun prices(): Flow<Double> = flow {
    emit(2.5)
    emit(3.75)
    throw IllegalStateException("Connection lost")
}

fun main() = runBlocking {
    prices()
        .map { "£$it" }
        .catch { e ->
            println("Error: ${e.message}")
            emit("(prices unavailable)")
        }
        .collect { println(it) }
    println("Done")
}
Output
£2.5
£3.75
Error: Connection lost
(prices unavailable)
Done
Line-by-line explanation
  1. Line 7The flow fails after emitting two values.
  2. Lines 13 to 16catch handles the exception and emits a fallback value instead of crashing.
  3. Line 18The program continues normally after collection.
Example 3

StateFlow for a counter

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

class CartModel {
    private val _count = MutableStateFlow(0)
    val count: StateFlow<Int> = _count.asStateFlow()

    fun add() = _count.update { it + 1 }
}

fun main() = runBlocking {
    val cart = CartModel()
    val watcher = launch {
        cart.count.collect { println("Items in cart: $it") }
    }
    delay(50)
    cart.add()
    delay(50)
    cart.add()
    delay(50)
    println("Current value: ${cart.count.value}")
    watcher.cancel()
}
Output
Items in cart: 0
Items in cart: 1
Items in cart: 2
Current value: 2
Line-by-line explanation
  1. Lines 5 to 6The mutable state is private; others see a read-only StateFlow.
  2. Line 8update changes the value safely, even from several coroutines.
  3. Lines 13 to 15Collect in its own coroutine, because a StateFlow never completes.
  4. Line 22Cancel the collector when it's no longer needed.

Common Kotlin Flows Mistakes

Expecting a flow to run without collecting

Don't
downloads().map { println(it) }      // nothing happens: no terminal operator
Do
downloads().collect { println(it) }

Code after collecting a StateFlow

Don't
state.collect { render(it) }
println("Never printed")             // StateFlow never completes
Do
launch { state.collect { render(it) } }
println("Printed straight away")

Exposing MutableStateFlow publicly

Don't
val count = MutableStateFlow(0)      // any code can change the state
Do
private val _count = MutableStateFlow(0)
val count: StateFlow<Int> = _count.asStateFlow()

Kotlin Flows in Real Projects

Flows in Android apps

Modern Android architecture uses a StateFlow<UiState> in each ViewModel, which the screen collects to redraw itself. Room returns flows from database queries, so lists update automatically when data changes, and DataStore exposes settings as flows. Search boxes use debounce and flatMapLatest to query only after the user pauses typing and to cancel outdated searches.

Kotlin Flows FAQ

What is the difference between a Flow and a Sequence in Kotlin?

Both produce values lazily, one at a time. A Sequence is synchronous: it can't call suspend functions, so it suits in-memory work such as processing a large list. A Flow can suspend between values, which suits values that arrive over time, such as network responses, database updates or user input.

When should I return a Flow instead of a suspend function?

Use a suspend function when there's exactly one result, such as loading a user once. Return a Flow when there are several values over time, or when callers should be told about every change, such as a list of messages that updates as new ones arrive.

Does StateFlow emit the same value twice?

No. Setting value to something equal to the current value does nothing, because StateFlow compares with equals and skips duplicates. Collectors also only see the latest value if several updates happen quickly. For events that must all be delivered, even repeated ones, use a SharedFlow instead.

How do I convert a Flow to a List?

Call the terminal operator toList() from a coroutine: val items = flow.toList(). It collects every value and returns once the flow completes, so only use it on flows that finish. On a StateFlow, which never completes, it would wait forever; read .value or use first() instead.

Why do I get "Flow invariant is violated" in Kotlin?

A flow { } builder must emit from the same coroutine context it's collected in, so calling emit inside withContext(Dispatchers.IO) or a launch throws this error. To run the producing code on another dispatcher, add flowOn(Dispatchers.IO) after the builder. If you need to emit from several coroutines at once, build the flow with channelFlow { } and send.

Kotlin Flows Exercises

Answer each one and check it instantly. If you get stuck, open the hint before the answer.

Fill in the blankseasy

Complete the flow builder so it sends the value 42 to collectors.

val f = flow {
    (42)
}
Multiple choiceeasy

When does the code inside flow { } run?

Predict the outputmedium

What does this print?

runBlocking {
    flowOf(1, 2, 3, 4).filter { it % 2 == 0 }.map { it * 10 }.collect { print("$it ") }
}
Multiple choicemedium

What's the key difference between StateFlow and a normal Flow?

Multiple choicehard

Which operator should a search box use so it only searches after the user stops typing for a moment?

Kotlin Flows Quiz

5 questions to check what you learned.

  1. Which function receives the values from a flow?

    • emit
    • await
    • collect
    • launch

  2. What does catch { } in a flow handle?

    • Exceptions anywhere in the program
    • Only exceptions in collect
    • Exceptions thrown upstream, before it in the chain
    • Timeouts only

  3. What does flowOn(Dispatchers.IO) do?

    • Changes where collect runs
    • Runs the upstream flow work on the IO dispatcher
    • Converts the flow to a list
    • Cancels the flow

  4. How do you change the value of a MutableStateFlow called state?

    • state.emit = 3
    • state.set(3)
    • state = 3
    • state.value = 3, or state.update { 3 }

  5. What does combine(a, b) { x, y -> x + y } emit?

    • A new value whenever either flow emits, using the latest value from each
    • Only once, at the end
    • Pairs strictly one-to-one
    • Nothing until both complete

Related Kotlin topics

22CoroutinesWrite asynchronous code that reads like normal code with coroutines: suspend functions, runBlocking, launch, async and await, delay, structured concurrency, dispatchers and timeouts.11Collection OperationsTransform, filter, sort, group and summarise collections in one readable line with map, filter, sumOf, sortedBy, groupBy, associate and friends.12LambdasWrite functions as values with lambdas, pass them to other functions, use it and trailing lambda syntax, return functions, and reference existing functions with ::.
Esc
↑↓ to navigateEnter to open