Flow docs (#1432)
* Flow guide by example
* Consistent chapter naming and ordering in docs
* Make "supervision" a subsection of "Exception Handling" chapter
* Knit: filter out sampleStart/End from Knit-generated source files
* Knit: Support example auto-numbering
diff --git a/docs/_nav.yml b/docs/_nav.yml
index 3a7dbfc..79a6969 100644
--- a/docs/_nav.yml
+++ b/docs/_nav.yml
@@ -7,21 +7,24 @@
- md: cancellation-and-timeouts.md
url: cancellation-and-timeouts.html
title: Cancellation and Timeouts
-- md: channels.md
- url: channels.html
- title: Channels
- md: composing-suspending-functions.md
url: composing-suspending-functions.html
title: Composing Suspending Functions
- md: coroutine-context-and-dispatchers.md
url: coroutine-context-and-dispatchers.html
title: Coroutine Context and Dispatchers
+- md: flow.md
+ url: flow.html
+ title: Asynchronous Flow
+- md: channels.md
+ url: channels.html
+ title: Channels
- md: exception-handling.md
url: exception-handling.html
- title: Exception Handling
-- md: select-expression.md
- url: select-expression.html
- title: Select Expression
+ title: Exception Handling and Supervision
- md: shared-mutable-state-and-concurrency.md
url: shared-mutable-state-and-concurrency.html
title: Shared Mutable State and Concurrency
+- md: select-expression.md
+ url: select-expression.html
+ title: Select Expression (experimental)
diff --git a/docs/basics.md b/docs/basics.md
index a7321bb..6a1248b 100644
--- a/docs/basics.md
+++ b/docs/basics.md
@@ -20,7 +20,7 @@
<!--- TOC -->
-* [Coroutine basics](#coroutine-basics)
+* [Coroutine Basics](#coroutine-basics)
* [Your first coroutine](#your-first-coroutine)
* [Bridging blocking and non-blocking worlds](#bridging-blocking-and-non-blocking-worlds)
* [Waiting for a job](#waiting-for-a-job)
@@ -33,7 +33,7 @@
<!--- END_TOC -->
-## Coroutine basics
+## Coroutine Basics
This section covers basic coroutine concepts.
diff --git a/docs/cancellation-and-timeouts.md b/docs/cancellation-and-timeouts.md
index afb8da9..ef4a9c9 100644
--- a/docs/cancellation-and-timeouts.md
+++ b/docs/cancellation-and-timeouts.md
@@ -19,7 +19,7 @@
<!--- TOC -->
-* [Cancellation and timeouts](#cancellation-and-timeouts)
+* [Cancellation and Timeouts](#cancellation-and-timeouts)
* [Cancelling coroutine execution](#cancelling-coroutine-execution)
* [Cancellation is cooperative](#cancellation-is-cooperative)
* [Making computation code cancellable](#making-computation-code-cancellable)
@@ -29,7 +29,7 @@
<!--- END_TOC -->
-## Cancellation and timeouts
+## Cancellation and Timeouts
This section covers coroutine cancellation and timeouts.
diff --git a/docs/composing-suspending-functions.md b/docs/composing-suspending-functions.md
index a910f49..0cd0276 100644
--- a/docs/composing-suspending-functions.md
+++ b/docs/composing-suspending-functions.md
@@ -20,7 +20,7 @@
<!--- TOC -->
-* [Composing suspending functions](#composing-suspending-functions)
+* [Composing Suspending Functions](#composing-suspending-functions)
* [Sequential by default](#sequential-by-default)
* [Concurrent using async](#concurrent-using-async)
* [Lazily started async](#lazily-started-async)
@@ -29,7 +29,7 @@
<!--- END_TOC -->
-## Composing suspending functions
+## Composing Suspending Functions
This section covers various approaches to composition of suspending functions.
diff --git a/docs/coroutine-context-and-dispatchers.md b/docs/coroutine-context-and-dispatchers.md
index 4769c1e..558b039 100644
--- a/docs/coroutine-context-and-dispatchers.md
+++ b/docs/coroutine-context-and-dispatchers.md
@@ -20,7 +20,7 @@
<!--- TOC -->
-* [Coroutine context and dispatchers](#coroutine-context-and-dispatchers)
+* [Coroutine Context and Dispatchers](#coroutine-context-and-dispatchers)
* [Dispatchers and threads](#dispatchers-and-threads)
* [Unconfined vs confined dispatcher](#unconfined-vs-confined-dispatcher)
* [Debugging coroutines and threads](#debugging-coroutines-and-threads)
@@ -35,7 +35,7 @@
<!--- END_TOC -->
-## Coroutine context and dispatchers
+## Coroutine Context and Dispatchers
Coroutines always execute in some context represented by a value of the
[CoroutineContext](https://kotlinlang.org/api/latest/jvm/stdlib/kotlin.coroutines/-coroutine-context/)
diff --git a/docs/coroutines-guide.md b/docs/coroutines-guide.md
index 3d0ca53..5f41d06 100644
--- a/docs/coroutines-guide.md
+++ b/docs/coroutines-guide.md
@@ -15,14 +15,15 @@
## Table of contents
-* [Coroutine basics](basics.md)
-* [Cancellation and timeouts](cancellation-and-timeouts.md)
-* [Composing suspending functions](composing-suspending-functions.md)
-* [Coroutine context and dispatchers](coroutine-context-and-dispatchers.md)
-* [Exception handling and supervision](exception-handling.md)
-* [Channels (experimental)](channels.md)
-* [Shared mutable state and concurrency](shared-mutable-state-and-concurrency.md)
-* [Select expression (experimental)](select-expression.md)
+* [Basics](basics.md)
+* [Cancellation and Timeouts](cancellation-and-timeouts.md)
+* [Composing Suspending Functions](composing-suspending-functions.md)
+* [Coroutine Context and Dispatchers](coroutine-context-and-dispatchers.md)
+* [Asynchronous Flow](flow.md)
+* [Channels](channels.md)
+* [Exception Handling and Supervision](exception-handling.md)
+* [Shared Mutable State and Concurrency](shared-mutable-state-and-concurrency.md)
+* [Select Expression (experimental)](select-expression.md)
## Additional references
diff --git a/docs/exception-handling.md b/docs/exception-handling.md
index 349b703..409acd5 100644
--- a/docs/exception-handling.md
+++ b/docs/exception-handling.md
@@ -19,19 +19,19 @@
<!--- TOC -->
-* [Exception handling](#exception-handling)
+* [Exception Handling](#exception-handling)
* [Exception propagation](#exception-propagation)
* [CoroutineExceptionHandler](#coroutineexceptionhandler)
* [Cancellation and exceptions](#cancellation-and-exceptions)
* [Exceptions aggregation](#exceptions-aggregation)
-* [Supervision](#supervision)
- * [Supervision job](#supervision-job)
- * [Supervision scope](#supervision-scope)
- * [Exceptions in supervised coroutines](#exceptions-in-supervised-coroutines)
+ * [Supervision](#supervision)
+ * [Supervision job](#supervision-job)
+ * [Supervision scope](#supervision-scope)
+ * [Exceptions in supervised coroutines](#exceptions-in-supervised-coroutines)
<!--- END_TOC -->
-## Exception handling
+## Exception Handling
This section covers exception handling and cancellation on exceptions.
@@ -355,7 +355,7 @@
```
<!--- TEST-->
-## Supervision
+### Supervision
As we have studied before, cancellation is a bidirectional relationship propagating through the whole
coroutines hierarchy. But what if unidirectional cancellation is required?
@@ -367,7 +367,7 @@
Another example is a server process that spawns several children jobs and needs to _supervise_
their execution, tracking their failures and restarting just those children jobs that had failed.
-### Supervision job
+#### Supervision job
For these purposes [SupervisorJob][SupervisorJob()] can be used. It is similar to a regular [Job][Job()] with the only exception that cancellation is propagated
only downwards. It is easy to demonstrate with an example:
@@ -421,7 +421,7 @@
<!--- TEST-->
-### Supervision scope
+#### Supervision scope
For *scoped* concurrency [supervisorScope] can be used instead of [coroutineScope] for the same purpose. It propagates cancellation
only in one direction and cancels all children only if it has failed itself. It also waits for all children before completion
@@ -469,7 +469,7 @@
```
<!--- TEST-->
-### Exceptions in supervised coroutines
+#### Exceptions in supervised coroutines
Another crucial difference between regular and supervisor jobs is exception handling.
Every child should handle its exceptions by itself via exception handling mechanisms.
diff --git a/docs/flow.md b/docs/flow.md
new file mode 100644
index 0000000..134d808
--- /dev/null
+++ b/docs/flow.md
@@ -0,0 +1,1855 @@
+<!--- INCLUDE .*/example-([a-z]+)-([0-9a-z]+)\.kt
+/*
+ * Copyright 2016-2019 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license.
+ */
+
+// This file was automatically generated from coroutines-guide.md by Knit tool. Do not edit.
+package kotlinx.coroutines.guide.$$1$$2
+-->
+<!--- KNIT ../kotlinx-coroutines-core/jvm/test/guide/.*-##\.kt -->
+<!--- TEST_OUT ../kotlinx-coroutines-core/jvm/test/guide/test/FlowGuideTest.kt
+// This file was automatically generated from flow.md by Knit tool. Do not edit.
+package kotlinx.coroutines.guide.test
+
+import org.junit.Test
+
+class FlowGuideTest {
+-->
+
+**Table of contents**
+
+<!--- TOC -->
+
+* [Asynchronous Flow](#asynchronous-flow)
+ * [Representing multiple values](#representing-multiple-values)
+ * [Sequences](#sequences)
+ * [Suspending functions](#suspending-functions)
+ * [Flows](#flows)
+ * [Flows are cold](#flows-are-cold)
+ * [Flow cancellation](#flow-cancellation)
+ * [Flow builders](#flow-builders)
+ * [Intermediate flow operators](#intermediate-flow-operators)
+ * [Transform operator](#transform-operator)
+ * [Size-limiting operators](#size-limiting-operators)
+ * [Terminal flow operators](#terminal-flow-operators)
+ * [Flows are sequential](#flows-are-sequential)
+ * [Flow context](#flow-context)
+ * [Wrong emission withContext](#wrong-emission-withcontext)
+ * [flowOn operator](#flowon-operator)
+ * [Buffering](#buffering)
+ * [Conflation](#conflation)
+ * [Processing the latest value](#processing-the-latest-value)
+ * [Composing multiple flows](#composing-multiple-flows)
+ * [Zip](#zip)
+ * [Combine](#combine)
+ * [Flattening flows](#flattening-flows)
+ * [flatMapConcat](#flatmapconcat)
+ * [flatMapMerge](#flatmapmerge)
+ * [flatMapLatest](#flatmaplatest)
+ * [Flow exceptions](#flow-exceptions)
+ * [Collector try and catch](#collector-try-and-catch)
+ * [Everything is caught](#everything-is-caught)
+ * [Exception transparency](#exception-transparency)
+ * [Transparent catch](#transparent-catch)
+ * [Catching declaratively](#catching-declaratively)
+ * [Flow completion](#flow-completion)
+ * [Imperative finally block](#imperative-finally-block)
+ * [Declarative handling](#declarative-handling)
+ * [Upstream exceptions only](#upstream-exceptions-only)
+ * [Imperative versus declarative](#imperative-versus-declarative)
+ * [Launching flow](#launching-flow)
+
+<!--- END_TOC -->
+
+## Asynchronous Flow
+
+Suspending functions asynchronously return a single value, but how can you return
+multiple asynchronously computed values? That is what Kotlin Flows are for.
+
+### Representing multiple values
+
+Multiple values can be represented in Kotlin using [collections].
+For example, we can have a function `foo()` that returns a [List]
+of three numbers and print them all using [forEach]:
+
+<div class="sample" markdown="1" theme="idea" data-min-compiler-version="1.3">
+
+```kotlin
+fun foo(): List<Int> = listOf(1, 2, 3)
+
+fun main() {
+ foo().forEach { value -> println(value) }
+}
+```
+
+</div>
+
+> You can get full code [here](../kotlinx-coroutines-core/jvm/test/guide/example-flow-01.kt).
+
+This code outputs:
+
+```text
+1
+2
+3
+```
+
+<!--- TEST -->
+
+#### Sequences
+
+If the numbers are computed with some CPU-consuming blocking code
+(each computation taking 100ms) then we can represent the numbers using a [Sequence]:
+
+<div class="sample" markdown="1" theme="idea" data-min-compiler-version="1.3">
+
+```kotlin
+fun foo(): Sequence<Int> = sequence { // sequence builder
+ for (i in 1..3) {
+ Thread.sleep(100) // pretend we are computing it
+ yield(i) // yield next value
+ }
+}
+
+fun main() {
+ foo().forEach { value -> println(value) }
+}
+```
+
+</div>
+
+> You can get full code [here](../kotlinx-coroutines-core/jvm/test/guide/example-flow-02.kt).
+
+This code outputs the same numbers, but it waits 100ms before printing each one.
+
+<!--- TEST
+1
+2
+3
+-->
+
+#### Suspending functions
+
+However, this computation blocks the main thread that is running the code.
+When those values are computed by an asynchronous code we can mark function `foo` with a `suspend` modifier,
+so that it can perform its work without blocking and return the result as a list:
+
+<div class="sample" markdown="1" theme="idea" data-min-compiler-version="1.3">
+
+```kotlin
+import kotlinx.coroutines.*
+
+//sampleStart
+suspend fun foo(): List<Int> {
+ delay(1000) // pretend we are doing something asynchronous here
+ return listOf(1, 2, 3)
+}
+
+fun main() = runBlocking<Unit> {
+ foo().forEach { value -> println(value) }
+}
+//sampleEnd
+```
+
+</div>
+
+> You can get full code [here](../kotlinx-coroutines-core/jvm/test/guide/example-flow-03.kt).
+
+This code prints the numbers after waiting for a second.
+
+<!--- TEST
+1
+2
+3
+-->
+
+#### Flows
+
+Using `List<Int>` result type we can only return all the values at once. To represent
+the stream of values that are being asynchronously computed we can use [`Flow<Int>`][Flow] type similarly
+to the `Sequence<Int>` type for synchronously computed values:
+
+<div class="sample" markdown="1" theme="idea" data-min-compiler-version="1.3">
+
+```kotlin
+import kotlinx.coroutines.*
+import kotlinx.coroutines.flow.*
+
+//sampleStart
+fun foo(): Flow<Int> = flow { // flow builder
+ for (i in 1..3) {
+ delay(100) // pretend we are doing something useful here
+ emit(i) // emit next value
+ }
+}
+
+fun main() = runBlocking<Unit> {
+ // Launch a concurrent coroutine to see that the main thread is not blocked
+ launch {
+ for (k in 1..3) {
+ println("I'm not blocked $k")
+ delay(100)
+ }
+ }
+ // Collect the flow
+ foo().collect { value -> println(value) }
+}
+//sampleEnd
+```
+
+</div>
+
+> You can get full code [here](../kotlinx-coroutines-core/jvm/test/guide/example-flow-04.kt).
+
+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:
+
+```text
+I'm not blocked 1
+1
+I'm not blocked 2
+2
+I'm not blocked 3
+3
+```
+
+<!--- TEST -->
+
+Notice the following differences of 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 function `foo()` is no longer marked with `suspend` modifier.
+* Values are _emitted_ from the flow using [emit][FlowCollector.emit] function.
+* Values are _collected_ from the flow using [collect][collect] function.
+
+> You can replace [delay] with `Thread.sleep` in the body of `foo`'s `flow { ... }` and see that the main
+thread is blocked in this case.
+
+### Flows are cold
+
+Flows are _cold_ streams similarly to sequences — the code inside a [flow] builder does not
+run until the flow is collected. This becomes clear in the following example:
+
+<div class="sample" markdown="1" theme="idea" data-min-compiler-version="1.3">
+
+```kotlin
+import kotlinx.coroutines.*
+import kotlinx.coroutines.flow.*
+
+//sampleStart
+fun foo(): Flow<Int> = flow {
+ println("Flow started")
+ for (i in 1..3) {
+ delay(100)
+ emit(i)
+ }
+}
+
+fun main() = runBlocking<Unit> {
+ println("Calling foo...")
+ val flow = foo()
+ println("Calling collect...")
+ flow.collect { value -> println(value) }
+ println("Calling collect again...")
+ flow.collect { value -> println(value) }
+}
+//sampleEnd
+```
+
+</div>
+
+> You can get full code [here](../kotlinx-coroutines-core/jvm/test/guide/example-flow-05.kt).
+
+Which prints:
+
+```text
+Calling foo...
+Calling collect...
+Flow started
+1
+2
+3
+Calling collect again...
+Flow started
+1
+2
+3
+```
+
+<!--- TEST -->
+
+That is a key reason why the `foo()` function (which returns a flow) is not marked with `suspend` modifier.
+By itself, `foo()` returns quickly and does not wait for anything. The flow starts every time it is collected,
+that is why we see that when we call `collect` again, we get "Flow started" printed again.
+
+### Flow cancellation
+
+Flow adheres to general cooperative cancellation of coroutines. However, flow infrastructure does not introduce
+additional cancellation points. It is fully transparent for cancellation. As usual, flow collection can be
+cancelled when the flow is suspended in a cancellable suspending function (like [delay]) and cannot be cancelled otherwise.
+
+The following example shows how the flow gets cancelled on timeout when running in [withTimeoutOrNull] block
+and stops executing its code:
+
+<div class="sample" markdown="1" theme="idea" data-min-compiler-version="1.3">
+
+```kotlin
+import kotlinx.coroutines.*
+import kotlinx.coroutines.flow.*
+
+//sampleStart
+fun foo(): Flow<Int> = flow {
+ for (i in 1..3) {
+ delay(100)
+ println("Emitting $i")
+ emit(i)
+ }
+}
+
+fun main() = runBlocking<Unit> {
+ withTimeoutOrNull(250) { // Timeout after 250ms
+ foo().collect { value -> println(value) }
+ }
+ println("Done")
+}
+//sampleEnd
+```
+
+</div>
+
+> You can get full code [here](../kotlinx-coroutines-core/jvm/test/guide/example-flow-06.kt).
+
+Notice how only two numbers get emitted by the flow in `foo()` function, producing the following output:
+
+```text
+Emitting 1
+1
+Emitting 2
+2
+Done
+```
+
+<!--- TEST -->
+
+### Flow builders
+
+The `flow { ... }` builder from the previous examples is the most basic one. There are other builders for
+convenient 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.
+
+Thus, the example that prints numbers from 1 to 3 from a flow can be written as:
+
+<div class="sample" markdown="1" theme="idea" data-min-compiler-version="1.3">
+
+```kotlin
+import kotlinx.coroutines.*
+import kotlinx.coroutines.flow.*
+
+fun main() = runBlocking<Unit> {
+//sampleStart
+ // Convert an integer range to a flow
+ (1..3).asFlow().collect { value -> println(value) }
+//sampleEnd
+}
+```
+
+</div>
+
+> You can get full code [here](../kotlinx-coroutines-core/jvm/test/guide/example-flow-07.kt).
+
+<!--- TEST
+1
+2
+3
+-->
+
+### Intermediate flow operators
+
+Flows can be transformed with operators similarly to 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 from sequences is that blocks of
+code inside those operators can call suspending functions.
+
+For example, a flow of incoming requests can be
+mapped to results with the [map] operator even when performing a request is a long-running
+operation that is implemented by a suspending function:
+
+<div class="sample" markdown="1" theme="idea" data-min-compiler-version="1.3">
+
+```kotlin
+import kotlinx.coroutines.*
+import kotlinx.coroutines.flow.*
+
+//sampleStart
+suspend fun performRequest(request: Int): String {
+ delay(1000) // imitate long-running asynchronous work
+ return "response $request"
+}
+
+fun main() = runBlocking<Unit> {
+ (1..3).asFlow() // a flow of requests
+ .map { request -> performRequest(request) }
+ .collect { response -> println(response) }
+}
+//sampleEnd
+```
+
+</div>
+
+> You can get full code [here](../kotlinx-coroutines-core/jvm/test/guide/example-flow-08.kt).
+
+It produces the following three lines, each line appearing after a second:
+
+```text
+response 1
+response 2
+response 3
+```
+
+<!--- TEST -->
+
+#### 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 `transform` operator, you can [emit][FlowCollector.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:
+
+<div class="sample" markdown="1" theme="idea" data-min-compiler-version="1.3">
+
+```kotlin
+import kotlinx.coroutines.*
+import kotlinx.coroutines.flow.*
+
+suspend fun performRequest(request: Int): String {
+ delay(1000) // imitate long-running asynchronous work
+ return "response $request"
+}
+
+fun main() = runBlocking<Unit> {
+//sampleStart
+ (1..3).asFlow() // a flow of requests
+ .transform { request ->
+ emit("Making request $request")
+ emit(performRequest(request))
+ }
+ .collect { response -> println(response) }
+//sampleEnd
+}
+```
+
+</div>
+
+> You can get full code [here](../kotlinx-coroutines-core/jvm/test/guide/example-flow-09.kt).
+
+The output of this code is:
+
+```text
+Making request 1
+response 1
+Making request 2
+response 2
+Making request 3
+response 3
+```
+
+<!--- TEST -->
+
+#### 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:
+
+<div class="sample" markdown="1" theme="idea" data-min-compiler-version="1.3">
+
+```kotlin
+import kotlinx.coroutines.*
+import kotlinx.coroutines.flow.*
+
+//sampleStart
+fun numbers(): Flow<Int> = flow {
+ try {
+ emit(1)
+ emit(2)
+ println("This line will not execute")
+ emit(3)
+ } finally {
+ println("Finally in numbers")
+ }
+}
+
+fun main() = runBlocking<Unit> {
+ numbers()
+ .take(2) // take only the first two
+ .collect { value -> println(value) }
+}
+//sampleEnd
+```
+
+</div>
+
+> You can get full code [here](../kotlinx-coroutines-core/jvm/test/guide/example-flow-10.kt).
+
+The output of this code clearly shows that execution of the `flow { ... }` body in `numbers()` function
+had stopped after emitting the second number:
+
+```text
+1
+2
+Finally in numbers
+```
+
+<!--- TEST -->
+
+### 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 for
+convenience:
+
+* 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:
+
+<div class="sample" markdown="1" theme="idea" data-min-compiler-version="1.3">
+
+```kotlin
+import kotlinx.coroutines.*
+import kotlinx.coroutines.flow.*
+
+fun main() = runBlocking<Unit> {
+//sampleStart
+ val sum = (1..5).asFlow()
+ .map { it * it } // squares of numbers from 1 to 5
+ .reduce { a, b -> a + b } // sum them (terminal operator)
+ println(sum)
+//sampleEnd
+}
+```
+
+</div>
+
+> You can get full code [here](../kotlinx-coroutines-core/jvm/test/guide/example-flow-11.kt).
+
+Prints a single number:
+
+```text
+55
+```
+
+<!--- TEST -->
+
+### 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 intermediate operators from
+upstream to downstream and is delivered to the terminal operator after that.
+
+See the following example that filters even integers and maps them to strings:
+
+<div class="sample" markdown="1" theme="idea" data-min-compiler-version="1.3">
+
+```kotlin
+import kotlinx.coroutines.*
+import kotlinx.coroutines.flow.*
+
+fun main() = runBlocking<Unit> {
+//sampleStart
+ (1..5).asFlow()
+ .filter {
+ println("Filter $it")
+ it % 2 == 0
+ }
+ .map {
+ println("Map $it")
+ "string $it"
+ }.collect {
+ println("Collect $it")
+ }
+//sampleEnd
+}
+```
+
+</div>
+
+> You can get full code [here](../kotlinx-coroutines-core/jvm/test/guide/example-flow-12.kt).
+
+Producing:
+
+```text
+Filter 1
+Filter 2
+Map 2
+Collect string 2
+Filter 3
+Filter 4
+Map 4
+Collect string 4
+Filter 5
+```
+
+<!--- TEST -->
+
+### Flow context
+
+Collection of a flow always happens in the context of the calling coroutine. For example, if there is
+a `foo` flow, then the following code runs in the context specified
+by the author of this code, regardless of implementation details of the `foo` flow:
+
+<div class="sample" markdown="1" theme="idea" data-highlight-only>
+
+```kotlin
+withContext(context) {
+ foo.collect { value ->
+ println(value) // run in the specified context
+ }
+}
+```
+
+</div>
+
+<!--- CLEAR -->
+
+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 `foo` that prints the thread
+it is called on and emits three numbers:
+
+<div class="sample" markdown="1" theme="idea" data-min-compiler-version="1.3">
+
+```kotlin
+import kotlinx.coroutines.*
+import kotlinx.coroutines.flow.*
+
+fun log(msg: String) = println("[${Thread.currentThread().name}] $msg")
+
+//sampleStart
+fun foo(): Flow<Int> = flow {
+ log("Started foo flow")
+ for (i in 1..3) {
+ emit(i)
+ }
+}
+
+fun main() = runBlocking<Unit> {
+ foo().collect { value -> log("Collected $value") }
+}
+//sampleEnd
+```
+
+</div>
+
+> You can get full code [here](../kotlinx-coroutines-core/jvm/test/guide/example-flow-13.kt).
+
+Running this code produces:
+
+```text
+[main @coroutine#1] Started foo flow
+[main @coroutine#1] Collected 1
+[main @coroutine#1] Collected 2
+[main @coroutine#1] Collected 3
+```
+
+<!--- TEST FLEXIBLE_THREAD -->
+
+Since `foo().collect` is called from the main thread, the body of `foo`'s flow is also called in the main thread.
+This is a 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 code using Kotlin coroutines, but code in the `flow { ... }` builder has to honor context
+preservation property and is not allowed to [emit][FlowCollector.emit] from a different context.
+
+Try running the following code:
+
+<div class="sample" markdown="1" theme="idea" data-min-compiler-version="1.3">
+
+```kotlin
+import kotlinx.coroutines.*
+import kotlinx.coroutines.flow.*
+
+//sampleStart
+fun foo(): Flow<Int> = flow {
+ // WRONG way to change context for CPU-consuming code in flow builder
+ kotlinx.coroutines.withContext(Dispatchers.Default) {
+ for (i in 1..3) {
+ Thread.sleep(100) // pretend we are computing it in CPU-consuming way
+ emit(i) // emit next value
+ }
+ }
+}
+
+fun main() = runBlocking<Unit> {
+ foo().collect { value -> println(value) }
+}
+//sampleEnd
+```
+
+</div>
+
+> You can get full code [here](../kotlinx-coroutines-core/jvm/test/guide/example-flow-14.kt).
+
+This code produces the following exception:
+
+<!--- TEST EXCEPTION
+Exception in thread "main" java.lang.IllegalStateException: Flow invariant is violated:
+ Flow was collected in [CoroutineId(1), "coroutine#1":BlockingCoroutine{Active}@5511c7f8, BlockingEventLoop@2eac3323],
+ but emission happened in [CoroutineId(1), "coroutine#1":DispatchedCoroutine{Active}@2dae0000, DefaultDispatcher].
+ Please refer to 'flow' documentation or use 'flowOn' instead
+ at ...
+-->
+
+> Note that we had to use a fully qualified name of [kotlinx.coroutines.withContext][withContext] function in this example to
+demonstrate this exception. A short name of `withContext` would have resolved to a special stub function that
+produces compilation error to prevent us from running into this problem.
+
+#### flowOn operator
+
+The exception refers to [flowOn] function that shall be used to change the context of flow emission.
+The correct way of changing the context of a flow is shown in the below example, which also prints
+names of the corresponding threads to show how it all works:
+
+<div class="sample" markdown="1" theme="idea" data-min-compiler-version="1.3">
+
+```kotlin
+import kotlinx.coroutines.*
+import kotlinx.coroutines.flow.*
+
+fun log(msg: String) = println("[${Thread.currentThread().name}] $msg")
+
+//sampleStart
+fun foo(): Flow<Int> = flow {
+ for (i in 1..3) {
+ Thread.sleep(100) // pretend we are computing it in CPU-consuming way
+ log("Emitting $i")
+ emit(i) // emit next value
+ }
+}.flowOn(Dispatchers.Default) // RIGHT way to change context for CPU-consuming code in flow builder
+
+fun main() = runBlocking<Unit> {
+ foo().collect { value ->
+ log("Collected $value")
+ }
+}
+//sampleEnd
+```
+
+</div>
+
+> You can get full code [here](../kotlinx-coroutines-core/jvm/test/guide/example-flow-15.kt).
+
+Notice how `flow { ... }` works in the background thread, while collection happens in the main thread:
+
+<!--- TEST FLEXIBLE_THREAD
+[DefaultDispatcher-worker-1 @coroutine#2] Emitting 1
+[main @coroutine#1] Collected 1
+[DefaultDispatcher-worker-1 @coroutine#2] Emitting 2
+[main @coroutine#1] Collected 2
+[DefaultDispatcher-worker-1 @coroutine#2] Emitting 3
+[main @coroutine#1] Collected 3
+-->
+
+Another observation here is that [flowOn] operator had 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 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 overall time it takes
+to collect the flow, especially when long-running asynchronous operations are involved. For example, consider a case when
+emission by `foo()` flow is slow, taking 100 ms to produce an element; and collector is also slow,
+taking 300 ms to process an element. Let us see how long does it take to collect such a flow with three numbers:
+
+<div class="sample" markdown="1" theme="idea" data-min-compiler-version="1.3">
+
+```kotlin
+import kotlinx.coroutines.*
+import kotlinx.coroutines.flow.*
+import kotlin.system.*
+
+//sampleStart
+fun foo(): Flow<Int> = flow {
+ for (i in 1..3) {
+ delay(100) // pretend we are asynchronously waiting 100 ms
+ emit(i) // emit next value
+ }
+}
+
+fun main() = runBlocking<Unit> {
+ val time = measureTimeMillis {
+ foo().collect { value ->
+ delay(300) // pretend we are processing it for 300 ms
+ println(value)
+ }
+ }
+ println("Collected in $time ms")
+}
+//sampleEnd
+```
+
+</div>
+
+> You can get full code [here](../kotlinx-coroutines-core/jvm/test/guide/example-flow-16.kt).
+
+It produces something like this, the whole collection taking around 1200 ms (three numbers times 400 ms each):
+
+```text
+1
+2
+3
+Collected in 1220 ms
+```
+
+<!--- TEST ARBITRARY_TIME -->
+
+We can use [buffer] operator on a flow to run emitting code of `foo()` concurrently with collecting code,
+as opposed to running them sequentially:
+
+<div class="sample" markdown="1" theme="idea" data-min-compiler-version="1.3">
+
+```kotlin
+import kotlinx.coroutines.*
+import kotlinx.coroutines.flow.*
+import kotlin.system.*
+
+fun foo(): Flow<Int> = flow {
+ for (i in 1..3) {
+ delay(100) // pretend we are asynchronously waiting 100 ms
+ emit(i) // emit next value
+ }
+}
+
+fun main() = runBlocking<Unit> {
+//sampleStart
+ val time = measureTimeMillis {
+ foo()
+ .buffer() // buffer emissions, don't wait
+ .collect { value ->
+ delay(300) // pretend we are processing it for 300 ms
+ println(value)
+ }
+ }
+ println("Collected in $time ms")
+//sampleEnd
+}
+```
+
+</div>
+
+> You can get full code [here](../kotlinx-coroutines-core/jvm/test/guide/example-flow-17.kt).
+
+It produces the same numbers faster, as we have effectively created a processing pipeline,
+only having to 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:
+
+```text
+1
+2
+3
+Collected in 1071 ms
+```
+
+<!--- TEST ARBITRARY_TIME -->
+
+> Note that [flowOn] operator uses the same buffering mechanism when it has to change [CoroutineDispatcher],
+but here we explicitly request buffering without changing execution context.
+
+#### Conflation
+
+When flow represents partial results of some operation or operation status updates, it may not be necessary
+to process each value, but only to process the most recent ones. In this case, [conflate] operator can be used to skip
+intermediate values when a collector is too slow to process them. Building on the previous example:
+
+<div class="sample" markdown="1" theme="idea" data-min-compiler-version="1.3">
+
+```kotlin
+import kotlinx.coroutines.*
+import kotlinx.coroutines.flow.*
+import kotlin.system.*
+
+fun foo(): Flow<Int> = flow {
+ for (i in 1..3) {
+ delay(100) // pretend we are asynchronously waiting 100 ms
+ emit(i) // emit next value
+ }
+}
+
+fun main() = runBlocking<Unit> {
+//sampleStart
+ val time = measureTimeMillis {
+ foo()
+ .conflate() // conflate emissions, don't process each one
+ .collect { value ->
+ delay(300) // pretend we are processing it for 300 ms
+ println(value)
+ }
+ }
+ println("Collected in $time ms")
+//sampleEnd
+}
+```
+
+</div>
+
+> You can get full code [here](../kotlinx-coroutines-core/jvm/test/guide/example-flow-18.kt).
+
+We see that while the first number was being processed the second and the third ones were already produced, so
+the second one was _conflated_ and only the most recent (the third one) was delivered to the collector:
+
+```text
+1
+3
+Collected in 758 ms
+```
+
+<!--- TEST ARBITRARY_TIME -->
+
+#### Processing the latest value
+
+Conflation is one way to speed up processing when both emitter and collector are slow. It does that by dropping emitted values.
+The other way is to cancel 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 `xxx` operator, but cancel the
+code in their block on a new value. Let us change the previous example from [conflate] to [collectLatest]:
+
+<div class="sample" markdown="1" theme="idea" data-min-compiler-version="1.3">
+
+```kotlin
+import kotlinx.coroutines.*
+import kotlinx.coroutines.flow.*
+import kotlin.system.*
+
+fun foo(): Flow<Int> = flow {
+ for (i in 1..3) {
+ delay(100) // pretend we are asynchronously waiting 100 ms
+ emit(i) // emit next value
+ }
+}
+
+fun main() = runBlocking<Unit> {
+//sampleStart
+ val time = measureTimeMillis {
+ foo()
+ .collectLatest { value -> // cancel & restart on the latest value
+ println("Collecting $value")
+ delay(300) // pretend we are processing it for 300 ms
+ println("Done $value")
+ }
+ }
+ println("Collected in $time ms")
+//sampleEnd
+}
+```
+
+</div>
+
+> You can get full code [here](../kotlinx-coroutines-core/jvm/test/guide/example-flow-19.kt).
+
+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:
+
+```text
+Collecting 1
+Collecting 2
+Collecting 3
+Done 3
+Collected in 741 ms
+```
+
+<!--- TEST ARBITRARY_TIME -->
+
+### Composing multiple flows
+
+There are several ways to compose multiple flows.
+
+#### Zip
+
+Similarly to [Sequence.zip] extension function in the Kotlin standard library,
+flows have [zip] operator that combines the corresponding values of two flows:
+
+<div class="sample" markdown="1" theme="idea" data-min-compiler-version="1.3">
+
+```kotlin
+import kotlinx.coroutines.*
+import kotlinx.coroutines.flow.*
+
+fun main() = runBlocking<Unit> {
+//sampleStart
+ val nums = (1..3).asFlow() // numbers 1..3
+ val strs = flowOf("one", "two", "three") // strings
+ nums.zip(strs) { a, b -> "$a -> $b" } // compose a single string
+ .collect { println(it) } // collect and print
+//sampleEnd
+}
+```
+
+</div>
+
+> You can get full code [here](../kotlinx-coroutines-core/jvm/test/guide/example-flow-20.kt).
+
+This example prints:
+
+```text
+1 -> one
+2 -> two
+3 -> three
+```
+
+<!--- TEST -->
+
+#### Combine
+
+When flow represents the most recent value of some variable or operation (see also a related
+section on [conflation](#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 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 [zip] operator would still produce the same result,
+albeit results are going to be printed every 400 ms:
+
+> We use [onEach] intermediate operator in this example to delay each element and thus make the code
+that emits sample flows more declarative and shorter.
+
+<div class="sample" markdown="1" theme="idea" data-min-compiler-version="1.3">
+
+```kotlin
+import kotlinx.coroutines.*
+import kotlinx.coroutines.flow.*
+
+fun main() = runBlocking<Unit> {
+//sampleStart
+ val nums = (1..3).asFlow().onEach { delay(300) } // numbers 1..3 every 300 ms
+ val strs = flowOf("one", "two", "three").onEach { delay(400) } // strings every 400 ms
+ val startTime = System.currentTimeMillis() // remember the start time
+ nums.zip(strs) { a, b -> "$a -> $b" } // compose a single string with "zip"
+ .collect { value -> // collect and print
+ println("$value at ${System.currentTimeMillis() - startTime} ms from start")
+ }
+//sampleEnd
+}
+```
+
+</div>
+
+> You can get full code [here](../kotlinx-coroutines-core/jvm/test/guide/example-flow-21.kt).
+
+<!--- TEST ARBITRARY_TIME
+1 -> one at 437 ms from start
+2 -> two at 837 ms from start
+3 -> three at 1243 ms from start
+-->
+
+However, using [combine] operator here instead of [zip]:
+
+<div class="sample" markdown="1" theme="idea" data-min-compiler-version="1.3">
+
+```kotlin
+import kotlinx.coroutines.*
+import kotlinx.coroutines.flow.*
+
+fun main() = runBlocking<Unit> {
+//sampleStart
+ val nums = (1..3).asFlow().onEach { delay(300) } // numbers 1..3 every 300 ms
+ val strs = flowOf("one", "two", "three").onEach { delay(400) } // strings every 400 ms
+ val startTime = currentTimeMillis() // remember the start time
+ nums.combine(strs) { a, b -> "$a -> $b" } // compose a single string with "combine"
+ .collect { value -> // collect and print
+ println("$value at ${System.currentTimeMillis() - startTime} ms from start")
+ }
+//sampleEnd
+}
+```
+
+</div>
+
+> You can get full code [here](../kotlinx-coroutines-core/jvm/test/guide/example-flow-22.kt).
+
+We get quite a different output, where a line is printed at each emission from either `nums` or `strs` flows:
+
+```text
+1 -> one at 452 ms from start
+2 -> one at 651 ms from start
+2 -> two at 854 ms from start
+3 -> two at 952 ms from start
+3 -> three at 1256 ms from start
+```
+
+<!--- TEST ARBITRARY_TIME -->
+
+### 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:
+
+<div class="sample" markdown="1" theme="idea" data-highlight-only>
+
+```kotlin
+fun requestFlow(i: Int): Flow<String> = flow {
+ emit("$i: First")
+ delay(500) // wait 500 ms
+ emit("$i: Second")
+}
+```
+
+</div>
+
+<!--- CLEAR -->
+
+Now if we have a flow of three integers and call `requestFlow` for each of them like this:
+
+<div class="sample" markdown="1" theme="idea" data-highlight-only>
+
+```kotlin
+(1..3).asFlow().map { requestFlow(it) }
+```
+
+</div>
+
+<!--- CLEAR -->
+
+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][Sequence.flatten] and [flatMap][Sequence.flatMap]
+operators for this purpose. However, the asynchronous nature of flows calls for different _modes_ of flattening
+thus 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 inner flow to complete before
+starting to collect the next one as the following example shows:
+
+<div class="sample" markdown="1" theme="idea" data-min-compiler-version="1.3">
+
+```kotlin
+import kotlinx.coroutines.*
+import kotlinx.coroutines.flow.*
+
+fun requestFlow(i: Int): Flow<String> = flow {
+ emit("$i: First")
+ delay(500) // wait 500 ms
+ emit("$i: Second")
+}
+
+fun main() = runBlocking<Unit> {
+//sampleStart
+ val startTime = currentTimeMillis() // remember the start time
+ (1..3).asFlow().onEach { delay(100) } // a number every 100 ms
+ .flatMapConcat { requestFlow(it) }
+ .collect { value -> // collect and print
+ println("$value at ${System.currentTimeMillis() - startTime} ms from start")
+ }
+//sampleEnd
+}
+```
+
+</div>
+
+> You can get full code [here](../kotlinx-coroutines-core/jvm/test/guide/example-flow-23.kt).
+
+The sequential nature of [flatMapConcat] is clearly seen in the output:
+
+```text
+1: First at 121 ms from start
+1: Second at 622 ms from start
+2: First at 727 ms from start
+2: Second at 1227 ms from start
+3: First at 1328 ms from start
+3: Second at 1829 ms from start
+```
+
+<!--- TEST ARBITRARY_TIME -->
+
+#### 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).
+
+<div class="sample" markdown="1" theme="idea" data-min-compiler-version="1.3">
+
+```kotlin
+import kotlinx.coroutines.*
+import kotlinx.coroutines.flow.*
+
+fun requestFlow(i: Int): Flow<String> = flow {
+ emit("$i: First")
+ delay(500) // wait 500 ms
+ emit("$i: Second")
+}
+
+fun main() = runBlocking<Unit> {
+//sampleStart
+ val startTime = currentTimeMillis() // remember the start time
+ (1..3).asFlow().onEach { delay(100) } // a number every 100 ms
+ .flatMapMerge { requestFlow(it) }
+ .collect { value -> // collect and print
+ println("$value at ${System.currentTimeMillis() - startTime} ms from start")
+ }
+//sampleEnd
+}
+```
+
+</div>
+
+> You can get full code [here](../kotlinx-coroutines-core/jvm/test/guide/example-flow-24.kt).
+
+The concurrent nature of [flatMapMerge] is obvious:
+
+```text
+1: First at 136 ms from start
+2: First at 231 ms from start
+3: First at 333 ms from start
+1: Second at 639 ms from start
+2: Second at 732 ms from start
+3: Second at 833 ms from start
+```
+
+<!--- TEST ARBITRARY_TIME -->
+
+> Note that [flatMapMerge] call its block of code (`{ requestFlow(it) }` in this example) sequentially, but
+collects the resulting flows concurrently, so it is equivalent to performing a sequential
+`map { requestFlow(it) }` first and then calling [flattenMerge] on the result.
+
+#### flatMapLatest
+
+In a similar way to [collectLatest] operator that was shown in
+["Processing the latest value"](#processing-the-latest-value) section, there is the corresponding "Latest"
+flattening mode where collection of the previous flow is cancelled as soon as new flow is emitted.
+It is implemented by [flatMapLatest] operator.
+
+<div class="sample" markdown="1" theme="idea" data-min-compiler-version="1.3">
+
+```kotlin
+import kotlinx.coroutines.*
+import kotlinx.coroutines.flow.*
+
+fun requestFlow(i: Int): Flow<String> = flow {
+ emit("$i: First")
+ delay(500) // wait 500 ms
+ emit("$i: Second")
+}
+
+fun main() = runBlocking<Unit> {
+//sampleStart
+ val startTime = currentTimeMillis() // remember the start time
+ (1..3).asFlow().onEach { delay(100) } // a number every 100 ms
+ .flatMapLatest { requestFlow(it) }
+ .collect { value -> // collect and print
+ println("$value at ${System.currentTimeMillis() - startTime} ms from start")
+ }
+//sampleEnd
+}
+```
+
+</div>
+
+> You can get full code [here](../kotlinx-coroutines-core/jvm/test/guide/example-flow-25.kt).
+
+The output of this example speaks for the way [flatMapLatest] works:
+
+```text
+1: First at 142 ms from start
+2: First at 322 ms from start
+3: First at 425 ms from start
+3: Second at 931 ms from start
+```
+
+<!--- TEST ARBITRARY_TIME -->
+
+> Note that [flatMapLatest] cancels all the code in its block (`{ requestFlow(it) }` in this example) on a new value.
+It makes no difference in this particular example, because the call to `requestFlow` itself is fast, not-suspending,
+and cannot be cancelled. However, it would show up if we were to use suspending functions like `delay` in there.
+
+### Flow exceptions
+
+Flow collection can complete with an exception when emitter or any code inside any of 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`][exceptions] block to handle exceptions:
+
+<div class="sample" markdown="1" theme="idea" data-min-compiler-version="1.3">
+
+```kotlin
+import kotlinx.coroutines.*
+import kotlinx.coroutines.flow.*
+
+//sampleStart
+fun foo(): Flow<Int> = flow {
+ for (i in 1..3) {
+ println("Emitting $i")
+ emit(i) // emit next value
+ }
+}
+
+fun main() = runBlocking<Unit> {
+ try {
+ foo().collect { value ->
+ println(value)
+ check(value <= 1) { "Collected $value" }
+ }
+ } catch (e: Throwable) {
+ println("Caught $e")
+ }
+}
+//sampleEnd
+```
+
+</div>
+
+> You can get full code [here](../kotlinx-coroutines-core/jvm/test/guide/example-flow-26.kt).
+
+This code successfully catches an exception in [collect] terminal operator and,
+as you can see, no more values are emitted after that:
+
+```text
+Emitting 1
+1
+Emitting 2
+2
+Caught java.lang.IllegalStateException: Collected 2
+```
+
+<!--- TEST -->
+
+#### Everything is caught
+
+The previous example actually catches any exception happening in emitter or in any intermediate or terminal operators.
+For example, let us change the code so that emitted values are [mapped][map] to strings,
+but the corresponding code produces an exception:
+
+<div class="sample" markdown="1" theme="idea" data-min-compiler-version="1.3">
+
+```kotlin
+import kotlinx.coroutines.*
+import kotlinx.coroutines.flow.*
+
+//sampleStart
+fun foo(): Flow<String> =
+ flow {
+ for (i in 1..3) {
+ println("Emitting $i")
+ emit(i) // emit next value
+ }
+ }
+ .map { value ->
+ check(value <= 1) { "Crashed on $value" }
+ "string $value"
+ }
+
+fun main() = runBlocking<Unit> {
+ try {
+ foo().collect { value -> println(value) }
+ } catch (e: Throwable) {
+ println("Caught $e")
+ }
+}
+//sampleEnd
+```
+
+</div>
+
+> You can get full code [here](../kotlinx-coroutines-core/jvm/test/guide/example-flow-27.kt).
+
+This exception is still caught and collection is stopped:
+
+```text
+Emitting 1
+string 1
+Emitting 2
+Caught java.lang.IllegalStateException: Crashed on 2
+```
+
+<!--- TEST -->
+
+### Exception transparency
+
+But how can code of emitter encapsulate its exception handling behavior?
+
+Flows must be _transparent to exceptions_ and it is a violation of exception transparency to [emit][FlowCollector.emit] values in the
+`flow { ... }` builder from inside of `try/catch` block. This guarantees that a collector throwing an exception
+can always catch it using `try/catch` as in the previous example.
+
+The emitter can use [catch] operator that preserves this exception transparency and allows encapsulation
+of its exception handling. The body of the `catch` operator can analyze an exception
+and react to it in different ways depending on which exception was caught:
+
+* Exceptions can be rethrown using `throw`.
+* Exceptions can be turned into emission of values using [emit][FlowCollector.emit] from the body of [catch].
+* Exceptions can be ignored, logged, or processed by some other code.
+
+For example, let us emit a text on catching an exception:
+
+<div class="sample" markdown="1" theme="idea" data-min-compiler-version="1.3">
+
+```kotlin
+import kotlinx.coroutines.*
+import kotlinx.coroutines.flow.*
+
+fun foo(): Flow<String> =
+ flow {
+ for (i in 1..3) {
+ println("Emitting $i")
+ emit(i) // emit next value
+ }
+ }
+ .map { value ->
+ check(value <= 1) { "Crashed on $value" }
+ "string $value"
+ }
+
+fun main() = runBlocking<Unit> {
+//sampleStart
+ foo()
+ .catch { e -> emit("Caught $e") } // emit on exception
+ .collect { value -> println(value) }
+//sampleEnd
+}
+```
+
+</div>
+
+> You can get full code [here](../kotlinx-coroutines-core/jvm/test/guide/example-flow-28.kt).
+
+The output of the example is the same, even though we do not have `try/catch` around the code anymore.
+
+<!--- TEST
+Emitting 1
+string 1
+Emitting 2
+Caught java.lang.IllegalStateException: Crashed on 2
+-->
+
+#### Transparent catch
+
+The [catch] intermediate operator, honoring exception transparency, catches only upstream exceptions
+(that is an exception from all the operators above `catch`, but not below it).
+If the block in `collect { ... }` (placed below `catch`) throws an exception then it escapes:
+
+<div class="sample" markdown="1" theme="idea" data-min-compiler-version="1.3">
+
+```kotlin
+import kotlinx.coroutines.*
+import kotlinx.coroutines.flow.*
+
+//sampleStart
+fun foo(): Flow<Int> = flow {
+ for (i in 1..3) {
+ println("Emitting $i")
+ emit(i)
+ }
+}
+
+fun main() = runBlocking<Unit> {
+ foo()
+ .catch { e -> println("Caught $e") } // does not catch downstream exceptions
+ .collect { value ->
+ check(value <= 1) { "Collected $value" }
+ println(value)
+ }
+}
+//sampleEnd
+```
+
+</div>
+
+> You can get full code [here](../kotlinx-coroutines-core/jvm/test/guide/example-flow-29.kt).
+
+The "Caught ..." message is not printed despite the `catch` operator:
+
+<!--- TEST EXCEPTION
+Emitting 1
+1
+Emitting 2
+Exception in thread "main" java.lang.IllegalStateException: Collected 2
+ at ...
+-->
+
+#### Catching declaratively
+
+We can combine a declarative nature of [catch] operator with a desire to handle all exceptions by moving the body
+of [collect] operator into [onEach] and putting it before the `catch` operator. Collection of this flow must
+be triggered by a call to `collect()` without parameters:
+
+<div class="sample" markdown="1" theme="idea" data-min-compiler-version="1.3">
+
+```kotlin
+import kotlinx.coroutines.*
+import kotlinx.coroutines.flow.*
+
+fun foo(): Flow<Int> = flow {
+ for (i in 1..3) {
+ println("Emitting $i")
+ emit(i)
+ }
+}
+
+fun main() = runBlocking<Unit> {
+//sampleStart
+ foo()
+ .onEach { value ->
+ check(value <= 1) { "Collected $value" }
+ println(value)
+ }
+ .catch { e -> println("Caught $e") }
+ .collect()
+//sampleEnd
+}
+```
+
+</div>
+
+> You can get full code [here](../kotlinx-coroutines-core/jvm/test/guide/example-flow-30.kt).
+
+Now we can see that "Caught ..." message is printed and thus we can catch all exceptions without explicitly
+using a `try/catch` block:
+
+<!--- TEST EXCEPTION
+Emitting 1
+1
+Emitting 2
+Caught java.lang.IllegalStateException: Collected 2
+-->
+
+### Flow completion
+
+When flow collection completes (normally or exceptionally) it may be needed to execute some action.
+As you might have already noticed, it also can be done in two ways: imperative and declarative.
+
+#### Imperative finally block
+
+In addition to `try`/`catch`, a collector can also use `finally` block to execute an action
+upon `collect` completion.
+
+<div class="sample" markdown="1" theme="idea" data-min-compiler-version="1.3">
+
+```kotlin
+import kotlinx.coroutines.*
+import kotlinx.coroutines.flow.*
+
+//sampleStart
+fun foo(): Flow<Int> = (1..3).asFlow()
+
+fun main() = runBlocking<Unit> {
+ try {
+ foo().collect { value -> println(value) }
+ } finally {
+ println("Done")
+ }
+}
+//sampleEnd
+```
+
+</div>
+
+> You can get full code [here](../kotlinx-coroutines-core/jvm/test/guide/example-flow-31.kt).
+
+This code prints three numbers produced by the `foo()` flow followed by "Done" string:
+
+```text
+1
+2
+3
+Done
+```
+
+<!--- TEST -->
+
+#### Declarative handling
+
+For declarative approach, flow has [onCompletion] intermediate operator that is invoked
+when the flow is completely collected.
+
+The previous example can be rewritten using [onCompletion] operator and produces the same output:
+
+<div class="sample" markdown="1" theme="idea" data-min-compiler-version="1.3">
+
+```kotlin
+import kotlinx.coroutines.*
+import kotlinx.coroutines.flow.*
+
+fun foo(): Flow<Int> = (1..3).asFlow()
+
+fun main() = runBlocking<Unit> {
+//sampleStart
+ foo()
+ .onCompletion { println("Done") }
+ .collect { value -> println(value) }
+//sampleEnd
+}
+```
+</div>
+
+> You can get full code [here](../kotlinx-coroutines-core/jvm/test/guide/example-flow-32.kt).
+
+<!--- TEST
+1
+2
+3
+Done
+-->
+
+The key advantage of [onCompletion] is a nullable `Throwable` parameter of the lambda that can be used
+to determine whether flow collection was completed normally or exceptionally. In the following
+example `foo()` flow throws exception after emitting number 1:
+
+<div class="sample" markdown="1" theme="idea" data-min-compiler-version="1.3">
+
+```kotlin
+import kotlinx.coroutines.*
+import kotlinx.coroutines.flow.*
+
+//sampleStart
+fun foo(): Flow<Int> = flow {
+ emit(1)
+ throw RuntimeException()
+}
+
+fun main() = runBlocking<Unit> {
+ foo()
+ .onCompletion { cause -> if (cause != null) println("Flow completed exceptionally") }
+ .catch { cause -> println("Caught exception") }
+ .collect { value -> println(value) }
+}
+//sampleEnd
+```
+</div>
+
+> You can get full code [here](../kotlinx-coroutines-core/jvm/test/guide/example-flow-33.kt).
+
+As you may expect, it prints:
+
+```text
+1
+Flow completed exceptionally
+Caught exception
+```
+
+<!--- TEST -->
+
+[onCompletion] operator, unlike [catch], does not handle the exception. As we can see from the above
+example code, the exception still flows downstream. It will be delivered to further `onCompletion` operators
+and can be handled with `catch` operator.
+
+#### Upstream exceptions only
+
+Just like [catch] operator, [onCompletion] sees only exception coming from upstream and does not
+see downstream exceptions. For example, run the following code:
+
+<div class="sample" markdown="1" theme="idea" data-min-compiler-version="1.3">
+
+```kotlin
+import kotlinx.coroutines.*
+import kotlinx.coroutines.flow.*
+
+//sampleStart
+fun foo(): Flow<Int> = (1..3).asFlow()
+
+fun main() = runBlocking<Unit> {
+ foo()
+ .onCompletion { cause -> println("Flow completed with $cause") }
+ .collect { value ->
+ check(value <= 1) { "Collected $value" }
+ println(value)
+ }
+}
+//sampleEnd
+```
+
+</div>
+
+> You can get full code [here](../kotlinx-coroutines-core/jvm/test/guide/example-flow-34.kt).
+
+And you can see the completion cause is null, yet collection failed with exception:
+
+```text
+1
+Flow completed with null
+Exception in thread "main" java.lang.IllegalStateException: Collected 2
+```
+
+<!--- TEST EXCEPTION -->
+
+### Imperative versus declarative
+
+Now we know how to collect flow, handle its completion and exceptions in both imperative and declarative ways.
+The natural question here is which approach should be preferred and why.
+As a library, we do not advocate for any particular approach and believe that both options
+are valid and should be selected according to your own preferences and code style.
+
+### Launching flow
+
+It is convenient to use flows to represent asynchronous events that are coming from some source.
+In this case, we need an analogue of `addEventListener` function that registers a piece of code with a reaction
+on incoming events and continues further work. The [onEach] operator can serve this role.
+However, `onEach` is an intermediate operator. We also need a terminal operator to collect the flow.
+Otherwise, just calling `onEach` has no effect.
+
+If we use [collect] terminal operator after `onEach`, then code after it waits until the flow is collected:
+
+<div class="sample" markdown="1" theme="idea" data-min-compiler-version="1.3">
+
+```kotlin
+import kotlinx.coroutines.*
+import kotlinx.coroutines.flow.*
+
+//sampleStart
+// Imitate a flow of events
+fun events(): Flow<Int> = (1..3).asFlow().onEach { delay(100) }
+
+fun main() = runBlocking<Unit> {
+ events()
+ .onEach { event -> println("Event: $event") }
+ .collect() // <--- Collecting the flow waits
+ println("Done")
+}
+//sampleEnd
+```
+
+</div>
+
+> You can get full code [here](../kotlinx-coroutines-core/jvm/test/guide/example-flow-35.kt).
+
+As you can see, it prints:
+
+```text
+Event: 1
+Event: 2
+Event: 3
+Done
+```
+
+<!--- TEST -->
+
+Here [launchIn] terminal operator comes in handy. Replacing `collect` with `launchIn` we can
+launch collection of the flow in a separate coroutine, so that execution of further code
+immediately continues:
+
+<div class="sample" markdown="1" theme="idea" data-min-compiler-version="1.3">
+
+```kotlin
+import kotlinx.coroutines.*
+import kotlinx.coroutines.flow.*
+
+// Imitate a flow of events
+fun events(): Flow<Int> = (1..3).asFlow().onEach { delay(100) }
+
+//sampleStart
+fun main() = runBlocking<Unit> {
+ events()
+ .onEach { event -> println("Event: $event") }
+ .launchIn(this) // <--- Launching the flow in a separate coroutine
+ println("Done")
+}
+//sampleEnd
+```
+
+</div>
+
+> You can get full code [here](../kotlinx-coroutines-core/jvm/test/guide/example-flow-36.kt).
+
+It prints:
+
+```text
+Done
+Event: 1
+Event: 2
+Event: 3
+```
+
+<!--- TEST -->
+
+The required parameter to `launchIn` must specify a [CoroutineScope] in which the coroutine to collect the flow is
+launched. In the above example this scope comes from [runBlocking]
+coroutine builder, so while the flow is running this [runBlocking] scope waits for completion of its child coroutine
+and keeps the main function from returning and terminating this example.
+
+In real applications a scope is going to come from some entity with a limited
+lifetime. As soon as the lifetime of this entity is terminated the corresponding scope is cancelled, cancelling
+collection of the corresponding flow. This way the pair of `onEach { ... }.launchIn(scope)` works
+like `addEventListener`. However, there is no need for the corresponding `removeEventListener` function,
+as cancellation and structured concurrency serve this purpose.
+
+Note, that [launchIn] also returns a [Job] which can be used to [cancel][Job.cancel] the corresponding flow collection
+coroutine only without cancelling the whole scope or to [join][Job.join] it.
+
+<!-- stdlib references -->
+
+[collections]: https://kotlinlang.org/docs/reference/collections-overview.html
+[List]: https://kotlinlang.org/api/latest/jvm/stdlib/kotlin.collections/-list/index.html
+[forEach]: https://kotlinlang.org/api/latest/jvm/stdlib/kotlin.collections/for-each.html
+[Sequence]: https://kotlinlang.org/api/latest/jvm/stdlib/kotlin.sequences/index.html
+[Sequence.zip]: https://kotlinlang.org/api/latest/jvm/stdlib/kotlin.sequences/zip.html
+[Sequence.flatten]: https://kotlinlang.org/api/latest/jvm/stdlib/kotlin.sequences/flatten.html
+[Sequence.flatMap]: https://kotlinlang.org/api/latest/jvm/stdlib/kotlin.sequences/flat-map.html
+[exceptions]: https://kotlinlang.org/docs/reference/exceptions.html
+
+<!--- MODULE kotlinx-coroutines-core -->
+<!--- INDEX kotlinx.coroutines -->
+[delay]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines/delay.html
+[withTimeoutOrNull]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines/with-timeout-or-null.html
+[Dispatchers.Default]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines/-dispatchers/-default.html
+[Dispatchers.Main]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines/-dispatchers/-main.html
+[withContext]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines/with-context.html
+[CoroutineDispatcher]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines/-coroutine-dispatcher/index.html
+[CoroutineScope]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines/-coroutine-scope/index.html
+[runBlocking]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines/run-blocking.html
+[Job]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines/-job/index.html
+[Job.cancel]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines/-job/cancel.html
+[Job.join]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines/-job/join.html
+<!--- INDEX kotlinx.coroutines.flow -->
+[Flow]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.flow/-flow/index.html
+[flow]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.flow/flow.html
+[FlowCollector.emit]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.flow/-flow-collector/emit.html
+[collect]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.flow/collect.html
+[flowOf]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.flow/flow-of.html
+[map]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.flow/map.html
+[filter]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.flow/filter.html
+[transform]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.flow/transform.html
+[take]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.flow/take.html
+[toList]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.flow/to-list.html
+[toSet]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.flow/to-set.html
+[first]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.flow/first.html
+[single]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.flow/single.html
+[reduce]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.flow/reduce.html
+[fold]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.flow/fold.html
+[flowOn]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.flow/flow-on.html
+[buffer]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.flow/buffer.html
+[conflate]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.flow/conflate.html
+[collectLatest]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.flow/collect-latest.html
+[zip]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.flow/zip.html
+[combine]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.flow/combine.html
+[onEach]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.flow/on-each.html
+[flatMapConcat]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.flow/flat-map-concat.html
+[flattenConcat]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.flow/flatten-concat.html
+[flatMapMerge]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.flow/flat-map-merge.html
+[flattenMerge]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.flow/flatten-merge.html
+[DEFAULT_CONCURRENCY]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.flow/-d-e-f-a-u-l-t_-c-o-n-c-u-r-r-e-n-c-y.html
+[flatMapLatest]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.flow/flat-map-latest.html
+[catch]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.flow/catch.html
+[onCompletion]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.flow/on-completion.html
+[launchIn]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.flow/launch-in.html
+<!--- END -->
diff --git a/docs/select-expression.md b/docs/select-expression.md
index 35480ab..f36fa09 100644
--- a/docs/select-expression.md
+++ b/docs/select-expression.md
@@ -21,7 +21,7 @@
<!--- TOC -->
-* [Select expression (experimental)](#select-expression-experimental)
+* [Select Expression (experimental)](#select-expression-experimental)
* [Selecting from channels](#selecting-from-channels)
* [Selecting on close](#selecting-on-close)
* [Selecting to send](#selecting-to-send)
@@ -32,7 +32,7 @@
-## Select expression (experimental)
+## Select Expression (experimental)
Select expression makes it possible to await multiple suspending functions simultaneously and _select_
the first one that becomes available.