09 - Kotlin Flow

Recap

In 08 - Kotlin Coroutines a suspend function returned one value: a login token, a list of session types, a file. Much of SportMap's data isn't one value but a stream: a GPS fix every second, a compass heading twenty times a second, the sessions table that changes whenever a session is saved, settings the user can change at any moment. Kotlin's type for an asynchronous stream is Flow. This lecture covers building, transforming and collecting flows, the hot StateFlow and SharedFlow that ViewModels expose, wrapping callback APIs with callbackFlow, and collecting safely from Compose. 10 - App Architecture, ViewModel & Hilt builds directly on it.

By the end of this lecture you should be able to:

• Explain what a cold flow is and when its code runs.
• Build flows and transform them with the common operators, including combine and flatMapLatest.
• Control threading with flowOn and handle errors with catch, retry and onCompletion.
• Explain the difference between StateFlow and SharedFlow, and convert cold flows into hot ones with stateIn/shareIn.
• Explain why SharingStarted.WhileSubscribed(5_000) is the usual choice.
• Wrap a callback-based Android API into a Flow with callbackFlow.
• Collect flows in Compose with collectAsStateWithLifecycle, and turn Compose state into a flow with snapshotFlow.

Demo code

Lecture demos: native-2026-fall

Flow basics

	One value	Many values over time
synchronous	T	Sequence<T> / List<T>
asynchronous	suspend fun (): T	Flow<T>

fun ticker(periodMs: Long): Flow<Int> = flow {
    var i = 0
    while (true) {
        emit(i++)          // hand a value to the collector
        delay(periodMs)    // suspend, don't block
    }
}

viewModelScope.launch {
    ticker(1_000)
        .take(5)                         // intermediate operator - returns a new Flow
        .collect { println("tick $it") } // terminal operator - this is where the code runs
}

A plain Flow is cold:

• Building it does nothing. ticker(1_000) only describes the work.
• The producer code runs per collector, when collect is called. Two collectors mean two independent tickers.
• collect is a suspend function. It returns when the flow completes, and for an endless flow like ticker that's never, until the collecting coroutine is cancelled.
• Cancelling the collector cancels the producer. Structured concurrency from lecture 08 applies all the way up the chain.

Builders:

val a: Flow<Int> = flow { emit(1); emit(2) }
val b: Flow<String> = flowOf("LOC", "WP", "CP")
val c: Flow<Int> = listOf(1, 2, 3).asFlow()
// callbackFlow { } for callback APIs - below
// MutableStateFlow / MutableSharedFlow - hot flows, below

Intermediate operators

Intermediate operators return a new flow and run lazily, value by value, when someone collects.

Operator	What it does	SportMap example
map	transform each value	Location → GpsLocation
filter	drop values	drop fixes with accuracy worse than 30 m
onEach	side effect, value passes through	log every fix
transform	emit zero, one or many values per input	emit a fix plus a "GPS lost" marker
take(n)	first n values, then complete	wait for the first fix
distinctUntilChanged	skip a value equal to the previous one	react only when the GPS interval really changed
debounce(ms)	emit only after the input is quiet for ms	search box: query after typing stops
sample(ms)	latest value once per period	refresh the notification at most once a second
conflate	slow collector skips to the latest value	compass needle doesn't lag behind the sensor
buffer	producer and collector run concurrently	don't slow down the producer
combine	latest of each flow, whenever any of them emits	tracking state + settings → map UI state
zip	pair values one by one	rarely in UIs
flatMapLatest	switch to a new inner flow for each value, cancelling the previous one	new location request when the interval setting changes

val goodFixes: Flow<Location> = rawLocations
    .filter { it.accuracy <= 30f }
    .onEach { Log.d(TAG, "fix ${it.latitude}, ${it.longitude} ±${it.accuracy} m") }

val firstFix: Location = goodFixes.first()     // suspends until one arrives

flatMapLatest: restart when the input changes

The user can change the GPS update interval in settings during a session. The location request has to be replaced:

