Coroutines - Asynchronous Flow - 《Kotlin Language Guide v1.4.21》

Asynchronous Flow

Table of contents

Asynchronous Flow
Representing multiple values
Flows are cold
Flow cancellation basics
Flow builders
Intermediate flow operators
- Transform operator
- Size-limiting operators
Terminal flow operators
Flows are sequential
Flow context
- Wrong emission withContext
- flowOn operator
Buffering
- Conflation
- Processing the latest value
Composing multiple flows
- Zip
- Combine
Flattening flows
Flow exceptions
- Collector try and catch
- Everything is caught
Exception transparency
- Transparent catch
- Catching declaratively
Flow completion
Imperative versus declarative
Launching flow
Flow cancellation checks
- Making busy flow cancellable
Flow and Reactive Streams

Asynchronous Flow

A suspending function asynchronously returns a single value, but how can we return multiple asynchronously computed values? This is where Kotlin Flows come in.

Representing multiple values

Multiple values can be represented in Kotlin using collections. For example, we can have a simple function that returns a List of three numbers and then print them all using forEach:

1. fun simple(): List<Int> = listOf(1, 2, 3)

3. fun main() {
4. simple().forEach { value -> println(value) }
5. }

This code outputs:

1. 1
2. 2
3. 3

Sequences

If we are computing the numbers with some CPU-consuming blocking code (each computation taking 100ms), then we can represent the numbers using a Sequence:

1. fun simple(): Sequence<Int> = sequence { // sequence builder
2. for (i in 1..3) {
3. Thread.sleep(100) // pretend we are computing it
4. yield(i) // yield next value
5. }
6. }

8. fun main() {
9. simple().forEach { value -> println(value) }
10. }

This code outputs the same numbers, but it waits 100ms before printing each one.

Suspending functions

However, this computation blocks the main thread that is running the code. When these values are computed by asynchronous code we can mark the simple function with a suspend modifier, so that it can perform its work without blocking and return the result as a list:

1. import kotlinx.coroutines.*

3. //sampleStart
4. suspend fun simple(): List<Int> {
5. delay(1000) // pretend we are doing something asynchronous here
6. return listOf(1, 2, 3)
7. }

9. fun main() = runBlocking<Unit> {
10. simple().forEach { value -> println(value) }
11. }
12. //sampleEnd

This code prints the numbers after waiting for a second.

Flows

Using the List<Int> result type, means we can only return all the values at once. To represent the stream of values that are being asynchronously computed, we can use a Flow<Int> type just like we would use the Sequence<Int> type for synchronously computed values:

1. import kotlinx.coroutines.*
2. import kotlinx.coroutines.flow.*

4. //sampleStart
5. fun simple(): Flow<Int> = flow { // flow builder
6. for (i in 1..3) {
7. delay(100) // pretend we are doing something useful here
8. emit(i) // emit next value
9. }
10. }

12. fun main() = runBlocking<Unit> {
13. // Launch a concurrent coroutine to check if the main thread is blocked
14. launch {
15. for (k in 1..3) {
16. println("I'm not blocked $k")
17. delay(100)
18. }
19. }
20. // Collect the flow
21. simple().collect { value -> println(value) }
22. }
23. //sampleEnd

This code waits 100ms before printing each number without blocking the main thread. This is verified by printing “I’m not blocked” every 100ms from a separate coroutine that is running in the main thread:

1. I'm not blocked 1
2. 1
3. I'm not blocked 2
4. 2
5. I'm not blocked 3
6. 3

Notice the following differences in the code with the Flow from the earlier examples:

A builder function for Flow type is called flow.
Code inside the flow { ... } builder block can suspend.
The simple function is no longer marked with suspend modifier.
Values are emitted from the flow using emit function.
Values are collected from the flow using collect function.

Flows are cold

Flows are cold streams similar to sequences — the code inside a flow builder does not run until the flow is collected. This becomes clear in the following example:

1. import kotlinx.coroutines.*
2. import kotlinx.coroutines.flow.*

4. //sampleStart
5. fun simple(): Flow<Int> = flow {
6. println("Flow started")
7. for (i in 1..3) {
8. delay(100)
9. emit(i)
10. }
11. }

