Flows

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?

// 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<Criterion>) {
    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.

// 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.

// Reading.kt — data class
data class Reading(val id: Int)

// 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.

// 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:

// 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.

// Main.kt — take cuts the flow short
import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*

fun main() = runBlocking {
    val log = mutableListOf<String>()
    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.

// 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<Int>()
    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.

// 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.

// 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.