@OptIn(ExperimentalCoroutinesApi::class)
fun locationsForSettings(settings: Flow<Settings>, locationClient: LocationClient): Flow<Location> =
    settings
        .map { it.gpsIntervalMs }
        .distinctUntilChanged()            // settings emits on ANY change - ignore unrelated ones
        .flatMapLatest { intervalMs ->
            locationClient.locationUpdates(intervalMs)   // the previous request is cancelled
        }

Settings and LocationClient are the SportMap types from lectures 10 and 13. When a new interval arrives, flatMapLatest cancels the collection of the old location flow (its awaitClose removes the old request, see callbackFlow) and starts collecting the new one. The foreground service in lecture 15 runs exactly this chain.

debounce, sample, conflate

@OptIn(FlowPreview::class, ExperimentalCoroutinesApi::class)
val results: Flow<List<String>> = queryFlow
    .debounce(300)                  // wait until the user stops typing
    .distinctUntilChanged()
    .flatMapLatest { query -> searchSessions(query) }

@OptIn(FlowPreview::class)
val notificationUpdates: Flow<TrackingState> = trackingState
    .sample(1_000)                  // at most one notification update per second

val heading: Flow<Float> = rawHeading
    .conflate()                     // 50 readings/s in, the slow UI gets only the latest

debounce and sample are marked @FlowPreview, and flatMapLatest is @ExperimentalCoroutinesApi. They're stable in practice and used everywhere, but you need the @OptIn.

Terminal operators

Operator	Result
collect { }	handles every value, suspends until the flow completes
collectLatest { }	like collect, but cancels the previous block when a new value arrives
first(), firstOrNull(), first { predicate }	the first (matching) value, then cancels the flow
toList()	all values, for finite flows only
single()	exactly one value, otherwise an exception
launchIn(scope)	scope.launch { collect() }, combined with onEach

collect suspends

Two collect calls in one coroutine: the second never runs, because the first never returns for an endless flow.

viewModelScope.launch {
    settings.collect { ... }
    trackingState.collect { ... }     // never reached
}

Launch each collection separately, or better, combine the flows into one (below).

Context: flowOn

A flow runs in the collector's context. Collect on the main thread and the producer runs on the main thread too. flowOn changes the dispatcher of everything upstream of it:

fun readGpxLines(file: File): Flow<String> = flow {
    file.bufferedReader().useLines { lines ->
        lines.forEach { emit(it) }          // blocking reads...
    }
}.flowOn(Dispatchers.IO)                    // ...run on IO; the collector stays on Main

warning

Don't use withContext inside flow { } to emit from another thread. It breaks the "context preservation" rule and throws IllegalStateException: Flow invariant is violated. Use flowOn, or channelFlow/callbackFlow when values really come from other threads.

Errors

val sessionTypes: Flow<List<String>> = flow { emit(api.getSessionTypes()) }
    .retryWhen { cause, attempt ->
        if (cause is IOException && attempt < 3) {
            delay(1_000L * (attempt + 1))       // 1 s, 2 s, 3 s
            true                                // retry
        } else {
            false                               // give up, pass the error downstream
        }
    }
    .catch { e -> emit(emptyList()) }           // fallback value instead of crashing
    .onCompletion { cause -> Log.d(TAG, "session types done, cause=$cause") }

• catch handles exceptions from upstream only. An exception thrown in the collect block isn't caught by it, so use try/catch around collect for that.
• Inside catch you can emit a fallback, rethrow, or emit an error state (UiState(error = ...)).
• retry(3) { it is IOException } is the short form of retryWhen.
• onCompletion runs on normal completion, on error and on cancellation, so it's useful for logging or hiding a progress indicator.

Hot flows: StateFlow and SharedFlow

A hot flow exists independently of collectors. It has its own state, and collectors only tap into it:

	Cold Flow	StateFlow	SharedFlow
runs	per collector, when collected	always, it's a holder	always
current value	no	yes, .value	no (only replay cache)
new collector gets	everything from the start	the current value immediately	the last replay values
equal consecutive values	all delivered	skipped (distinctUntilChanged built in)	all delivered
typical use	repository/database/sensor streams	UI state	broadcasts, events with several listeners