13. fun main() = runBlocking<Unit> {
14. println("Calling simple function...")
15. val flow = simple()
16. println("Calling collect...")
17. flow.collect { value -> println(value) }
18. println("Calling collect again...")
19. flow.collect { value -> println(value) }
20. }
21. //sampleEnd

Which prints:

1. Calling simple function...
2. Calling collect...
3. Flow started
4. 1
5. 2
6. 3
7. Calling collect again...
8. Flow started
9. 1
10. 2
11. 3

This is a key reason the simple function (which returns a flow) is not marked with suspend modifier. By itself, simple() call returns quickly and does not wait for anything. The flow starts every time it is collected, that is why we see “Flow started” when we call collect again.

Flow cancellation basics

Flow adheres to the general cooperative cancellation of coroutines. As usual, flow collection can be cancelled when the flow is suspended in a cancellable suspending function (like delay). The following example shows how the flow gets cancelled on a timeout when running in a withTimeoutOrNull block and stops executing its code:

1. import kotlinx.coroutines.*
2. import kotlinx.coroutines.flow.*

4. //sampleStart
5. fun simple(): Flow<Int> = flow {
6. for (i in 1..3) {
7. delay(100)
8. println("Emitting $i")
9. emit(i)
10. }
11. }

13. fun main() = runBlocking<Unit> {
14. withTimeoutOrNull(250) { // Timeout after 250ms
15. simple().collect { value -> println(value) }
16. }
17. println("Done")
18. }
19. //sampleEnd

Notice how only two numbers get emitted by the flow in the simple function, producing the following output:

1. Emitting 1
2. 1
3. Emitting 2
4. 2
5. Done

See Flow cancellation checks section for more details.

Flow builders

The flow { ... } builder from the previous examples is the most basic one. There are other builders for easier declaration of flows:

flowOf builder that defines a flow emitting a fixed set of values.
Various collections and sequences can be converted to flows using .asFlow() extension functions.

So, the example that prints the numbers from 1 to 3 from a flow can be written as:

1. import kotlinx.coroutines.*
2. import kotlinx.coroutines.flow.*

4. fun main() = runBlocking<Unit> {
5. //sampleStart
6. // Convert an integer range to a flow
7. (1..3).asFlow().collect { value -> println(value) }
8. //sampleEnd
9. }

Intermediate flow operators

Flows can be transformed with operators, just as you would with collections and sequences. Intermediate operators are applied to an upstream flow and return a downstream flow. These operators are cold, just like flows are. A call to such an operator is not a suspending function itself. It works quickly, returning the definition of a new transformed flow.

The basic operators have familiar names like map and filter. The important difference to sequences is that blocks of code inside these operators can call suspending functions.

For example, a flow of incoming requests can be mapped to the results with the map operator, even when performing a request is a long-running operation that is implemented by a suspending function:

1. import kotlinx.coroutines.*
2. import kotlinx.coroutines.flow.*

4. //sampleStart
5. suspend fun performRequest(request: Int): String {
6. delay(1000) // imitate long-running asynchronous work
7. return "response $request"
8. }

10. fun main() = runBlocking<Unit> {
11. (1..3).asFlow() // a flow of requests
12. .map { request -> performRequest(request) }
13. .collect { response -> println(response) }
14. }
15. //sampleEnd

It produces the following three lines, each line appearing after each second:

1. response 1
2. response 2
3. response 3

Transform operator

Among the flow transformation operators, the most general one is called transform. It can be used to imitate simple transformations like map and filter, as well as implement more complex transformations. Using the transform operator, we can emit arbitrary values an arbitrary number of times.

For example, using transform we can emit a string before performing a long-running asynchronous request and follow it with a response:

1. import kotlinx.coroutines.*
2. import kotlinx.coroutines.flow.*

4. suspend fun performRequest(request: Int): String {
5. delay(1000) // imitate long-running asynchronous work
6. return "response $request"
7. }

9. fun main() = runBlocking<Unit> {
10. //sampleStart
11. (1..3).asFlow() // a flow of requests
12. .transform { request ->
13. emit("Making request $request")
14. emit(performRequest(request))
15. }
16. .collect { response -> println(response) }
17. //sampleEnd
18. }

