{
-> You can get full code [here](../kotlinx-coroutines-core/jvm/test/guide/example-context-09.kt).
+> You can get the full code [here](../kotlinx-coroutines-core/jvm/test/guide/example-context-09.kt).
The output of this code with the `-Dkotlinx.coroutines.debug` JVM option is:
@@ -501,21 +529,8 @@ class Activity {
-Alternatively, we can implement the [CoroutineScope] interface in this `Activity` class. The best way to do it is
-to use delegation with default factory functions.
-We also can combine the desired dispatcher (we used [Dispatchers.Default] in this example) with the scope:
-
-
-
-```kotlin
- class Activity : CoroutineScope by CoroutineScope(Dispatchers.Default) {
- // to be continued ...
-```
-
-
-
-Now, we can launch coroutines in the scope of this `Activity` without having to explicitly
-specify their context. For the demo, we launch ten coroutines that delay for a different time:
+Now, we can launch coroutines in the scope of this `Activity` using the defined `scope`.
+For the demo, we launch ten coroutines that delay for a different time:
@@ -524,7 +539,7 @@ specify their context. For the demo, we launch ten coroutines that delay for a d
fun doSomething() {
// launch ten coroutines for a demo, each working for a different time
repeat(10) { i ->
- launch {
+ mainScope.launch {
delay((i + 1) * 200L) // variable delay 200ms, 400ms, ... etc
println("Coroutine $i is done")
}
@@ -544,21 +559,19 @@ of the activity no more messages are printed, even if we wait a little longer.
```kotlin
-import kotlin.coroutines.*
import kotlinx.coroutines.*
-class Activity : CoroutineScope by CoroutineScope(Dispatchers.Default) {
-
+class Activity {
+ private val mainScope = CoroutineScope(Dispatchers.Default) // use Default for test purposes
+
fun destroy() {
- cancel() // Extension on CoroutineScope
+ mainScope.cancel()
}
- // to be continued ...
- // class Activity continues
fun doSomething() {
// launch ten coroutines for a demo, each working for a different time
repeat(10) { i ->
- launch {
+ mainScope.launch {
delay((i + 1) * 200L) // variable delay 200ms, 400ms, ... etc
println("Coroutine $i is done")
}
@@ -581,7 +594,7 @@ fun main() = runBlocking {
-> You can get full code [here](../kotlinx-coroutines-core/jvm/test/guide/example-context-10.kt).
+> You can get the full code [here](../kotlinx-coroutines-core/jvm/test/guide/example-context-10.kt).
The output of this example is:
@@ -597,6 +610,9 @@ Destroying activity!
As you can see, only the first two coroutines print a message and the others are cancelled
by a single invocation of `job.cancel()` in `Activity.destroy()`.
+> Note, that Android has first-party support for coroutine scope in all entities with the lifecycle.
+See [the corresponding documentation](https://developer.android.com/topic/libraries/architecture/coroutines#lifecyclescope).
+
### Thread-local data
Sometimes it is convenient to have an ability to pass some thread-local data to or between coroutines.
@@ -632,12 +648,12 @@ fun main() = runBlocking
{
@@ -41,13 +41,13 @@ It can be demonstrated by a simple example that creates coroutines in the [Globa
import kotlinx.coroutines.*
fun main() = runBlocking {
- val job = GlobalScope.launch {
+ val job = GlobalScope.launch { // root coroutine with launch
println("Throwing exception from launch")
throw IndexOutOfBoundsException() // Will be printed to the console by Thread.defaultUncaughtExceptionHandler
}
job.join()
println("Joined failed job")
- val deferred = GlobalScope.async {
+ val deferred = GlobalScope.async { // root coroutine with async
println("Throwing exception from async")
throw ArithmeticException() // Nothing is printed, relying on user to call await
}
@@ -62,7 +62,7 @@ fun main() = runBlocking {
-> You can get full code [here](../kotlinx-coroutines-core/jvm/test/guide/example-exceptions-01.kt).
+> You can get the full code [here](../kotlinx-coroutines-core/jvm/test/guide/example-exceptions-01.kt).
The output of this code is (with [debug](https://github.com/Kotlin/kotlinx.coroutines/blob/master/docs/coroutine-context-and-dispatchers.md#debugging-coroutines-and-threads)):
@@ -78,9 +78,13 @@ Caught ArithmeticException
### CoroutineExceptionHandler
-But what if one does not want to print all exceptions to the console?
-[CoroutineExceptionHandler] context element is used as generic `catch` block of coroutine where custom logging or exception handling may take place.
-It is similar to using [`Thread.uncaughtExceptionHandler`](https://docs.oracle.com/javase/8/docs/api/java/lang/Thread.html#setUncaughtExceptionHandler(java.lang.Thread.UncaughtExceptionHandler)).
+It is possible to customize the default behavior of printing **uncaught** exceptions to the console.
+[CoroutineExceptionHandler] context element on a _root_ coroutine can be used as generic `catch` block for
+this root coroutine and all its children where custom exception handling may take place.
+It is similar to [`Thread.uncaughtExceptionHandler`](https://docs.oracle.com/javase/8/docs/api/java/lang/Thread.html#setUncaughtExceptionHandler(java.lang.Thread.UncaughtExceptionHandler)).
+You cannot recover from the exception in the `CoroutineExceptionHandler`. The coroutine had already completed
+with the corresponding exception when the handler is called. Normally, the handler is used to
+log the exception, show some kind of error message, terminate, and/or restart the application.
On JVM it is possible to redefine global exception handler for all coroutines by registering [CoroutineExceptionHandler] via
[`ServiceLoader`](https://docs.oracle.com/javase/8/docs/api/java/util/ServiceLoader.html).
@@ -89,8 +93,15 @@ Global exception handler is similar to
which is used when no more specific handlers are registered.
On Android, `uncaughtExceptionPreHandler` is installed as a global coroutine exception handler.
-[CoroutineExceptionHandler] is invoked only on exceptions which are not expected to be handled by the user,
-so registering it in [async] builder and the like of it has no effect.
+`CoroutineExceptionHandler` is invoked only on **uncaught** exceptions — exceptions that were not handled in any other way.
+In particular, all _children_ coroutines (coroutines created in the context of another [Job]) delegate handling of
+their exceptions to their parent coroutine, which also delegates to the parent, and so on until the root,
+so the `CoroutineExceptionHandler` installed in their context is never used.
+In addition to that, [async] builder always catches all exceptions and represents them in the resulting [Deferred] object,
+so its `CoroutineExceptionHandler` has no effect either.
+
+> Coroutines running in supervision scope do not propagate exceptions to their parent and are
+excluded from this rule. A further [Supervision](#supervision) section of this document gives more details.
@@ -100,12 +111,12 @@ import kotlinx.coroutines.*
fun main() = runBlocking {
//sampleStart
val handler = CoroutineExceptionHandler { _, exception ->
- println("Caught $exception")
+ println("CoroutineExceptionHandler got $exception")
}
- val job = GlobalScope.launch(handler) {
+ val job = GlobalScope.launch(handler) { // root coroutine, running in GlobalScope
throw AssertionError()
}
- val deferred = GlobalScope.async(handler) {
+ val deferred = GlobalScope.async(handler) { // also root, but async instead of launch
throw ArithmeticException() // Nothing will be printed, relying on user to call deferred.await()
}
joinAll(job, deferred)
@@ -115,19 +126,19 @@ fun main() = runBlocking {
-> You can get full code [here](../kotlinx-coroutines-core/jvm/test/guide/example-exceptions-02.kt).
+> You can get the full code [here](../kotlinx-coroutines-core/jvm/test/guide/example-exceptions-02.kt).
The output of this code is:
```text
-Caught java.lang.AssertionError
+CoroutineExceptionHandler got java.lang.AssertionError
```
### Cancellation and exceptions
-Cancellation is tightly bound with exceptions. Coroutines internally use `CancellationException` for cancellation, these
+Cancellation is closely related to exceptions. Coroutines internally use `CancellationException` for cancellation, these
exceptions are ignored by all handlers, so they should be used only as the source of additional debug information, which can
be obtained by `catch` block.
When a coroutine is cancelled using [Job.cancel], it terminates, but it does not cancel its parent.
@@ -161,7 +172,7 @@ fun main() = runBlocking {
-> You can get full code [here](../kotlinx-coroutines-core/jvm/test/guide/example-exceptions-03.kt).
+> You can get the full code [here](../kotlinx-coroutines-core/jvm/test/guide/example-exceptions-03.kt).
The output of this code is:
@@ -175,15 +186,17 @@ Parent is not cancelled
If a coroutine encounters an exception other than `CancellationException`, it cancels its parent with that exception.
This behaviour cannot be overridden and is used to provide stable coroutines hierarchies for
-[structured concurrency](https://github.com/Kotlin/kotlinx.coroutines/blob/master/docs/composing-suspending-functions.md#structured-concurrency-with-async) which do not depend on
-[CoroutineExceptionHandler] implementation.
-The original exception is handled by the parent when all its children terminate.
+[structured concurrency](https://github.com/Kotlin/kotlinx.coroutines/blob/master/docs/composing-suspending-functions.md#structured-concurrency-with-async).
+[CoroutineExceptionHandler] implementation is not used for child coroutines.
-> This also a reason why, in these examples, [CoroutineExceptionHandler] is always installed to a coroutine
+> In these examples [CoroutineExceptionHandler] is always installed to a coroutine
that is created in [GlobalScope]. It does not make sense to install an exception handler to a coroutine that
is launched in the scope of the main [runBlocking], since the main coroutine is going to be always cancelled
when its child completes with exception despite the installed handler.
+The original exception is handled by the parent only when all its children terminate,
+which is demonstrated by the following example.
+
```kotlin
@@ -192,7 +205,7 @@ import kotlinx.coroutines.*
fun main() = runBlocking {
//sampleStart
val handler = CoroutineExceptionHandler { _, exception ->
- println("Caught $exception")
+ println("CoroutineExceptionHandler got $exception")
}
val job = GlobalScope.launch(handler) {
launch { // the first child
@@ -219,7 +232,7 @@ fun main() = runBlocking {
-> You can get full code [here](../kotlinx-coroutines-core/jvm/test/guide/example-exceptions-04.kt).
+> You can get the full code [here](../kotlinx-coroutines-core/jvm/test/guide/example-exceptions-04.kt).
The output of this code is:
@@ -227,22 +240,15 @@ The output of this code is:
Second child throws an exception
Children are cancelled, but exception is not handled until all children terminate
The first child finished its non cancellable block
-Caught java.lang.ArithmeticException
+CoroutineExceptionHandler got java.lang.ArithmeticException
```
### Exceptions aggregation
-What happens if multiple children of a coroutine throw an exception?
-The general rule is "the first exception wins", so the first thrown exception is exposed to the handler.
-But that may cause lost exceptions, for example if coroutine throws an exception in its `finally` block.
-So, additional exceptions are suppressed.
-
-> One of the solutions would have been to report each exception separately,
-but then [Deferred.await] should have had the same mechanism to avoid behavioural inconsistency and this
-would cause implementation details of a coroutines (whether it had delegated parts of its work to its children or not)
-to leak to its exception handler.
-
+When multiple children of a coroutine fail with an exception, the
+general rule is "the first exception wins", so the first exception gets handled.
+All additional exceptions that happen after the first one are attached to the first exception as suppressed ones.
-> Note, this mechanism currently works only on Java version 1.7+.
-Limitation on JS and Native is temporary and will be fixed in the future.
+> Note that this mechanism currently only works on Java version 1.7+.
+The JS and Native restrictions are temporary and will be lifted in the future.
-Cancellation exceptions are transparent and unwrapped by default:
+Cancellation exceptions are transparent and are unwrapped by default:
@@ -304,13 +310,13 @@ import java.io.*
fun main() = runBlocking {
//sampleStart
val handler = CoroutineExceptionHandler { _, exception ->
- println("Caught original $exception")
+ println("CoroutineExceptionHandler got $exception")
}
val job = GlobalScope.launch(handler) {
- val inner = launch {
+ val inner = launch { // all this stack of coroutines will get cancelled
launch {
launch {
- throw IOException()
+ throw IOException() // the original exception
}
}
}
@@ -318,7 +324,7 @@ fun main() = runBlocking {
inner.join()
} catch (e: CancellationException) {
println("Rethrowing CancellationException with original cause")
- throw e
+ throw e // cancellation exception is rethrown, yet the original IOException gets to the handler
}
}
job.join()
@@ -328,32 +334,33 @@ fun main() = runBlocking {
-> You can get full code [here](../kotlinx-coroutines-core/jvm/test/guide/example-exceptions-06.kt).
+> You can get the full code [here](../kotlinx-coroutines-core/jvm/test/guide/example-exceptions-06.kt).
The output of this code is:
```text
Rethrowing CancellationException with original cause
-Caught original java.io.IOException
+CoroutineExceptionHandler got java.io.IOException
```
### Supervision
As we have studied before, cancellation is a bidirectional relationship propagating through the whole
-coroutines hierarchy. But what if unidirectional cancellation is required?
+hierarchy of coroutines. Let us take a look at the case when unidirectional cancellation is required.
A good example of such a requirement is a UI component with the job defined in its scope. If any of the UI's child tasks
have failed, it is not always necessary to cancel (effectively kill) the whole UI component,
-but if UI component is destroyed (and its job is cancelled), then it is necessary to fail all child jobs as their results are no longer required.
+but if UI component is destroyed (and its job is cancelled), then it is necessary to fail all child jobs as their results are no longer needed.
-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.
+Another example is a server process that spawns multiple child jobs and needs to _supervise_
+their execution, tracking their failures and only restarting the failed ones.
#### 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:
+The [SupervisorJob][SupervisorJob()] can be used for these purposes.
+It is similar to a regular [Job][Job()] with the only exception that cancellation is propagated
+only downwards. This can easily be demonstrated using the following example:
@@ -365,24 +372,24 @@ fun main() = runBlocking {
with(CoroutineScope(coroutineContext + supervisor)) {
// launch the first child -- its exception is ignored for this example (don't do this in practice!)
val firstChild = launch(CoroutineExceptionHandler { _, _ -> }) {
- println("First child is failing")
- throw AssertionError("First child is cancelled")
+ println("The first child is failing")
+ throw AssertionError("The first child is cancelled")
}
// launch the second child
val secondChild = launch {
firstChild.join()
// Cancellation of the first child is not propagated to the second child
- println("First child is cancelled: ${firstChild.isCancelled}, but second one is still active")
+ println("The first child is cancelled: ${firstChild.isCancelled}, but the second one is still active")
try {
delay(Long.MAX_VALUE)
} finally {
// But cancellation of the supervisor is propagated
- println("Second child is cancelled because supervisor is cancelled")
+ println("The second child is cancelled because the supervisor was cancelled")
}
}
// wait until the first child fails & completes
firstChild.join()
- println("Cancelling supervisor")
+ println("Cancelling the supervisor")
supervisor.cancel()
secondChild.join()
}
@@ -391,24 +398,24 @@ fun main() = runBlocking {
-> You can get full code [here](../kotlinx-coroutines-core/jvm/test/guide/example-supervision-01.kt).
+> You can get the full code [here](../kotlinx-coroutines-core/jvm/test/guide/example-supervision-01.kt).
The output of this code is:
```text
-First child is failing
-First child is cancelled: true, but second one is still active
-Cancelling supervisor
-Second child is cancelled because supervisor is cancelled
+The first child is failing
+The first child is cancelled: true, but the second one is still active
+Cancelling the supervisor
+The second child is cancelled because the supervisor was cancelled
```
#### 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
-just like [coroutineScope] does.
+Instead of [coroutineScope][_coroutineScope], we can use [supervisorScope][_supervisorScope] for _scoped_ concurrency. It propagates the cancellation
+in one direction only and cancels all its children only if it failed itself. It also waits for all children before completion
+just like [coroutineScope][_coroutineScope] does.
@@ -421,42 +428,45 @@ fun main() = runBlocking {
supervisorScope {
val child = launch {
try {
- println("Child is sleeping")
+ println("The child is sleeping")
delay(Long.MAX_VALUE)
} finally {
- println("Child is cancelled")
+ println("The child is cancelled")
}
}
// Give our child a chance to execute and print using yield
yield()
- println("Throwing exception from scope")
+ println("Throwing an exception from the scope")
throw AssertionError()
}
} catch(e: AssertionError) {
- println("Caught assertion error")
+ println("Caught an assertion error")
}
}
```
-> You can get full code [here](../kotlinx-coroutines-core/jvm/test/guide/example-supervision-02.kt).
+> You can get the full code [here](../kotlinx-coroutines-core/jvm/test/guide/example-supervision-02.kt).
The output of this code is:
```text
-Child is sleeping
-Throwing exception from scope
-Child is cancelled
-Caught assertion error
+The child is sleeping
+Throwing an exception from the scope
+The child is cancelled
+Caught an assertion error
```
#### 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.
-This difference comes from the fact that child's failure is not propagated to the parent.
+Every child should handle its exceptions by itself via the exception handling mechanism.
+This difference comes from the fact that child's failure does not propagate to the parent.
+It means that coroutines launched directly inside the [supervisorScope][_supervisorScope] _do_ use the [CoroutineExceptionHandler]
+that is installed in their scope in the same way as root coroutines do
+(see the [CoroutineExceptionHandler](#coroutineexceptionhandler) section for details).
@@ -466,30 +476,30 @@ import kotlinx.coroutines.*
fun main() = runBlocking {
val handler = CoroutineExceptionHandler { _, exception ->
- println("Caught $exception")
+ println("CoroutineExceptionHandler got $exception")
}
supervisorScope {
val child = launch(handler) {
- println("Child throws an exception")
+ println("The child throws an exception")
throw AssertionError()
}
- println("Scope is completing")
+ println("The scope is completing")
}
- println("Scope is completed")
+ println("The scope is completed")
}
```
-> You can get full code [here](../kotlinx-coroutines-core/jvm/test/guide/example-supervision-03.kt).
+> You can get the full code [here](../kotlinx-coroutines-core/jvm/test/guide/example-supervision-03.kt).
The output of this code is:
```text
-Scope is completing
-Child throws an exception
-Caught java.lang.AssertionError
-Scope is completed
+The scope is completing
+The child throws an exception
+CoroutineExceptionHandler got java.lang.AssertionError
+The scope is completed
```
@@ -501,12 +511,14 @@ Scope is completed
[Deferred.await]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines/-deferred/await.html
[GlobalScope]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines/-global-scope/index.html
[CoroutineExceptionHandler]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines/-coroutine-exception-handler/index.html
+[Job]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines/-job/index.html
+[Deferred]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines/-deferred/index.html
[Job.cancel]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines/-job/cancel.html
[runBlocking]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines/run-blocking.html
[SupervisorJob()]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines/-supervisor-job.html
[Job()]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines/-job.html
-[supervisorScope]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines/supervisor-scope.html
-[coroutineScope]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines/coroutine-scope.html
+[_coroutineScope]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines/coroutine-scope.html
+[_supervisorScope]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines/supervisor-scope.html
[actor]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.channels/actor.html
[produce]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.channels/produce.html
diff --git a/docs/flow.md b/docs/flow.md
index 705f338b20..4374e7aa86 100644
--- a/docs/flow.md
+++ b/docs/flow.md
@@ -10,7 +10,7 @@
* [Suspending functions](#suspending-functions)
* [Flows](#flows)
* [Flows are cold](#flows-are-cold)
- * [Flow cancellation](#flow-cancellation)
+ * [Flow cancellation basics](#flow-cancellation-basics)
* [Flow builders](#flow-builders)
* [Intermediate flow operators](#intermediate-flow-operators)
* [Transform operator](#transform-operator)
@@ -39,31 +39,33 @@
* [Flow completion](#flow-completion)
* [Imperative finally block](#imperative-finally-block)
* [Declarative handling](#declarative-handling)
- * [Upstream exceptions only](#upstream-exceptions-only)
+ * [Successful completion](#successful-completion)
* [Imperative versus declarative](#imperative-versus-declarative)
* [Launching flow](#launching-flow)
+ * [Flow cancellation checks](#flow-cancellation-checks)
+ * [Making busy flow cancellable](#making-busy-flow-cancellable)
* [Flow and Reactive Streams](#flow-and-reactive-streams)
## Asynchronous Flow
-Suspending functions asynchronously returns a single value, but how can we return
+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 function `foo()` that returns a [List]
+For example, we can have a `simple` function that returns a [List]
of three numbers and then print them all using [forEach]:
```kotlin
-fun foo(): List
= listOf(1, 2, 3)
+fun simple(): List = listOf(1, 2, 3)
fun main() {
- foo().forEach { value -> println(value) }
+ simple().forEach { value -> println(value) }
}
```
@@ -89,7 +91,7 @@ If we are computing the numbers with some CPU-consuming blocking code
```kotlin
-fun foo(): Sequence
= sequence { // sequence builder
+fun simple(): Sequence = sequence { // sequence builder
for (i in 1..3) {
Thread.sleep(100) // pretend we are computing it
yield(i) // yield next value
@@ -97,7 +99,7 @@ fun foo(): Sequence = sequence { // sequence builder
}
fun main() {
- foo().forEach { value -> println(value) }
+ simple().forEach { value -> println(value) }
}
```
@@ -116,7 +118,7 @@ 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 function `foo` with a `suspend` modifier,
+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:
@@ -125,13 +127,13 @@ so that it can perform its work without blocking and return the result as a list
import kotlinx.coroutines.*
//sampleStart
-suspend fun foo(): List
{
+suspend fun simple(): List {
delay(1000) // pretend we are doing something asynchronous here
return listOf(1, 2, 3)
}
fun main() = runBlocking {
- foo().forEach { value -> println(value) }
+ simple().forEach { value -> println(value) }
}
//sampleEnd
```
@@ -151,7 +153,7 @@ This code prints the numbers after waiting for a second.
#### Flows
Using the `List` 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`][Flow] type just like we would the `Sequence` type for synchronously computed values:
+the stream of values that are being asynchronously computed, we can use a [`Flow`][Flow] type just like we would use the `Sequence` type for synchronously computed values:
@@ -160,7 +162,7 @@ import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*
//sampleStart
-fun foo(): Flow
= flow { // flow builder
+fun simple(): Flow = flow { // flow builder
for (i in 1..3) {
delay(100) // pretend we are doing something useful here
emit(i) // emit next value
@@ -176,7 +178,7 @@ fun main() = runBlocking {
}
}
// Collect the flow
- foo().collect { value -> println(value) }
+ simple().collect { value -> println(value) }
}
//sampleEnd
```
@@ -201,18 +203,18 @@ I'm not blocked 3
Notice the following differences in the code with the [Flow] from the earlier examples:
-* A builder function for [Flow] type is called [flow].
+* A builder function for [Flow] type is called [flow][_flow].
* Code inside the `flow { ... }` builder block can suspend.
-* The function `foo()` is no longer marked with `suspend` modifier.
+* The `simple` function 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.
-> We can replace [delay] with `Thread.sleep` in the body of `foo`'s `flow { ... }` and see that the main
+> We can replace [delay] with `Thread.sleep` in the body of `simple`'s `flow { ... }` and see that the main
thread is blocked in this case.
### Flows are cold
-Flows are _cold_ streams similar to sequences — the code inside a [flow] builder does not
+Flows are _cold_ streams similar to sequences — the code inside a [flow][_flow] builder does not
run until the flow is collected. This becomes clear in the following example:
@@ -222,7 +224,7 @@ import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*
//sampleStart
-fun foo(): Flow
= flow {
+fun simple(): Flow = flow {
println("Flow started")
for (i in 1..3) {
delay(100)
@@ -231,8 +233,8 @@ fun foo(): Flow = flow {
}
fun main() = runBlocking {
- println("Calling foo...")
- val flow = foo()
+ println("Calling simple function...")
+ val flow = simple()
println("Calling collect...")
flow.collect { value -> println(value) }
println("Calling collect again...")
@@ -248,7 +250,7 @@ fun main() = runBlocking {
Which prints:
```text
-Calling foo...
+Calling simple function...
Calling collect...
Flow started
1
@@ -263,16 +265,14 @@ Flow started
-This is a key reason 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,
+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
-
-Flow adheres to the 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.
+### 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:
@@ -283,7 +283,7 @@ import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*
//sampleStart
-fun foo(): Flow = flow {
+fun simple(): Flow = flow {
for (i in 1..3) {
delay(100)
println("Emitting $i")
@@ -293,7 +293,7 @@ fun foo(): Flow = flow {
fun main() = runBlocking {
withTimeoutOrNull(250) { // Timeout after 250ms
- foo().collect { value -> println(value) }
+ simple().collect { value -> println(value) }
}
println("Done")
}
@@ -304,7 +304,7 @@ fun main() = runBlocking {
> You can get the full code from [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:
+Notice how only two numbers get emitted by the flow in the `simple` function, producing the following output:
```text
Emitting 1
@@ -316,6 +316,8 @@ Done
+See [Flow cancellation checks](#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
@@ -590,14 +592,14 @@ Filter 5
### 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 the implementation details of the `foo` flow:
+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:
```kotlin
withContext(context) {
- foo.collect { value ->
+ simple().collect { value ->
println(value) // run in the specified context
}
}
@@ -610,7 +612,7 @@ withContext(context) {
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
+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:
@@ -622,15 +624,15 @@ import kotlinx.coroutines.flow.*
fun log(msg: String) = println("[${Thread.currentThread().name}] $msg")
//sampleStart
-fun foo(): Flow
= flow {
- log("Started foo flow")
+fun simple(): Flow = flow {
+ log("Started simple flow")
for (i in 1..3) {
emit(i)
}
}
fun main() = runBlocking {
- foo().collect { value -> log("Collected $value") }
+ simple().collect { value -> log("Collected $value") }
}
//sampleEnd
```
@@ -642,7 +644,7 @@ fun main() = runBlocking {
Running this code produces:
```text
-[main @coroutine#1] Started foo flow
+[main @coroutine#1] Started simple flow
[main @coroutine#1] Collected 1
[main @coroutine#1] Collected 2
[main @coroutine#1] Collected 3
@@ -650,7 +652,7 @@ Running this code produces:
-Since `foo().collect` is called from the main thread, the body of `foo`'s flow is also called in the main thread.
+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.
@@ -670,7 +672,7 @@ import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*
//sampleStart
-fun foo(): Flow = flow {
+fun simple(): Flow = flow {
// The WRONG way to change context for CPU-consuming code in flow builder
kotlinx.coroutines.withContext(Dispatchers.Default) {
for (i in 1..3) {
@@ -681,7 +683,7 @@ fun foo(): Flow = flow {
}
fun main() = runBlocking {
- foo().collect { value -> println(value) }
+ simple().collect { value -> println(value) }
}
//sampleEnd
```
@@ -695,7 +697,7 @@ This code produces the following exception:
```text
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].
+ but emission happened in [CoroutineId(1), "coroutine#1":DispatchedCoroutine{Active}@2dae0000, Dispatchers.Default].
Please refer to 'flow' documentation or use 'flowOn' instead
at ...
```
@@ -717,7 +719,7 @@ import kotlinx.coroutines.flow.*
fun log(msg: String) = println("[${Thread.currentThread().name}] $msg")
//sampleStart
-fun foo(): Flow = flow {
+fun simple(): Flow = flow {
for (i in 1..3) {
Thread.sleep(100) // pretend we are computing it in CPU-consuming way
log("Emitting $i")
@@ -726,7 +728,7 @@ fun foo(): Flow = flow {
}.flowOn(Dispatchers.Default) // RIGHT way to change context for CPU-consuming code in flow builder
fun main() = runBlocking {
- foo().collect { value ->
+ simple().collect { value ->
log("Collected $value")
}
}
@@ -757,7 +759,7 @@ creates another coroutine for an upstream flow when it has to change the [Corout
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 `foo()` flow is slow, taking 100 ms to produce an element; and collector is also slow,
+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:
@@ -768,7 +770,7 @@ import kotlinx.coroutines.flow.*
import kotlin.system.*
//sampleStart
-fun foo(): Flow
= flow {
+fun simple(): Flow = flow {
for (i in 1..3) {
delay(100) // pretend we are asynchronously waiting 100 ms
emit(i) // emit next value
@@ -777,7 +779,7 @@ fun foo(): Flow = flow {
fun main() = runBlocking {
val time = measureTimeMillis {
- foo().collect { value ->
+ simple().collect { value ->
delay(300) // pretend we are processing it for 300 ms
println(value)
}
@@ -802,7 +804,7 @@ Collected in 1220 ms
-We can use a [buffer] operator on a flow to run emitting code of `foo()` concurrently with collecting code,
+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:
@@ -812,7 +814,7 @@ import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*
import kotlin.system.*
-fun foo(): Flow
= flow {
+fun simple(): Flow = flow {
for (i in 1..3) {
delay(100) // pretend we are asynchronously waiting 100 ms
emit(i) // emit next value
@@ -822,7 +824,7 @@ fun foo(): Flow = flow {
fun main() = runBlocking {
//sampleStart
val time = measureTimeMillis {
- foo()
+ simple()
.buffer() // buffer emissions, don't wait
.collect { value ->
delay(300) // pretend we are processing it for 300 ms
@@ -867,7 +869,7 @@ import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*
import kotlin.system.*
-fun foo(): Flow = flow {
+fun simple(): Flow = flow {
for (i in 1..3) {
delay(100) // pretend we are asynchronously waiting 100 ms
emit(i) // emit next value
@@ -877,7 +879,7 @@ fun foo(): Flow = flow {
fun main() = runBlocking {
//sampleStart
val time = measureTimeMillis {
- foo()
+ simple()
.conflate() // conflate emissions, don't process each one
.collect { value ->
delay(300) // pretend we are processing it for 300 ms
@@ -918,7 +920,7 @@ import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*
import kotlin.system.*
-fun foo(): Flow = flow {
+fun simple(): Flow = flow {
for (i in 1..3) {
delay(100) // pretend we are asynchronously waiting 100 ms
emit(i) // emit next value
@@ -928,7 +930,7 @@ fun foo(): Flow = flow {
fun main() = runBlocking {
//sampleStart
val time = measureTimeMillis {
- foo()
+ simple()
.collectLatest { value -> // cancel & restart on the latest value
println("Collecting $value")
delay(300) // pretend we are processing it for 300 ms
@@ -1110,7 +1112,7 @@ Now if we have a flow of three integers and call `requestFlow` for each of them
Then we end up with a flow of flows (`Flow>`) 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. However, due the asynchronous nature of flows they call for different _modes_ of flattening,
+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
@@ -1279,7 +1281,7 @@ import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*
//sampleStart
-fun foo(): Flow = flow {
+fun simple(): Flow = flow {
for (i in 1..3) {
println("Emitting $i")
emit(i) // emit next value
@@ -1288,7 +1290,7 @@ fun foo(): Flow = flow {
fun main() = runBlocking {
try {
- foo().collect { value ->
+ simple().collect { value ->
println(value)
check(value <= 1) { "Collected $value" }
}
@@ -1329,7 +1331,7 @@ import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*
//sampleStart
-fun foo(): Flow =
+fun simple(): Flow =
flow {
for (i in 1..3) {
println("Emitting $i")
@@ -1343,7 +1345,7 @@ fun foo(): Flow =
fun main() = runBlocking {
try {
- foo().collect { value -> println(value) }
+ simple().collect { value -> println(value) }
} catch (e: Throwable) {
println("Caught $e")
}
@@ -1390,7 +1392,7 @@ For example, let us emit the text on catching an exception:
import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*
-fun foo(): Flow =
+fun simple(): Flow =
flow {
for (i in 1..3) {
println("Emitting $i")
@@ -1404,7 +1406,7 @@ fun foo(): Flow =
fun main() = runBlocking {
//sampleStart
- foo()
+ simple()
.catch { e -> emit("Caught $e") } // emit on exception
.collect { value -> println(value) }
//sampleEnd
@@ -1437,7 +1439,7 @@ import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*
//sampleStart
-fun foo(): Flow = flow {
+fun simple(): Flow = flow {
for (i in 1..3) {
println("Emitting $i")
emit(i)
@@ -1445,7 +1447,7 @@ fun foo(): Flow = flow {
}
fun main() = runBlocking {
- foo()
+ simple()
.catch { e -> println("Caught $e") } // does not catch downstream exceptions
.collect { value ->
check(value <= 1) { "Collected $value" }
@@ -1461,13 +1463,15 @@ fun main() = runBlocking {
A "Caught ..." message is not printed despite there being a `catch` operator:
-
+```
+
+
#### Catching declaratively
@@ -1481,7 +1485,7 @@ be triggered by a call to `collect()` without parameters:
import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*
-fun foo(): Flow = flow {
+fun simple(): Flow = flow {
for (i in 1..3) {
println("Emitting $i")
emit(i)
@@ -1490,7 +1494,7 @@ fun foo(): Flow = flow {
fun main() = runBlocking {
//sampleStart
- foo()
+ simple()
.onEach { value ->
check(value <= 1) { "Collected $value" }
println(value)
@@ -1508,12 +1512,14 @@ fun main() = runBlocking {
Now we can see that a "Caught ..." message is printed and so we can catch all the exceptions without explicitly
using a `try/catch` block:
-
+```
+
+
### Flow completion
@@ -1532,11 +1538,11 @@ import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*
//sampleStart
-fun foo(): Flow = (1..3).asFlow()
+fun simple(): Flow = (1..3).asFlow()
fun main() = runBlocking {
try {
- foo().collect { value -> println(value) }
+ simple().collect { value -> println(value) }
} finally {
println("Done")
}
@@ -1548,7 +1554,7 @@ fun main() = runBlocking {
> You can get the full code from [here](../kotlinx-coroutines-core/jvm/test/guide/example-flow-31.kt).
-This code prints three numbers produced by the `foo()` flow followed by a "Done" string:
+This code prints three numbers produced by the `simple` flow followed by a "Done" string:
```text
1
@@ -1572,11 +1578,11 @@ The previous example can be rewritten using an [onCompletion] operator and produ
import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*
-fun foo(): Flow = (1..3).asFlow()
+fun simple(): Flow = (1..3).asFlow()
fun main() = runBlocking {
//sampleStart
- foo()
+ simple()
.onCompletion { println("Done") }
.collect { value -> println(value) }
//sampleEnd
@@ -1595,7 +1601,7 @@ Done
The key advantage of [onCompletion] is a nullable `Throwable` parameter of the lambda that can be used
to determine whether the flow collection was completed normally or exceptionally. In the following
-example the `foo()` flow throws an exception after emitting the number 1:
+example the `simple` flow throws an exception after emitting the number 1:
@@ -1604,13 +1610,13 @@ import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*
//sampleStart
-fun foo(): Flow
= flow {
+fun simple(): Flow = flow {
emit(1)
throw RuntimeException()
}
fun main() = runBlocking {
- foo()
+ simple()
.onCompletion { cause -> if (cause != null) println("Flow completed exceptionally") }
.catch { cause -> println("Caught exception") }
.collect { value -> println(value) }
@@ -1635,10 +1641,10 @@ The [onCompletion] operator, unlike [catch], does not handle the exception. As w
example code, the exception still flows downstream. It will be delivered to further `onCompletion` operators
and can be handled with a `catch` operator.
-#### Upstream exceptions only
+#### Successful completion
-Just like the [catch] operator, [onCompletion] only sees exceptions coming from upstream and does not
-see downstream exceptions. For example, run the following code:
+Another difference with [catch] operator is that [onCompletion] sees all exceptions and receives
+a `null` exception only on successful completion of the upstream flow (without cancellation or failure).
@@ -1647,10 +1653,10 @@ import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*
//sampleStart
-fun foo(): Flow
= (1..3).asFlow()
+fun simple(): Flow = (1..3).asFlow()
fun main() = runBlocking {
- foo()
+ simple()
.onCompletion { cause -> println("Flow completed with $cause") }
.collect { value ->
check(value <= 1) { "Collected $value" }
@@ -1664,11 +1670,11 @@ fun main() = runBlocking {
> You can get the full code from [here](../kotlinx-coroutines-core/jvm/test/guide/example-flow-34.kt).
-We can see the completion cause is null, yet collection failed with exception:
+We can see the completion cause is not null, because the flow was aborted due to downstream exception:
```text
1
-Flow completed with null
+Flow completed with java.lang.IllegalStateException: Collected 2
Exception in thread "main" java.lang.IllegalStateException: Collected 2
```
@@ -1777,6 +1783,127 @@ 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.
+### Flow cancellation checks
+
+For convenience, the [flow][_flow] builder performs additional [ensureActive] checks for cancellation on each emitted value.
+It means that a busy loop emitting from a `flow { ... }` is cancellable:
+
+
+
+```kotlin
+import kotlinx.coroutines.*
+import kotlinx.coroutines.flow.*
+
+//sampleStart
+fun foo(): Flow = flow {
+ for (i in 1..5) {
+ println("Emitting $i")
+ emit(i)
+ }
+}
+
+fun main() = runBlocking {
+ foo().collect { value ->
+ if (value == 3) cancel()
+ println(value)
+ }
+}
+//sampleEnd
+```
+
+
+
+> You can get the full code from [here](../kotlinx-coroutines-core/jvm/test/guide/example-flow-37.kt).
+
+We get only numbers up to 3 and a [CancellationException] after trying to emit number 4:
+
+```text
+Emitting 1
+1
+Emitting 2
+2
+Emitting 3
+3
+Emitting 4
+Exception in thread "main" kotlinx.coroutines.JobCancellationException: BlockingCoroutine was cancelled; job="coroutine#1":BlockingCoroutine{Cancelled}@6d7b4f4c
+```
+
+
+
+However, most other flow operators do not do additional cancellation checks on their own for performance reasons.
+For example, if you use [IntRange.asFlow] extension to write the same busy loop and don't suspend anywhere,
+then there are no checks for cancellation:
+
+
+
+```kotlin
+import kotlinx.coroutines.*
+import kotlinx.coroutines.flow.*
+
+//sampleStart
+fun main() = runBlocking {
+ (1..5).asFlow().collect { value ->
+ if (value == 3) cancel()
+ println(value)
+ }
+}
+//sampleEnd
+```
+
+
+
+> You can get the full code from [here](../kotlinx-coroutines-core/jvm/test/guide/example-flow-38.kt).
+
+All numbers from 1 to 5 are collected and cancellation gets detected only before return from `runBlocking`:
+
+```text
+1
+2
+3
+4
+5
+Exception in thread "main" kotlinx.coroutines.JobCancellationException: BlockingCoroutine was cancelled; job="coroutine#1":BlockingCoroutine{Cancelled}@3327bd23
+```
+
+
+
+#### Making busy flow cancellable
+
+In the case where you have a busy loop with coroutines you must explicitly check for cancellation.
+You can add `.onEach { currentCoroutineContext().ensureActive() }`, but there is a ready-to-use
+[cancellable] operator provided to do that:
+
+
+
+```kotlin
+import kotlinx.coroutines.*
+import kotlinx.coroutines.flow.*
+
+//sampleStart
+fun main() = runBlocking {
+ (1..5).asFlow().cancellable().collect { value ->
+ if (value == 3) cancel()
+ println(value)
+ }
+}
+//sampleEnd
+```
+
+
+
+> You can get the full code from [here](../kotlinx-coroutines-core/jvm/test/guide/example-flow-39.kt).
+
+With the `cancellable` operator only the numbers from 1 to 3 are collected:
+
+```text
+1
+2
+3
+Exception in thread "main" kotlinx.coroutines.JobCancellationException: BlockingCoroutine was cancelled; job="coroutine#1":BlockingCoroutine{Cancelled}@5ec0a365
+```
+
+
+
### Flow and Reactive Streams
For those who are familiar with [Reactive Streams](https://www.reactive-streams.org/) or reactive frameworks such as RxJava and project Reactor,
@@ -1786,7 +1913,7 @@ Indeed, its design was inspired by Reactive Streams and its various implementati
be Kotlin and suspension friendly and respect structured concurrency. Achieving this goal would be impossible without reactive pioneers and their tremendous work. You can read the complete story in [Reactive Streams and Kotlin Flows](https://medium.com/@elizarov/reactive-streams-and-kotlin-flows-bfd12772cda4) article.
While being different, conceptually, Flow *is* a reactive stream and it is possible to convert it to the reactive (spec and TCK compliant) Publisher and vice versa.
-Such converters are provided by `kotlinx.coroutines` out-of-the-box and can be found in corresponding reactive modules (`kotlinx-coroutines-reactive` for Reactive Streams, `kotlinx-coroutines-reactor` for Project Reactor and `kotlinx-coroutines-rx2` for RxJava2).
+Such converters are provided by `kotlinx.coroutines` out-of-the-box and can be found in corresponding reactive modules (`kotlinx-coroutines-reactive` for Reactive Streams, `kotlinx-coroutines-reactor` for Project Reactor and `kotlinx-coroutines-rx2`/`kotlinx-coroutines-rx3` for RxJava2/RxJava3).
Integration modules include conversions from and to `Flow`, integration with Reactor's `Context` and suspension-friendly ways to work with various reactive entities.
@@ -1813,9 +1940,11 @@ Integration modules include conversions from and to `Flow`, integration with Rea
[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
+[ensureActive]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines/ensure-active.html
+[CancellationException]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines/-cancellation-exception/index.html
[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
+[_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
@@ -1845,4 +1974,6 @@ Integration modules include conversions from and to `Flow`, integration with Rea
[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
+[IntRange.asFlow]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.flow/kotlin.ranges.-int-range/as-flow.html
+[cancellable]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.flow/cancellable.html
diff --git a/docs/images/coroutine-idea-debugging-1.png b/docs/images/coroutine-idea-debugging-1.png
new file mode 100644
index 0000000000..0afe992515
Binary files /dev/null and b/docs/images/coroutine-idea-debugging-1.png differ
diff --git a/docs/knit.properties b/docs/knit.properties
index ab2508a114..2028ecb416 100644
--- a/docs/knit.properties
+++ b/docs/knit.properties
@@ -4,19 +4,7 @@
knit.package=kotlinx.coroutines.guide
knit.dir=../kotlinx-coroutines-core/jvm/test/guide/
-knit.pattern=example-[a-zA-Z0-9-]+-##\\.kt
-knit.include=knit.code.include
test.package=kotlinx.coroutines.guide.test
test.dir=../kotlinx-coroutines-core/jvm/test/guide/test/
-test.template=knit.test.template
-# Various test validation modes and their corresponding methods from TestUtil
-test.mode.=verifyLines
-test.mode.STARTS_WITH=verifyLinesStartWith
-test.mode.ARBITRARY_TIME=verifyLinesArbitraryTime
-test.mode.FLEXIBLE_TIME=verifyLinesFlexibleTime
-test.mode.FLEXIBLE_THREAD=verifyLinesFlexibleThread
-test.mode.LINES_START_UNORDERED=verifyLinesStartUnordered
-test.mode.LINES_START=verifyLinesStart
-test.mode.EXCEPTION=verifyExceptions
\ No newline at end of file
diff --git a/docs/knit.test.template b/docs/knit.test.template
index a912555a43..727493c662 100644
--- a/docs/knit.test.template
+++ b/docs/knit.test.template
@@ -5,6 +5,7 @@
// This file was automatically generated from ${file.name} by Knit tool. Do not edit.
package ${test.package}
+import kotlinx.coroutines.knit.*
import org.junit.Test
class ${test.name} {
diff --git a/docs/select-expression.md b/docs/select-expression.md
index 5809e7b93e..f0e5ae4681 100644
--- a/docs/select-expression.md
+++ b/docs/select-expression.md
@@ -125,7 +125,7 @@ fun main() = runBlocking {
-> You can get full code [here](../kotlinx-coroutines-core/jvm/test/guide/example-select-05.kt).
+> You can get the full code [here](../kotlinx-coroutines-core/jvm/test/guide/example-select-05.kt).
The result of this code:
diff --git a/docs/shared-mutable-state-and-concurrency.md b/docs/shared-mutable-state-and-concurrency.md
index 1a3c406472..8b83ad0b20 100644
--- a/docs/shared-mutable-state-and-concurrency.md
+++ b/docs/shared-mutable-state-and-concurrency.md
@@ -24,7 +24,7 @@ but others are unique.
### The problem
-Let us launch a hundred coroutines all doing the same action thousand times.
+Let us launch a hundred coroutines all doing the same action a thousand times.
We'll also measure their completion time for further comparisons:
@@ -90,7 +90,7 @@ fun main() = runBlocking {
-> You can get full code [here](../kotlinx-coroutines-core/jvm/test/guide/example-sync-01.kt).
+> You can get the full code [here](../kotlinx-coroutines-core/jvm/test/guide/example-sync-01.kt).
@@ -144,7 +144,7 @@ fun main() = runBlocking {