private val _state = MutableStateFlow(TrackingState())
val state: StateFlow<TrackingState> = _state.asStateFlow()

fun stop() {
    _state.update { it.copy(isTracking = false) }      // atomic read-modify-write
}

Lecture 10's TrackingRepository and every ViewModel are built exactly like this: a private MutableStateFlow, exposed read-only. Write with update { } rather than _state.value = _state.value.copy(...), which can lose an update when two coroutines write at the same time.

StateFlow and mutable objects

StateFlow compares values with equals. Mutating a list inside the current state (state.value.track.add(x)) produces no emission, because it's still the same, equal object. Always emit a new immutable value with copy and +.

MutableSharedFlow(replay = 0, extraBufferCapacity = 64) is a broadcast without a current value. emit suspends when the buffer is full, and tryEmit returns false instead. With replay = 0, values emitted while nobody collects are lost. That's why lecture 10 models one-off UI events as state rather than as a SharedFlow.

stateIn and shareIn

A repository typically exposes a cold flow (a database query, a DataStore). A ViewModel turns it into a StateFlow the UI can read:

val settings: StateFlow<Settings> = settingsRepository.settings
    .stateIn(
        scope = viewModelScope,
        started = SharingStarted.WhileSubscribed(5_000),
        initialValue = Settings(),
    )

stateIn starts one collection of the upstream in scope and shares the latest value with every collector. The started policy decides when that upstream collection runs:

Policy	Upstream runs
Eagerly	immediately, until the scope is cancelled
Lazily	from the first collector until the scope is cancelled
WhileSubscribed(5_000)	while at least one collector exists, plus 5 s after the last one leaves

Why WhileSubscribed(5_000):

• Rotation: the activity and its collectors disappear and come back within a second or two. Thanks to the 5 s grace period the upstream (a database query, a location request) keeps running, and the new UI gets the current value without a restart.
• Background: collectAsStateWithLifecycle stops collecting when the app leaves the screen. Five seconds later the upstream stops, so the database stops being queried and the sensor is unregistered, and the battery is saved. When the user returns, it restarts.

shareIn(scope, started, replay = 1) does the same without an initial value. It suits a stream that several collectors share, for example one compass sensor registration feeding both the map rotation and the compass widget.

Define shared flows once

fun settings() = repository.settings.stateIn(...) creates a new shared flow (and upstream collection) on every call. Declare it as a property (val settings = ...) so all collectors share one.

Combining flows into one UI state

A screen usually depends on several sources. combine re-runs its lambda with the latest value of each input whenever any of them emits:

data class TrackingSummaryUiState(
    val isTracking: Boolean = false,
    val pointCount: Int = 0,
    val keepCentered: Boolean = true,
    val compassVisible: Boolean = false,
)

class TrackingSummaryViewModel(
    trackingState: Flow<TrackingState>,     // lecture 10: TrackingRepository.state
    settings: Flow<Settings>,               // lecture 10: SettingsRepository.settings
) : ViewModel() {

    val uiState: StateFlow<TrackingSummaryUiState> = combine(trackingState, settings) { tracking, settings ->
        TrackingSummaryUiState(
            isTracking = tracking.isTracking,
            pointCount = tracking.track.size,
            keepCentered = settings.keepCentered,
            compassVisible = settings.compassVisible,
        )
    }.stateIn(viewModelScope, SharingStarted.WhileSubscribed(5_000), TrackingSummaryUiState())
}

The flows are passed in directly here to keep the example small. The real MapViewModel of lectures 15–18 combines the same flows (plus more), taken from injected repositories (lecture 10), but the combine + stateIn part looks exactly like this. combine accepts up to five flows with typed parameters, plus a vararg version beyond that.

callbackFlow: wrapping callback APIs

Many Android APIs push values to a callback or listener: location updates, sensors, broadcast receivers. callbackFlow turns such an API into a cold Flow. The callback is registered when collection starts and unregistered when it stops.