The output of this code is:

1. Making request 1
2. response 1
3. Making request 2
4. response 2
5. Making request 3
6. response 3

Size-limiting operators

Size-limiting intermediate operators like take cancel the execution of the flow when the corresponding limit is reached. Cancellation in coroutines is always performed by throwing an exception, so that all the resource-management functions (like try { ... } finally { ... } blocks) operate normally in case of cancellation:

1. import kotlinx.coroutines.*
2. import kotlinx.coroutines.flow.*

4. //sampleStart
5. fun numbers(): Flow<Int> = flow {
6. try {
7. emit(1)
8. emit(2)
9. println("This line will not execute")
10. emit(3)
11. } finally {
12. println("Finally in numbers")
13. }
14. }

16. fun main() = runBlocking<Unit> {
17. numbers()
18. .take(2) // take only the first two
19. .collect { value -> println(value) }
20. }
21. //sampleEnd

The output of this code clearly shows that the execution of the flow { ... } body in the numbers() function stopped after emitting the second number:

1. 1
2. 2
3. Finally in numbers

Terminal flow operators

Terminal operators on flows are suspending functions that start a collection of the flow. The collect operator is the most basic one, but there are other terminal operators, which can make it easier:

Conversion to various collections like toList and toSet.
Operators to get the first value and to ensure that a flow emits a single value.
Reducing a flow to a value with reduce and fold.

For example:

1. import kotlinx.coroutines.*
2. import kotlinx.coroutines.flow.*

4. fun main() = runBlocking<Unit> {
5. //sampleStart
6. val sum = (1..5).asFlow()
7. .map { it * it } // squares of numbers from 1 to 5
8. .reduce { a, b -> a + b } // sum them (terminal operator)
9. println(sum)
10. //sampleEnd
11. }

Prints a single number:

1. 55

Flows are sequential

Each individual collection of a flow is performed sequentially unless special operators that operate on multiple flows are used. The collection works directly in the coroutine that calls a terminal operator. No new coroutines are launched by default. Each emitted value is processed by all the intermediate operators from upstream to downstream and is then delivered to the terminal operator after.

See the following example that filters the even integers and maps them to strings:

1. import kotlinx.coroutines.*
2. import kotlinx.coroutines.flow.*

4. fun main() = runBlocking<Unit> {
5. //sampleStart
6. (1..5).asFlow()
7. .filter {
8. println("Filter $it")
9. it % 2 == 0
10. }
11. .map {
12. println("Map $it")
13. "string $it"
14. }.collect {
15. println("Collect $it")
16. }
17. //sampleEnd
18. }

Producing:

1. Filter 1
2. Filter 2
3. Map 2
4. Collect string 2
5. Filter 3
6. Filter 4
7. Map 4
8. Collect string 4
9. Filter 5

Flow context

Collection of a flow always happens in the context of the calling coroutine. For example, if there is a simple flow, then the following code runs in the context specified by the author of this code, regardless of the implementation details of the simple flow:

1. withContext(context) {
2. simple().collect { value ->
3. println(value) // run in the specified context
4. }
5. }

This property of a flow is called context preservation.

So, by default, code in the flow { ... } builder runs in the context that is provided by a collector of the corresponding flow. For example, consider the implementation of a simple function that prints the thread it is called on and emits three numbers:

1. import kotlinx.coroutines.*
2. import kotlinx.coroutines.flow.*

4. fun log(msg: String) = println("[${Thread.currentThread().name}] $msg")

6. //sampleStart
7. fun simple(): Flow<Int> = flow {
8. log("Started simple flow")
9. for (i in 1..3) {
10. emit(i)
11. }
12. }

14. fun main() = runBlocking<Unit> {
15. simple().collect { value -> log("Collected $value") }
16. }
17. //sampleEnd

Running this code produces:

