Kotlin Flows in Practice

A suspend function returns one value after waiting. But many things in an app produce values over time: a database that changes as the user adds rows, a countdown timer, a stream of GPS locations, a search box where the user keeps typing. For those you need something that can deliver a sequence of values, and in Kotlin that is a Flow.

Flows are the glue of modern Android architecture. Room returns them (lesson 34), DataStore returns them (lesson 35), and your ViewModel exposes UI state through them (lessons 24 and 25). This lesson makes sure you understand them properly instead of copying the pattern.

This lesson covers the two items of Pathway 2, Unit 6 in the official Android Basics with Compose course that build on flows: Kotlin Flows in Practice and Testing Flows. Most examples run in the free Kotlin Playground (https://play.kotlinlang.org), so you can experiment on any computer.

By the end you will be able to:

• Explain the difference between a suspend function, a List, a Sequence and a Flow.
• Build a flow with flow { emit() }, and collect it.
• Use operators: map, filter, onEach, take, catch, onCompletion, combine, debounce, flatMapLatest.
• Explain cold versus hot flows, and StateFlow versus SharedFlow.
• Turn a cold flow into a StateFlow with stateIn and SharingStarted.WhileSubscribed(5000).
• Collect flows safely in Compose with collectAsStateWithLifecycle.
• Write tests for flows with runTest.

Big idea: a Flow is a suspend function that can return many values. It is cold: nothing happens until somebody collects it.

• Coroutines, launch, suspend, dispatchers and structured concurrency (lesson 28). You cannot understand flows without them.
• StateFlow basics in a ViewModel (lessons 24 and 25).
• Lambdas (lesson 12). Operators take lambdas.
• Kotlin collections (lesson 14). map and filter on lists work the same conceptually.
• Dependencies. In the Playground, import kotlinx.coroutines.flow.* works with no setup. In Android Studio, coroutines come with Compose and Lifecycle libraries, plus kotlinx-coroutines-test for tests (lesson 25).

An analogy: a suspend function is ordering one meal. A list is a buffet already on the table. A flow is a conveyor-belt sushi bar: dishes arrive one after another, and you take them as they pass.

Three roles in any flow:

Producer  ->  [operators: map, filter, ...]  ->  Collector
 (emits)         (intermediate, transform)        (terminal, receives)

• The producer emits values (a database, a sensor, a loop).
• Intermediate operators transform the stream without starting it.
• The collector is the terminal step that actually runs everything, and receives each value.

Open the Kotlin Playground and run:

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*
fun countdown(from: Int): Flow<Int> = flow {
 for (i in from downTo 1) {
 delay(500)
 emit(i)
 }
}
fun main() = runBlocking {
 println("Before collecting")
 countdown(3).collect { value ->
 println("Got $value")
 }
 println("Finished")
}

Before collecting
Got 3
Got 2
Got 1
Finished

What happened:

• flow { ... } is a builder. The block is the producer; emit(value) sends a value downstream.
• countdown(3) just creates the flow. Nothing runs yet.
• .collect { } is the terminal operator. It starts the producer, and runs the lambda for each emitted value. It is a suspend function, so the surrounding coroutine waits (without blocking the thread) until the flow completes.
• The delay inside the producer suspends, it does not block, so other coroutines can run.

Cold: each collector gets its own run

fun main() = runBlocking {
 val flow = countdown(2)
 println("First collector")
 flow.collect { println("A got $it") }
 println("Second collector")
 flow.collect { println("B got $it") }
}

First collector
A got 2
A got 1
Second collector
B got 2
B got 1

The second collect starts the producer from scratch. This is what cold means: a cold flow is like a recipe, and every collector cooks it again. It does no work when nobody is collecting, and it stops when the collector stops (cancellation).

Other ways to create flows

val a = flowOf(1, 2, 3) // fixed values
val b = listOf("x", "y").asFlow() // from a collection
val c = (1..5).asFlow() // from a range

Use flowOf and asFlow for tests and previews; use flow { } for real producers.

Operators transform a flow. They are intermediate: they return a new flow and do nothing until collected.

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*
fun main() = runBlocking {
 (1..10).asFlow()
 .filter { it % 2 == 0 } // keep evens: 2, 4, 6, 8, 10
 .map { it * it } // square: 4, 16, 36, 64, 100
 .onEach { println("About to use $it") }
 .take(3) // only the first three: 4, 16, 36
 .collect { println("Result: $it") }
}

About to use 4
Result: 4
About to use 16
Result: 16
About to use 36
Result: 36

Notice the interleaving: values travel through the whole pipeline one at a time, not stage by stage. After take(3) receives its third item, it cancels the upstream flow, so 8 and 10 are never produced.

The operators you will use most

Handling errors with catch

fun main() = runBlocking {
 flow {
 emit(1)
 emit(2)
 throw IllegalStateException("Sensor failed")
 }
 .map { it * 10 }
 .catch { e -> println("Caught: ${e.message}"); emit(-1) }
 .collect { println("Value: $it") }
}

Value: 10
Value: 20
Caught: Sensor failed
Value: -1

catch handles exceptions from upstream (operators above it) only. It does not catch exceptions thrown inside collect { }, because that is downstream. Put a try/catch around the collection for those. And remember lesson 28's rule: do not swallow CancellationException; catch correctly ignores cancellation.

flowOn: choosing the thread

By default a flow runs in the collector's context. To do heavy upstream work on another dispatcher without changing the collector:

flow { emit(readHugeFile()) }
 .flowOn(Dispatchers.IO) // affects everything ABOVE this line
 .collect { show(it) } // runs where it was collected, e.g. Main

Never call withContext inside flow { } to emit from another dispatcher: Kotlin's context-preservation rule throws an exception. Use flowOn.

combine: join several streams

Suppose a search screen has two independent inputs, the query text and a "sort by" choice. Whenever either changes, you recompute the result:

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*
fun main() = runBlocking {
 val query = MutableStateFlow("b")
 val ascending = MutableStateFlow(true)
 val names = listOf("Bola", "Ada", "Banji", "Tunde")
 val job = launch {
 combine(query, ascending) { q, asc ->
 val matches = names.filter { it.contains(q, ignoreCase = true) }
 if (asc) matches.sorted() else matches.sortedDescending()
 }.collect { println(it) }
 }
 delay(100); ascending.value = false // change sorting
 delay(100); query.value = "a" // change search text
 delay(100)
 job.cancel()
}

[Banji, Bola]
[Bola, Banji]
[Bola, Banji, Ada]

Work through it by hand. With the query b, the names containing the letter b (ignoring case) are Bola and Banji, so ascending order gives [Banji, Bola]. Flipping ascending to false recomputes with the same query: [Bola, Banji]. Changing the query to a keeps the descending order and now matches Bola, Banji and Ada (Tunde has no a), giving [Bola, Banji, Ada]. Notice that combine recomputed every time either input changed, and that it emitted once at the start because both flows already had values. Predict output first, run it second: that habit catches mistakes in your own tutorials as well as in code.

debounce and flatMapLatest: the search-as-you-type pattern

A search box that calls the server on every keystroke wastes bandwidth. The standard recipe:

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*
class SearchRepository {
 fun search(query: String): Flow<List<String>> = flow {
 delay(300) // pretend network
 emit(listOf("$query result 1", "$query result 2"))
 }
}
@OptIn(FlowPreview::class)
fun resultsFor(queries: Flow<String>, repo: SearchRepository): Flow<List<String>> =
 queries
 .debounce(400) // wait until the user pauses for 400 ms
 .map { it.trim() }
 .filter { it.length >= 2 } // ignore very short text
 .distinctUntilChanged() // do not repeat the same search
 .flatMapLatest { repo.search(it) } // cancel the old search if text changes

• debounce is marked as a preview API in some versions, hence @OptIn(FlowPreview::class).
• flatMapLatest cancels an in-flight search when a newer query arrives, which also fixes the race condition from lesson 29 (Exercise 2).

A hot flow exists independently of collectors. It produces values whether or not anyone listens, and collectors just tune in.

val state = MutableStateFlow(0) // needs an initial value
state.value = 1 // set
println(state.value) // get
val events = MutableSharedFlow<String>() // no value stored
events.emit("Saved") // suspends until delivered to collectors

Events: a word of caution

Early tutorials used SharedFlow or Channel for one-off events such as "show a snackbar" or "navigate". Current Android guidance recommends modelling events as state where possible: put message: String? in the UI state, show it, and have the UI call onMessageShown() to clear it. That way an event is never lost during rotation, and never delivered twice. See lesson 25 on UI state and the official guide to UI events (https://developer.android.com/topic/architecture/ui-layer/events).

stateIn: from cold to hot, safely

Say a repository exposes a cold Flow<List<Item>>. A ViewModel should expose a StateFlow that the UI collects. stateIn makes the conversion:

import androidx.lifecycle.ViewModel
import androidx.lifecycle.viewModelScope
import kotlinx.coroutines.flow.SharingStarted
import kotlinx.coroutines.flow.StateFlow
import kotlinx.coroutines.flow.map
import kotlinx.coroutines.flow.stateIn
class ItemsViewModel(repository: ItemsRepository) : ViewModel() {
 val uiState: StateFlow<ItemsUiState> = repository.getItems() // Flow<List<Item>>
 .map { items -> ItemsUiState(items) }
 .stateIn(
 scope = viewModelScope,
 started = SharingStarted.WhileSubscribed(5_000),
 initialValue = ItemsUiState()
 )
}

The three parameters:

• scope: where the sharing coroutine lives (viewModelScope, so it is cancelled with the ViewModel).
• started: when to run the upstream flow. WhileSubscribed(5_000) starts it when the first collector arrives and stops it 5 seconds after the last collector leaves. That timeout is the key detail: when the user rotates the phone, the UI stops collecting for a moment, but the upstream (say a database query) is kept alive for the 5 seconds, so it does not restart unnecessarily. When the user leaves the app for longer than 5 seconds, the work stops and saves battery.
• initialValue: what the UI sees before the first emission.

Alternatives: SharingStarted.Eagerly (start at once, never stop) and SharingStarted.Lazily (start at the first collector, never stop). WhileSubscribed(5_000) is the recommended default for UI state.

You have already used:

val uiState by viewModel.uiState.collectAsStateWithLifecycle()

collectAsStateWithLifecycle() collects only while the lifecycle is at least STARTED (lesson 23), and cancels and restarts as the app goes to the background and back. The older collectAsState() keeps collecting in the background.

For a flow that is not a StateFlow, give an initial value:

val time by timeFlow.collectAsStateWithLifecycle(initialValue = "--:--")

To collect a flow for side effects (for example showing a snackbar for each event) use LaunchedEffect with the flow as a key, and repeatOnLifecycle when you must respect the lifecycle (lesson 55 covers effects):

LaunchedEffect(viewModel) {
 viewModel.messages.collect { message ->
 snackbarHostState.showSnackbar(message)
 }
}

In classic Activity code, the safe pattern is:

lifecycleScope.launch {
 repeatOnLifecycle(Lifecycle.State.STARTED) {
 viewModel.uiState.collect { render(it) }
 }
}

A common beginner bug is calling collect directly in lifecycleScope.launch { } with no repeatOnLifecycle, which keeps collecting while the app is in the background.

The second item in this lesson is Testing Flows. You already know runTest from lesson 25. Flows add one idea: the flow produces values over time, so decide what you want to check.

1. Check the latest state

For a StateFlow, usually just read .value after advancing the scheduler:

@Test
fun query_updatesResults() = runTest {
 val vm = SearchViewModel(FakeSearchRepository())
 vm.onQueryChange("ada")
 advanceUntilIdle()
 assertEquals(listOf("Ada"), vm.uiState.value.results)
}

2. Check a whole sequence with toList

For a finite cold flow, collect everything:

import kotlinx.coroutines.flow.flowOf
import kotlinx.coroutines.flow.map
import kotlinx.coroutines.flow.toList
import kotlinx.coroutines.test.runTest
import org.junit.Assert.assertEquals
import org.junit.Test
class OperatorTest {
 @Test
 fun squares_areEmittedInOrder() = runTest {
 val result = flowOf(1, 2, 3).map { it * it }.toList()
 assertEquals(listOf(1, 4, 9), result)
 }
}

toList() never returns for an endless flow such as a StateFlow (it never completes), so the test would hang. For those use first(), take(n).toList(), or collect in a background job.

3. Collect a hot flow in a background job

When you need to see every value of a StateFlow over time, start a collector before triggering the changes:

import kotlinx.coroutines.ExperimentalCoroutinesApi
import kotlinx.coroutines.flow.toList
import kotlinx.coroutines.launch
import kotlinx.coroutines.test.UnconfinedTestDispatcher
import kotlinx.coroutines.test.runTest
@OptIn(ExperimentalCoroutinesApi::class)
@Test
fun counter_emitsEachValue() = runTest {
 val counter = MutableStateFlow(0)
 val seen = mutableListOf<Int>()
 val job = launch(UnconfinedTestDispatcher(testScheduler)) {
 counter.toList(seen) // appends every value into `seen`
 }
 counter.value = 1
 counter.value = 2
 job.cancel()
 assertEquals(listOf(0, 1, 2), seen)
}

UnconfinedTestDispatcher starts the collector eagerly, so it is already listening before the first change. Because StateFlow conflates, rapid changes in a standard dispatcher could skip values, so the unconfined dispatcher is what makes this deterministic. The toList(destination) overload appends into an existing list, so you can read it after cancelling.

4. Testing with a fake that exposes a flow

A fake repository can expose a MutableSharedFlow or MutableStateFlow that your test controls:

class FakeItemsRepository : ItemsRepository {
 val items = MutableStateFlow<List<Item>>(emptyList())
 override fun getItems(): Flow<List<Item>> = items
}

Then the test sets repo.items.value = listOf(Item(1, "Rice")) and asserts that the ViewModel's uiState changes. A third-party library named Turbine is popular for readable flow tests (flow.test { awaitItem() }); it is not part of the official course, so check its current documentation and release notes before adopting it. The built-in tools above are enough for everything in this course.

Testing stateIn with WhileSubscribed

A StateFlow made with WhileSubscribed does nothing until someone collects it. A test that just reads .value sees only the initial value. Start a collector first:

val vm = ItemsViewModel(repo)
val job = launch(UnconfinedTestDispatcher(testScheduler)) { vm.uiState.collect {} }
repo.items.value = listOf(Item(1, "Rice"))
assertEquals(1, vm.uiState.value.items.size)
job.cancel()

Forgetting this is the most common reason a stateIn test returns the initial state forever.

Mistake #1: Forgetting that cold flows do nothing until collected. Symptom: "my onEach never runs". The fix: there must be a terminal operator such as collect, first, toList or launchIn.

Mistake #2: collect in the wrong place. What happens: collect suspends forever for a hot flow, so code after it never runs. The fix: launch the collection in its own coroutine.

Mistake #3: catch placed below the failure. What happens: not caught. The fix: catch only handles upstream exceptions; place it after the code that can throw.

Mistake #4: Using collectAsState() for streams that hit the network or database. What happens: collection continues in the background. The fix: collectAsStateWithLifecycle().

Mistake #5: withContext inside flow { }. Symptom: Flow invariant is violated. The fix: flowOn.

Mistake #6: Expecting StateFlow to emit equal values. What happens: setting the same value twice triggers nothing. The fix: that is by design; use immutable states so changes are real changes.

Mistake #7: Using SharedFlow for one-time UI events and losing them on rotation. The fix: model events as state and clear them after handling.

Mistake #8: toList() on an endless flow in a test. Symptom: the test hangs. The fix: take(n), first() or a background collector.

Mistake #9: Trusting an expected output you did not calculate. What happens: wrong assertions in tests. The fix: hand-calculate, then run it.

Question 1. What does it mean that a Flow is cold?

Question 2. What is the difference between StateFlow and SharedFlow?

Question 3. What does stateIn(viewModelScope, SharingStarted.WhileSubscribed(5000), initial) do?

Question 4. What does flatMapLatest do and why is it useful for search?

Question 5. Why can a test of a stateIn flow see only the initial value?

Go deeper: official resources

• Android Basics with Compose, Unit 6, Pathway 2: Kotlin Flows
• Kotlin coroutines on Android
• Kotlin documentation: Flow
• State production in the UI layer