Battery level arrives as the system broadcast ACTION_BATTERY_CHANGED, with a BroadcastReceiver as in lecture 02:

fun Context.batteryLevel(): Flow<Int> = callbackFlow {
    val receiver = object : BroadcastReceiver() {
        override fun onReceive(context: Context, intent: Intent) {
            val level = intent.getIntExtra(BatteryManager.EXTRA_LEVEL, -1)
            val scale = intent.getIntExtra(BatteryManager.EXTRA_SCALE, -1)
            if (level >= 0 && scale > 0) {
                trySend(level * 100 / scale)          // callbacks aren't suspend - use trySend
            }
        }
    }
    // ACTION_BATTERY_CHANGED is sticky: registering immediately delivers the current value
    ContextCompat.registerReceiver(
        this@batteryLevel,
        receiver,
        IntentFilter(Intent.ACTION_BATTERY_CHANGED),
        ContextCompat.RECEIVER_NOT_EXPORTED,
    )
    awaitClose { unregisterReceiver(receiver) }       // runs when the collector goes away
}.distinctUntilChanged()

The pattern always has the same three parts:

1. create the callback, which calls trySend(value) (or close(exception) on a fatal error);
2. register it;
3. awaitClose { unregister }. It keeps the flow open, and it's mandatory: callbackFlow throws without it, and forgetting to unregister leaks the listener and drains the battery.

The values sit in a buffer between the callback and the collector (64 by default). For fast producers like sensors, where only the newest value matters, add .conflate() (or .buffer(Channel.CONFLATED)) so a slow collector never works through stale readings. trySend returns a failed result instead of suspending when the buffer is full or the flow is closed.

Lecture 13 wraps fused location updates (LocationClient.locationUpdates()) and lecture 17 wraps the compass sensor (CompassSensor.azimuthDeg) the same way.

Collecting in Compose

@Composable
fun TrackingSummaryRoute(viewModel: TrackingSummaryViewModel) {
    val state by viewModel.uiState.collectAsStateWithLifecycle()
    TrackingSummary(state)
}

collectAsStateWithLifecycle() (from lifecycle-runtime-compose, already in the catalog since lecture 05) turns a flow into Compose State. It collects only while the lifecycle is at least STARTED, and stops when the app goes to the background.

collectAsState() from Compose itself collects as long as the composable is in the composition, including while the app is in the background. With a location or sensor flow behind it, that means GPS and sensors keep running for an invisible screen. Prefer collectAsStateWithLifecycle on Android. (Background tracking is the foreground service's job, not the UI's.)

For a cold flow, pass an initial value, and remember the flow, otherwise every recomposition creates a new flow and restarts collection:

@Composable
fun BatteryIndicator(modifier: Modifier = Modifier) {
    val context = LocalContext.current
    val batteryFlow = remember(context) { context.batteryLevel() }
    val level by batteryFlow.collectAsStateWithLifecycle(initialValue = null)
    Text(text = level?.let { "Battery $it %" } ?: "Battery --", modifier = modifier)
}

Outside Compose (an activity, a service) use lifecycleScope plus repeatOnLifecycle(Lifecycle.State.STARTED) from lecture 08.

snapshotFlow: Compose state → Flow

The other direction: snapshotFlow { } turns reads of Compose State into a flow that emits whenever that state changes. Then you can apply flow operators to UI state:

@Composable
fun SessionList(onScrolledChange: (Boolean) -> Unit) {
    val listState = rememberLazyListState()
    LaunchedEffect(listState) {
        snapshotFlow { listState.firstVisibleItemIndex }
            .map { it > 0 }
            .distinctUntilChanged()
            .collect { scrolled -> onScrolledChange(scrolled) }
    }
    LazyColumn(state = listState) { /* ... */ }
}

LiveData

Older code and tutorials (and the 2025 version of these lectures) use LiveData. It's a lifecycle-aware observable holder from 2017: main-thread only, with observe(lifecycleOwner) { } and barely any operators. StateFlow + collectAsStateWithLifecycle replaces it in new code. When you meet it, flow.asLiveData() and liveData.asFlow() (lifecycle-livedata-ktx) convert between the two, and observeAsState() (compose-runtime-livedata) collects it in Compose.