1. [main @coroutine#1] Started simple flow
2. [main @coroutine#1] Collected 1
3. [main @coroutine#1] Collected 2
4. [main @coroutine#1] Collected 3

Since simple().collect is called from the main thread, the body of simple‘s flow is also called in the main thread. This is the perfect default for fast-running or asynchronous code that does not care about the execution context and does not block the caller.

Wrong emission withContext

However, the long-running CPU-consuming code might need to be executed in the context of Dispatchers.Default and UI-updating code might need to be executed in the context of Dispatchers.Main. Usually, withContext is used to change the context in the code using Kotlin coroutines, but code in the flow { ... } builder has to honor the context preservation property and is not allowed to emit from a different context.

Try running the following code:

1. import kotlinx.coroutines.*
2. import kotlinx.coroutines.flow.*

4. //sampleStart
5. fun simple(): Flow<Int> = flow {
6. // The WRONG way to change context for CPU-consuming code in flow builder
7. kotlinx.coroutines.withContext(Dispatchers.Default) {
8. for (i in 1..3) {
9. Thread.sleep(100) // pretend we are computing it in CPU-consuming way
10. emit(i) // emit next value
11. }
12. }
13. }

15. fun main() = runBlocking<Unit> {
16. simple().collect { value -> println(value) }
17. }
18. //sampleEnd

This code produces the following exception:

1. Exception in thread "main" java.lang.IllegalStateException: Flow invariant is violated:
2. Flow was collected in [CoroutineId(1), "coroutine#1":BlockingCoroutine{Active}@5511c7f8, BlockingEventLoop@2eac3323],
3. but emission happened in [CoroutineId(1), "coroutine#1":DispatchedCoroutine{Active}@2dae0000, Dispatchers.Default].
4. Please refer to 'flow' documentation or use 'flowOn' instead
5. at ...

flowOn operator

The exception refers to the flowOn function that shall be used to change the context of the flow emission. The correct way to change the context of a flow is shown in the example below, which also prints the names of the corresponding threads to show how it all works:

1. import kotlinx.coroutines.*
2. import kotlinx.coroutines.flow.*

4. fun log(msg: String) = println("[${Thread.currentThread().name}] $msg")

6. //sampleStart
7. fun simple(): Flow<Int> = flow {
8. for (i in 1..3) {
9. Thread.sleep(100) // pretend we are computing it in CPU-consuming way
10. log("Emitting $i")
11. emit(i) // emit next value
12. }
13. }.flowOn(Dispatchers.Default) // RIGHT way to change context for CPU-consuming code in flow builder

15. fun main() = runBlocking<Unit> {
16. simple().collect { value ->
17. log("Collected $value")
18. }
19. }
20. //sampleEnd

Notice how flow { ... } works in the background thread, while collection happens in the main thread:

Another thing to observe here is that the flowOn operator has changed the default sequential nature of the flow. Now collection happens in one coroutine (“coroutine#1”) and emission happens in another coroutine (“coroutine#2”) that is running in another thread concurrently with the collecting coroutine. The flowOn operator creates another coroutine for an upstream flow when it has to change the CoroutineDispatcher in its context.

Buffering

Running different parts of a flow in different coroutines can be helpful from the standpoint of the overall time it takes to collect the flow, especially when long-running asynchronous operations are involved. For example, consider a case when the emission by a simple flow is slow, taking 100 ms to produce an element; and collector is also slow, taking 300 ms to process an element. Let’s see how long it takes to collect such a flow with three numbers:

1. import kotlinx.coroutines.*
2. import kotlinx.coroutines.flow.*
3. import kotlin.system.*

5. //sampleStart
6. fun simple(): Flow<Int> = flow {
7. for (i in 1..3) {
8. delay(100) // pretend we are asynchronously waiting 100 ms
9. emit(i) // emit next value
10. }
11. }

13. fun main() = runBlocking<Unit> {
14. val time = measureTimeMillis {
15. simple().collect { value ->
16. delay(300) // pretend we are processing it for 300 ms
17. println(value)
18. }
19. }
20. println("Collected in $time ms")
21. }
22. //sampleEnd

It produces something like this, with the whole collection taking around 1200 ms (three numbers, 400 ms for each):

1. 1
2. 2
3. 3
4. Collected in 1220 ms

We can use a buffer operator on a flow to run emitting code of the simple flow concurrently with collecting code, as opposed to running them sequentially:

1. import kotlinx.coroutines.*
2. import kotlinx.coroutines.flow.*
3. import kotlin.system.*

5. fun simple(): Flow<Int> = flow {
6. for (i in 1..3) {
7. delay(100) // pretend we are asynchronously waiting 100 ms
8. emit(i) // emit next value
9. }
10. }

12. fun main() = runBlocking<Unit> {
13. //sampleStart
14. val time = measureTimeMillis {
15. simple()
16. .buffer() // buffer emissions, don't wait
17. .collect { value ->
18. delay(300) // pretend we are processing it for 300 ms
19. println(value)
20. }
21. }
22. println("Collected in $time ms")
23. //sampleEnd
24. }

It produces the same numbers just faster, as we have effectively created a processing pipeline, having to only wait 100 ms for the first number and then spending only 300 ms to process each number. This way it takes around 1000 ms to run:

1. 1
2. 2
3. 3
4. Collected in 1071 ms

Conflation

When a flow represents partial results of the operation or operation status updates, it may not be necessary to process each value, but instead, only most recent ones. In this case, the conflate operator can be used to skip intermediate values when a collector is too slow to process them. Building on the previous example:

1. import kotlinx.coroutines.*
2. import kotlinx.coroutines.flow.*
3. import kotlin.system.*

5. fun simple(): Flow<Int> = flow {
6. for (i in 1..3) {
7. delay(100) // pretend we are asynchronously waiting 100 ms
8. emit(i) // emit next value
9. }
10. }

12. fun main() = runBlocking<Unit> {
13. //sampleStart
14. val time = measureTimeMillis {
15. simple()
16. .conflate() // conflate emissions, don't process each one
17. .collect { value ->
18. delay(300) // pretend we are processing it for 300 ms
19. println(value)
20. }
21. }
22. println("Collected in $time ms")
23. //sampleEnd
24. }

We see that while the first number was still being processed the second, and third were already produced, so the second one was conflated and only the most recent (the third one) was delivered to the collector:

1. 1
2. 3
3. Collected in 758 ms

Processing the latest value

Conflation is one way to speed up processing when both the emitter and collector are slow. It does it by dropping emitted values. The other way is to cancel a slow collector and restart it every time a new value is emitted. There is a family of xxxLatest operators that perform the same essential logic of a xxx operator, but cancel the code in their block on a new value. Let’s try changing conflate to collectLatest in the previous example:

1. import kotlinx.coroutines.*
2. import kotlinx.coroutines.flow.*
3. import kotlin.system.*

5. fun simple(): Flow<Int> = flow {
6. for (i in 1..3) {
7. delay(100) // pretend we are asynchronously waiting 100 ms
8. emit(i) // emit next value
9. }
10. }

12. fun main() = runBlocking<Unit> {
13. //sampleStart
14. val time = measureTimeMillis {
15. simple()
16. .collectLatest { value -> // cancel & restart on the latest value
17. println("Collecting $value")
18. delay(300) // pretend we are processing it for 300 ms
19. println("Done $value")
20. }
21. }
22. println("Collected in $time ms")
23. //sampleEnd
24. }

Since the body of collectLatest takes 300 ms, but new values are emitted every 100 ms, we see that the block is run on every value, but completes only for the last value:

1. Collecting 1
2. Collecting 2
3. Collecting 3
4. Done 3
5. Collected in 741 ms

Composing multiple flows

There are lots of ways to compose multiple flows.

Zip

Just like the Sequence.zip extension function in the Kotlin standard library, flows have a zip operator that combines the corresponding values of two flows:

1. import kotlinx.coroutines.*
2. import kotlinx.coroutines.flow.*

4. fun main() = runBlocking<Unit> {
5. //sampleStart
6. val nums = (1..3).asFlow() // numbers 1..3
7. val strs = flowOf("one", "two", "three") // strings
8. nums.zip(strs) { a, b -> "$a -> $b" } // compose a single string
9. .collect { println(it) } // collect and print
10. //sampleEnd
11. }

This example prints:

1. 1 -> one
2. 2 -> two
3. 3 -> three

Combine

When flow represents the most recent value of a variable or operation (see also the related section on conflation), it might be needed to perform a computation that depends on the most recent values of the corresponding flows and to recompute it whenever any of the upstream flows emit a value. The corresponding family of operators is called combine.

For example, if the numbers in the previous example update every 300ms, but strings update every 400 ms, then zipping them using the zip operator will still produce the same result, albeit results that are printed every 400 ms:

1. import kotlinx.coroutines.*
2. import kotlinx.coroutines.flow.*

4. fun main() = runBlocking<Unit> {
5. //sampleStart
6. val nums = (1..3).asFlow().onEach { delay(300) } // numbers 1..3 every 300 ms
7. val strs = flowOf("one", "two", "three").onEach { delay(400) } // strings every 400 ms
8. val startTime = System.currentTimeMillis() // remember the start time
9. nums.zip(strs) { a, b -> "$a -> $b" } // compose a single string with "zip"
10. .collect { value -> // collect and print
11. println("$value at ${System.currentTimeMillis() - startTime} ms from start")
12. }
13. //sampleEnd
14. }

However, when using a combine operator here instead of a zip:

1. import kotlinx.coroutines.*
2. import kotlinx.coroutines.flow.*

4. fun main() = runBlocking<Unit> {
5. //sampleStart
6. val nums = (1..3).asFlow().onEach { delay(300) } // numbers 1..3 every 300 ms
7. val strs = flowOf("one", "two", "three").onEach { delay(400) } // strings every 400 ms
8. val startTime = System.currentTimeMillis() // remember the start time
9. nums.combine(strs) { a, b -> "$a -> $b" } // compose a single string with "combine"
10. .collect { value -> // collect and print
11. println("$value at ${System.currentTimeMillis() - startTime} ms from start")
12. }
13. //sampleEnd
14. }

We get quite a different output, where a line is printed at each emission from either nums or strs flows:

1. 1 -> one at 452 ms from start
2. 2 -> one at 651 ms from start
3. 2 -> two at 854 ms from start
4. 3 -> two at 952 ms from start
5. 3 -> three at 1256 ms from start

Flattening flows

Flows represent asynchronously received sequences of values, so it is quite easy to get in a situation where each value triggers a request for another sequence of values. For example, we can have the following function that returns a flow of two strings 500 ms apart:

1. fun requestFlow(i: Int): Flow<String> = flow {
2. emit("$i: First")
3. delay(500) // wait 500 ms
4. emit("$i: Second")
5. }

Now if we have a flow of three integers and call requestFlow for each of them like this:

1. (1..3).asFlow().map { requestFlow(it) }

Then we end up with a flow of flows (Flow<Flow<String>>) that needs to be flattened into a single flow for further processing. Collections and sequences have flatten and flatMap operators for this. However, due to the asynchronous nature of flows they call for different modes of flattening, as such, there is a family of flattening operators on flows.

flatMapConcat

Concatenating mode is implemented by flatMapConcat and flattenConcat operators. They are the most direct analogues of the corresponding sequence operators. They wait for the inner flow to complete before starting to collect the next one as the following example shows:

1. import kotlinx.coroutines.*
2. import kotlinx.coroutines.flow.*

4. fun requestFlow(i: Int): Flow<String> = flow {
5. emit("$i: First")
6. delay(500) // wait 500 ms
7. emit("$i: Second")
8. }

10. fun main() = runBlocking<Unit> {
11. //sampleStart
12. val startTime = System.currentTimeMillis() // remember the start time
13. (1..3).asFlow().onEach { delay(100) } // a number every 100 ms
14. .flatMapConcat { requestFlow(it) }
15. .collect { value -> // collect and print
16. println("$value at ${System.currentTimeMillis() - startTime} ms from start")
17. }
18. //sampleEnd
19. }

The sequential nature of flatMapConcat is clearly seen in the output:

1. 1: First at 121 ms from start
2. 1: Second at 622 ms from start
3. 2: First at 727 ms from start
4. 2: Second at 1227 ms from start
5. 3: First at 1328 ms from start
6. 3: Second at 1829 ms from start

flatMapMerge

Another flattening mode is to concurrently collect all the incoming flows and merge their values into a single flow so that values are emitted as soon as possible. It is implemented by flatMapMerge and flattenMerge operators. They both accept an optional concurrency parameter that limits the number of concurrent flows that are collected at the same time (it is equal to DEFAULT_CONCURRENCY by default).

1. import kotlinx.coroutines.*
2. import kotlinx.coroutines.flow.*

4. fun requestFlow(i: Int): Flow<String> = flow {
5. emit("$i: First")
6. delay(500) // wait 500 ms
7. emit("$i: Second")
8. }

10. fun main() = runBlocking<Unit> {
11. //sampleStart
12. val startTime = System.currentTimeMillis() // remember the start time
13. (1..3).asFlow().onEach { delay(100) } // a number every 100 ms
14. .flatMapMerge { requestFlow(it) }
15. .collect { value -> // collect and print
16. println("$value at ${System.currentTimeMillis() - startTime} ms from start")
17. }
18. //sampleEnd
19. }

The concurrent nature of flatMapMerge is obvious:

1. 1: First at 136 ms from start
2. 2: First at 231 ms from start
3. 3: First at 333 ms from start
4. 1: Second at 639 ms from start
5. 2: Second at 732 ms from start
6. 3: Second at 833 ms from start

flatMapLatest

In a similar way to the collectLatest operator, that was shown in “Processing the latest value” section, there is the corresponding “Latest” flattening mode where a collection of the previous flow is cancelled as soon as new flow is emitted. It is implemented by the flatMapLatest operator.

1. import kotlinx.coroutines.*
2. import kotlinx.coroutines.flow.*

4. fun requestFlow(i: Int): Flow<String> = flow {
5. emit("$i: First")
6. delay(500) // wait 500 ms
7. emit("$i: Second")
8. }

10. fun main() = runBlocking<Unit> {
11. //sampleStart
12. val startTime = System.currentTimeMillis() // remember the start time
13. (1..3).asFlow().onEach { delay(100) } // a number every 100 ms
14. .flatMapLatest { requestFlow(it) }
15. .collect { value -> // collect and print
16. println("$value at ${System.currentTimeMillis() - startTime} ms from start")
17. }
18. //sampleEnd
19. }

The output here in this example is a good demonstration of how flatMapLatest works:

1. 1: First at 142 ms from start
2. 2: First at 322 ms from start
3. 3: First at 425 ms from start
4. 3: Second at 931 ms from start

Flow exceptions

Flow collection can complete with an exception when an emitter or code inside the operators throw an exception. There are several ways to handle these exceptions.

Collector try and catch

A collector can use Kotlin’s try/catch block to handle exceptions:

1. import kotlinx.coroutines.*
2. import kotlinx.coroutines.flow.*

4. //sampleStart
5. fun simple(): Flow<Int> = flow {
6. for (i in 1..3) {
7. println("Emitting $i")
8. emit(i) // emit next value
9. }
10. }

12. fun main() = runBlocking<Unit> {
13. try {
14. simple().collect { value ->
15. println(value)
16. check(value <= 1) { "Collected $value" }
17. }
18. } catch (e: Throwable) {
19. println("Caught $e")
20. }
21. }
22. //sampleEnd

This code successfully catches an exception in collect terminal operator and, as we see, no more values are emitted after that:

1. Emitting 1
2. 1
3. Emitting 2
4. 2
5. Caught java.lang.IllegalStateException: Collected 2

Everything is caught

The previous example actually catches any exception happening in the emitter or in any intermediate or terminal operators. For example, let’s change the code so that emitted values are mapped to strings, but the corresponding code produces an exception:

1. import kotlinx.coroutines.*
2. import kotlinx.coroutines.flow.*

4. //sampleStart
5. fun simple(): Flow<String> =
6. flow {
7. for (i in 1..3) {
8. println("Emitting $i")
9. emit(i) // emit next value
10. }
11. }
12. .map { value ->
13. check(value <= 1) { "Crashed on $value" }
14. "string $value"
15. }

17. fun main() = runBlocking<Unit> {
18. try {
19. simple().collect { value -> println(value) }
20. } catch (e: Throwable) {
21. println("Caught $e")
22. }
23. }
24. //sampleEnd

This exception is still caught and collection is stopped:

1. Emitting 1
2. string 1
3. Emitting 2
4. Caught java.lang.IllegalStateException: Crashed on 2

Exception transparency

But how can code of the emitter encapsulate its exception handling behavior?

Flows must be transparent to exceptions and it is a violation of the exception transparency to emit values in the