--- title: Flows source: 'https://academia.sh/en/courses/kotlin/flows' course: 'The Kotlin Language' language: en updated: '2026-10-21T05:02:39+00:00' license: 'CC BY-SA 4.0' objectives: - 'Measures with a counter that a cold flow reruns on every collecting' - 'Transforms and collects the values of a flow with map, filter and toList' - 'Shows that distinctUntilChanged and MutableStateFlow drop consecutive equal values by content' - 'Measures that a data class and a plain class change how many elements are dropped, and that two flows are not counted as equal' --- # Flows Measures that a cold flow reruns on every collecting, that consecutive equal values are dropped by content, and that two flows are not counted as equal. The previous lesson dealt with jobs that produce a single value. Most jobs produce not one value but a sequence of values that arrive over time: readings from a sensor, the pages of a list, each step of a counter. A sequence cannot wait for such values; while producing an element it blocks the caller. A **flow (`Flow`)** builds the same idea with coroutines: it produces values one at a time, and the producer and the collector can be suspended in between. A flow has two faces. A **cold flow** is built with `flow { }` or `flowOf` and does nothing on its own; it waits until someone collects it. For every collector the producer block runs from the start. A **hot flow** carries a state independent of the collector; in this lesson only `MutableStateFlow` appears, a hot flow that holds a last value. `collect` is used for collecting; to reduce all the values of a flow to a list, `toList()` is enough. Both are suspending functions and are called inside a coroutine; this lesson again sets up a `runBlocking` scope, and next to `import kotlinx.coroutines.*` the flows need `import kotlinx.coroutines.flow.*`. Timing is not measured, only values and counts are read. The measuring question is as in the earlier lessons: **what makes these two values equal?** ```kotlin // Criterion.kt — classifies the criterion that makes two values equal class Criterion(val label: String, first: Any?, twin: Any?, other: Any?) { val sameObject = first === twin val equalTwin = first == twin val equalOther = first == other val name: String = when { sameObject -> "singleton" !equalTwin -> "identity" !equalOther -> "content" else -> "partial-content" } fun report() { println("$label: same-object=$sameObject, twin=$equalTwin, other=$equalOther -> $name") } } fun summarize(checks: List) { for (name in listOf("singleton", "identity", "content", "partial-content")) { println("$name: ${checks.count { it.name == name }}") } } ``` ## Cold Flow, Transformation and Hot Flow The block below first builds a flow with `flow { }`; each time the producer block runs it raises a counter. The flow is collected twice and the counter is read. Then transformation with `map` and `filter` and collecting with `toList()` are shown. Finally a `MutableStateFlow` is built, its value is changed and read with `value`. ```kotlin // Main.kt — cold flow and hot flow import kotlinx.coroutines.* import kotlinx.coroutines.flow.* fun main() = runBlocking { var runs = 0 val numbers = flow { runs++ emit(1) emit(2) emit(3) } println(runs) println(numbers.toList()) println(numbers.toList()) println(runs) println(flowOf(1, 2, 3, 4).map { it * 10 }.filter { it > 10 }.toList()) val state = MutableStateFlow(5) state.value = 6 println(state.value) } ``` ``` 0 [1, 2, 3] [1, 2, 3] 2 [20, 30, 40] 6 ``` The first line shows that when the flow was built the producer block did not run at all: `runs` is zero. The two `toList()` calls each give `[1, 2, 3]`, and the fourth line says that the producer block ran **twice**. This is the definition of a cold flow: every collector runs the flow from the start, and the flow remembers nothing from the previous collecting. On the fifth line `map` multiplies each value by ten, `filter` drops the ten, and `[20, 30, 40]` remains; as with the sequence, the intermediate operations only describe the flow and run at collecting time. On the last line `MutableStateFlow` holds a value and the last assigned value is read with `value`: `6`. **CC27.** These measurements are for single-collector flows and a single-threaded `runBlocking` scope. Several collectors collecting at the same time, changing the dispatcher with `flowOn`, and cancelling a flow from outside with `cancel` were not measured; cutting a flow short with `take` is measured in a later section. ## The Criterion: Content `MutableStateFlow` and `distinctUntilChanged` ask the same question: is the newly arrived value equal to the previous one? If it is equal, they skip it. This question is the measuring question of this course, so in a flow the criterion decides equality. Below, two helper classes are defined first: `Reading` is a data class, and `Raw` is a plain class with the same field. ```kotlin // Reading.kt — data class data class Reading(val id: Int) ``` ```kotlin // Raw.kt — plain class class Raw(val id: Int) ``` The block below applies `distinctUntilChanged()` to a `flowOf(A, A, B, B, A)` flow with `Reading` elements and counts how many values remain. Right beside it a `Criterion` measurement is made for `Reading`. ```kotlin // Main.kt — consecutive equality import kotlinx.coroutines.* import kotlinx.coroutines.flow.* fun main() = runBlocking { val a = Reading(1) val b = Reading(2) val kept = flowOf(a, Reading(1), b, Reading(2), Reading(1)).distinctUntilChanged().toList() println(kept.size) Criterion("data-class", Reading(1), Reading(1), Reading(2)).report() } ``` ``` 3 data-class: same-object=false, twin=true, other=false -> content ``` Of the five values, three remained: `A, B, A`. The second `Reading(1)` and the second `Reading(2)`, which arrived right after an equal value, were dropped, because `distinctUntilChanged` compares with the previous value using `==` and a data class defines `equals` by its fields. `Reading(1)`, a separate object, was counted as equal to the previous one. The second line says the same result with `Criterion`: the twin is equal, the other is not, and the criterion is **content**. This is the criterion of this lesson: **content**. **CC28.** These measurements are for a data class with an `Int` field and a `flowOf` with five elements. `distinctUntilChangedBy`, which selects a key, the `distinctUntilChanged { a, b -> ... }` form, which takes a comparator, and how many values can be dropped in long flows were not measured. ## The Bounding Measurement A block that measures the same flow with plain-class elements, assigning an equal value to a `MutableStateFlow`, and two separate flows themselves: ```kotlin // Main.kt — plain class, state and flow import kotlinx.coroutines.* import kotlinx.coroutines.flow.* fun main() = runBlocking { val raws = flowOf(Raw(1), Raw(1), Raw(2), Raw(2), Raw(1)).distinctUntilChanged().toList() println(raws.size) val original = Reading(1) val state = MutableStateFlow(original) state.value = Reading(1) println(state.value === original) val checks = listOf( Criterion("flow", flowOf(1, 2), flowOf(1, 2), flowOf(1, 3)), Criterion("data-class", Reading(1), Reading(1), Reading(2)), Criterion("plain-class", Raw(1), Raw(1), Raw(2)), ) for (check in checks) check.report() summarize(checks) } ``` ``` 5 true flow: same-object=false, twin=false, other=false -> identity data-class: same-object=false, twin=true, other=false -> content plain-class: same-object=false, twin=false, other=false -> identity singleton: 0 identity: 2 content: 1 partial-content: 0 ``` The first line shows that the same `distinctUntilChanged` call drops no value from the plain-class elements: all five of the five elements remained. Even though `Raw(1)` and `Raw(1)` carry the same field as content, `equals` falls back to identity and consecutive values are never counted as equal. On the second line a separate but equal object, `Reading(1)`, was assigned to the `MutableStateFlow`, and `value` still points to the first object (`true`): the flow did not count the equal value as a new value and left the old one in place. The hot flow, like the cold flow, measures equality with `==`. The remaining lines put three measurements side by side. Two separate `flowOf(1, 2)` flows give the **identity** criterion: a flow itself does not define equality by content, just like the sequence and the `Deferred`. The data class comes out `content`, the plain class `identity`. The bounding measurement is this: **the values inside a flow follow the content criterion only if the type of the element carries it.** The recipe that builds the flow is compared by identity, while the value passing through the flow is compared by the equality of its type. The summary table says the same: two `identity` (the flow and the plain class), one `content` (the data class). **CC29.** These measurements are for `MutableStateFlow` and cover assigning an equal value with `value`. The values a `StateFlow` delivers to a collector and `SharedFlow` were not measured; assignments with `update` and `compareAndSet` are measured in a later section. **CC30.** In this lesson no timing, thread name or dispatcher order was printed. The values of the flows were collected with `toList()`, and the order in the output is the producer's order. ## Cutting a Flow Short The producer of a flow can produce many values, but the collector may not want all of them. `take(n)` takes the first `n` values from the flow and then cancels the flow; the producer block does not continue past that point. It is the same idea as `take` on a sequence, but here what is cut is a coroutine. The block below builds a flow ready to produce five values. The producer leaves a record before each value, so after `take(2)` how far the producer got can be read from the list. In addition, the start and the end of the flow are recorded with `onStart` and `onCompletion`. ```kotlin // Main.kt — take cuts the flow short import kotlinx.coroutines.* import kotlinx.coroutines.flow.* fun main() = runBlocking { val log = mutableListOf() val source = flow { for (i in 1..5) { log.add("produce$i") emit(i) } } val firstTwo = source .onStart { log.add("start") } .onCompletion { log.add("done") } .take(2) .toList() println(firstTwo) println(log) } ``` ``` [1, 2] [start, produce1, produce2, done] ``` The collected list became `[1, 2]`, so it holds only two values. The record list opens with `start`, continues with `produce1` and `produce2`, and closes with `done`; there is no `produce3` record at all. The records show that the producer never reached for the third value: `take(2)` cut the flow after taking the second value. The `onStart` and `onCompletion` records are also in the list, and the completion record drops on a cut flow as well. This order is guaranteed, because when `toList()` returns the flow is complete; the records are read after this point. The practical consequence is this: if the first few values of a flow are enough, using `take` or `first()` does not run the rest of the producer block. Where the producer does work that has a side effect, this difference is the difference between work that runs and work that does not. **CC31.** These measurements are for `take`, `onStart` and `onCompletion`, with a single collector. The `try/finally` cleanup in the producer block while the flow is being cut, and cutting with `takeWhile`, were not measured. ## Terminal Operations and the Empty Flow Turning a flow into a list is not the only way. `fold` and `reduce` reduce all the values to a single value, `count` gives the number of values, and `first()` returns the first value. The empty flow is the boundary case: `first()` throws an exception, while `firstOrNull()` gives `null`. The block below calls these operations on a flow with four values; then it tries two of them on an empty flow. `fold` asks for an initial value; `reduce` counts the first value as the initial value, and cannot work on an empty flow. ```kotlin // Main.kt — terminal operations and the empty flow import kotlinx.coroutines.* import kotlinx.coroutines.flow.* fun main() = runBlocking { val numbers = flowOf(1, 2, 3, 4) println(numbers.fold(100) { sum, n -> sum + n }) println(numbers.reduce { sum, n -> sum + n }) println(numbers.count { it % 2 == 0 }) println(numbers.first()) val empty = emptyFlow() try { empty.first() } catch (e: NoSuchElementException) { println(e::class.simpleName) } println(empty.firstOrNull()) } ``` ``` 110 10 2 1 NoSuchElementException null ``` `fold` added the four values to its initial value of `100`. `reduce` started from the first value without an initial value and gave only the sum of the four values. `count` counted only the even ones. `first()` is the first value of the flow. On the empty flow `first()` threw `NoSuchElementException`, while `firstOrNull()` returned `null`. This is the flow counterpart of the `first` and `firstOrNull` pair on collections: the first is chosen if the error should be visible, the second if emptiness is acceptable. **CC32.** These measurements are for finite flows of `Int` values. Calling `fold` and `count` on a flow that never ends, and the class name of the exception that `reduce` throws on an empty flow, were not measured. ## Transforming and Combining `map` produces one value from each value. To produce more than one value or no value from a value, `transform` is used: the block takes the incoming value and calls `emit` as many times as it likes. To read two flows side by side, `zip` takes one value at a time from the two flows and combines them; when the shorter flow ends, the result ends too. The block below measures both. The second flow carries two values instead of three. ```kotlin // Main.kt — transform and zip import kotlinx.coroutines.* import kotlinx.coroutines.flow.* fun main() = runBlocking { val doubled = flowOf(1, 2, 3).transform { n -> emit(n) if (n % 2 == 1) emit(n * 10) }.toList() println(doubled) val pairs = flowOf(1, 2, 3).zip(flowOf("a", "b")) { n, s -> "$n$s" }.toList() println(pairs) } ``` ``` [1, 10, 2, 3, 30] [1a, 2b] ``` `transform` produced two values for the odd numbers and one value for the even number. The result list carries the input values and the derived values one after another. `zip` paired one value from each of the two flows and produced `"1a"` and `"2b"`; because the third value of the first flow was left without a partner, it produced no result. The result of these two operations depends not on timing but only on the input values and their order, so they can be measured safely. The rule is clear: `map` does a one-to-one, `transform` a one-to-many or one-to-zero transformation; `zip` stops at the length of the shorter one. **CC33.** These measurements are for `transform` and `zip`, with two finite flows. `combine`, which joins the latest values of two flows as they change over time, and `flatMapConcat` were not measured; their result can depend on timing. ## Updating the State Assigning with `value` is enough to put a value into a `MutableStateFlow`, but the gap between reading and assigning is not safe. The `update { }` block computes the new value from the old one and makes the assignment in a single step. `compareAndSet(expected, new)` assigns only if the current value is equal to the expected one, and returns whether it assigned. The comparison is with `==`, so these two operations also depend on the equality rule of the earlier sections. The block below updates a number and checks it. ```kotlin // Main.kt — update and compareAndSet import kotlinx.coroutines.* import kotlinx.coroutines.flow.* fun main() = runBlocking { val state = MutableStateFlow(0) state.update { it + 1 } println(state.value) println(state.compareAndSet(1, 5)) println(state.value) println(state.compareAndSet(1, 9)) println(state.value) println(state.first()) } ``` ``` 1 true 5 false 5 5 ``` `update` raised the value from `0` to `1`. The first `compareAndSet(1, 5)` assigned, because the current value was `1`, and returned `true`. The second `compareAndSet(1, 9)` did not assign, because the value was now `5`, and returned `false`; the value stayed `5`. `first()` gave the current value of the hot flow: a state flow, when collected, always starts with the current value. This separates `update` and `compareAndSet` from a plain `value` assignment: both read the current value and decide, and they do it in a single operation. This is the difference from a cold flow: a hot flow has a value in hand, and gives it first to a new collector. **CC34.** These measurements are for a `MutableStateFlow` of `Int` values and a single coroutine. Several coroutines calling `update` at the same time, and whether a `StateFlow` collector sees the intermediate values, were not measured. ## Summary - A cold flow runs the producer block from the start on every collecting; nothing runs when it is built. `map` and `filter` transform the flow, `toList()` collects it. - `MutableStateFlow` is a hot flow that holds the last value; when an equal value is assigned, the old object stays in place. - `distinctUntilChanged` drops consecutive equal values with `==`. The criterion of this lesson: **content**. With the data class 3 of 5 values remain, with the plain class all 5. - The bounding measurement: the flow itself is compared by identity, and the values inside it by the equality of their type. ## Next Step Until now every lesson asked whether a value is equal. The last lesson asks whether an error is equal instead of a value: are two exceptions that carry the same message equal? The Exception Handling lesson measures this with try, catch and finally.