Testing flows

Finite flows are easy to test with the terminal operators inside runTest, and virtual time (lecture 08) applies:

@OptIn(ExperimentalCoroutinesApi::class)
class FlowTest {
    @Test
    fun tickerEmitsOncePerSecond() = runTest {
        val values = ticker(1_000).take(3).toList()
        assertEquals(listOf(0, 1, 2), values)
        assertEquals(2_000, currentTime)              // virtual time, the test takes milliseconds
    }

    @Test
    fun poorFixesAreDropped() = runTest {
        val accuracies = flowOf(5f, 50f, 12f).filter { it <= 30f }.toList()
        assertEquals(listOf(5f, 12f), accuracies)
    }
}

To test a StateFlow built with WhileSubscribed, start a collector in backgroundScope first, as in the ViewModel test of lecture 10. For longer sequences the Turbine library helps: flow.test { assertEquals(0, awaitItem()) }.

In the final project

• Every changing data source is a Flow: LocationClient.locationUpdates() (lecture 13) and CompassSensor.azimuthDeg (17) are callbackFlows, Room DAO queries return Flow (18), and SettingsRepository.settings is a DataStore flow (10).
• The service runs settings.map { it.gpsIntervalMs }.distinctUntilChanged().flatMapLatest { locationClient.locationUpdates(it) }, so the "GPS update frequency" setting applies in a running session.
• The notification is refreshed from trackingRepository.state.sample(1_000). The spec asks for an update after every position, and with a 1 s GPS interval that's once per update, without hammering the notification system with faster bursts.
• MapViewModel (lectures 15–18) builds its UI state the way TrackingSummaryViewModel does here, with combine(trackingState, settings, ...) and stateIn(WhileSubscribed(5_000)). Rotation keeps the data, and the compass sensor stops 5 s after the app goes to the background.
• The map screen uses collectAsStateWithLifecycle, so an invisible screen consumes no sensor or GPS updates. Recording continues in the service.

Self preparation QA

1. What does "a flow is cold" mean? — Building it does nothing. The producer code runs separately for each collector when collect is called, and stops when the collector is cancelled.
2. What is the difference between intermediate and terminal operators? — Intermediate operators (map, filter, combine...) return a new flow and run lazily. Terminal operators (collect, first, toList) are suspend functions that start the flow and consume values.
3. Why does flatMapLatest fit the "GPS interval changed" case? — For every new interval it cancels the collection of the previous location flow (unregistering its request) and starts a new one, so exactly one request with the latest setting is active.
4. How do debounce, sample and conflate differ? — debounce emits only after the input has been quiet for a while, sample emits the latest value once per period, and conflate lets a slow collector skip intermediate values and get the newest.
5. Which part of a flow does flowOn affect, and why not use withContext inside flow { }? — flowOn changes the dispatcher of everything upstream of it. Emitting from another context inside flow { } violates context preservation and throws.
6. What is the difference between StateFlow and SharedFlow? — StateFlow always has a current value, gives it to new collectors and skips equal consecutive values, which makes it ideal for UI state. SharedFlow has no value, replays a configurable number of past values and delivers all of them, for broadcasts.
7. Why is SharingStarted.WhileSubscribed(5_000) the usual choice in ViewModels? — The 5 s grace period keeps the upstream alive through a rotation, but stops it shortly after the app goes to the background, saving battery.
8. What are the three parts of a callbackFlow, and what happens without awaitClose? — A callback that calls trySend, registering it, and awaitClose { unregister }. Without awaitClose, callbackFlow throws, and without the unregister the listener leaks.
9. Why prefer collectAsStateWithLifecycle over collectAsState on Android? — It stops collecting when the app isn't at least STARTED, so sensors, GPS and queries behind the flow don't run for an invisible screen.
10. Why must you remember a cold flow that you create in a composable before collecting it? — Otherwise each recomposition creates a new flow object, and collection restarts every time.