Flow exceptions handling facilities (#1280)

* Flow documentation introduces and explains "exception transparency"
  concept in detail.
* Introduce exception handling operators:
  * catch intermediate operator.
  * retry and retryWhen intermediate operators.
  * collect() without lambda terminal operator.
* onErrorXxx operators deprecated and moved to migration file when
  it was appropriate.

Author: elizarov

binary-compatibility-validator/reference-public-api/kotlinx-coroutines-core.txt

@@ -816,7 +816,9 @@ public final class kotlinx/coroutines/flow/FlowKt {
 	public static final fun buffer (Lkotlinx/coroutines/flow/Flow;I)Lkotlinx/coroutines/flow/Flow;
 	public static synthetic fun buffer$default (Lkotlinx/coroutines/flow/Flow;IILjava/lang/Object;)Lkotlinx/coroutines/flow/Flow;
 	public static final fun callbackFlow (Lkotlin/jvm/functions/Function2;)Lkotlinx/coroutines/flow/Flow;
+	public static final fun catch (Lkotlinx/coroutines/flow/Flow;Lkotlin/jvm/functions/Function3;)Lkotlinx/coroutines/flow/Flow;
 	public static final fun channelFlow (Lkotlin/jvm/functions/Function2;)Lkotlinx/coroutines/flow/Flow;
+	public static final fun collect (Lkotlinx/coroutines/flow/Flow;Lkotlin/coroutines/Continuation;)Ljava/lang/Object;
 	public static final fun collect (Lkotlinx/coroutines/flow/Flow;Lkotlin/jvm/functions/Function2;Lkotlin/coroutines/Continuation;)Ljava/lang/Object;
 	public static final fun combineLatest (Lkotlinx/coroutines/flow/Flow;Lkotlinx/coroutines/flow/Flow;Lkotlin/jvm/functions/Function3;)Lkotlinx/coroutines/flow/Flow;
 	public static final fun combineLatest (Lkotlinx/coroutines/flow/Flow;Lkotlinx/coroutines/flow/Flow;Lkotlinx/coroutines/flow/Flow;Lkotlin/jvm/functions/Function4;)Lkotlinx/coroutines/flow/Flow;
@@ -863,12 +865,13 @@ public final class kotlinx/coroutines/flow/FlowKt {
 	public static final fun onEach (Lkotlinx/coroutines/flow/Flow;Lkotlin/jvm/functions/Function2;)Lkotlinx/coroutines/flow/Flow;
 	public static final fun onErrorCollect (Lkotlinx/coroutines/flow/Flow;Lkotlinx/coroutines/flow/Flow;Lkotlin/jvm/functions/Function1;)Lkotlinx/coroutines/flow/Flow;
 	public static synthetic fun onErrorCollect$default (Lkotlinx/coroutines/flow/Flow;Lkotlinx/coroutines/flow/Flow;Lkotlin/jvm/functions/Function1;ILjava/lang/Object;)Lkotlinx/coroutines/flow/Flow;
-	public static final fun onErrorReturn (Lkotlinx/coroutines/flow/Flow;Ljava/lang/Object;Lkotlin/jvm/functions/Function1;)Lkotlinx/coroutines/flow/Flow;
-	public static synthetic fun onErrorReturn$default (Lkotlinx/coroutines/flow/Flow;Ljava/lang/Object;Lkotlin/jvm/functions/Function1;ILjava/lang/Object;)Lkotlinx/coroutines/flow/Flow;
 	public static final fun produceIn (Lkotlinx/coroutines/flow/Flow;Lkotlinx/coroutines/CoroutineScope;)Lkotlinx/coroutines/channels/ReceiveChannel;
 	public static final fun reduce (Lkotlinx/coroutines/flow/Flow;Lkotlin/jvm/functions/Function3;Lkotlin/coroutines/Continuation;)Ljava/lang/Object;
-	public static final fun retry (Lkotlinx/coroutines/flow/Flow;ILkotlin/jvm/functions/Function1;)Lkotlinx/coroutines/flow/Flow;
+	public static final synthetic fun retry (Lkotlinx/coroutines/flow/Flow;ILkotlin/jvm/functions/Function1;)Lkotlinx/coroutines/flow/Flow;
+	public static final fun retry (Lkotlinx/coroutines/flow/Flow;JLkotlin/jvm/functions/Function2;)Lkotlinx/coroutines/flow/Flow;
 	public static synthetic fun retry$default (Lkotlinx/coroutines/flow/Flow;ILkotlin/jvm/functions/Function1;ILjava/lang/Object;)Lkotlinx/coroutines/flow/Flow;
+	public static synthetic fun retry$default (Lkotlinx/coroutines/flow/Flow;JLkotlin/jvm/functions/Function2;ILjava/lang/Object;)Lkotlinx/coroutines/flow/Flow;
+	public static final fun retryWhen (Lkotlinx/coroutines/flow/Flow;Lkotlin/jvm/functions/Function4;)Lkotlinx/coroutines/flow/Flow;
 	public static final fun sample (Lkotlinx/coroutines/flow/Flow;J)Lkotlinx/coroutines/flow/Flow;
 	public static final fun scan (Lkotlinx/coroutines/flow/Flow;Ljava/lang/Object;Lkotlin/jvm/functions/Function3;)Lkotlinx/coroutines/flow/Flow;
 	public static final fun scanReduce (Lkotlinx/coroutines/flow/Flow;Lkotlin/jvm/functions/Function3;)Lkotlinx/coroutines/flow/Flow;
@@ -899,6 +902,10 @@ public final class kotlinx/coroutines/flow/MigrationKt {
 	public static final fun merge (Lkotlinx/coroutines/flow/Flow;)Lkotlinx/coroutines/flow/Flow;
 	public static final fun observeOn (Lkotlinx/coroutines/flow/Flow;Lkotlin/coroutines/CoroutineContext;)Lkotlinx/coroutines/flow/Flow;
 	public static final fun onErrorResume (Lkotlinx/coroutines/flow/Flow;Lkotlinx/coroutines/flow/Flow;)Lkotlinx/coroutines/flow/Flow;
+	public static final fun onErrorResumeNext (Lkotlinx/coroutines/flow/Flow;Lkotlinx/coroutines/flow/Flow;)Lkotlinx/coroutines/flow/Flow;
+	public static final fun onErrorReturn (Lkotlinx/coroutines/flow/Flow;Ljava/lang/Object;)Lkotlinx/coroutines/flow/Flow;
+	public static final fun onErrorReturn (Lkotlinx/coroutines/flow/Flow;Ljava/lang/Object;Lkotlin/jvm/functions/Function1;)Lkotlinx/coroutines/flow/Flow;
+	public static synthetic fun onErrorReturn$default (Lkotlinx/coroutines/flow/Flow;Ljava/lang/Object;Lkotlin/jvm/functions/Function1;ILjava/lang/Object;)Lkotlinx/coroutines/flow/Flow;
 	public static final fun publishOn (Lkotlinx/coroutines/flow/Flow;Lkotlin/coroutines/CoroutineContext;)Lkotlinx/coroutines/flow/Flow;
 	public static final fun scanFold (Lkotlinx/coroutines/flow/Flow;Ljava/lang/Object;Lkotlin/jvm/functions/Function3;)Lkotlinx/coroutines/flow/Flow;
 	public static final fun skip (Lkotlinx/coroutines/flow/Flow;I)Lkotlinx/coroutines/flow/Flow;

kotlinx-coroutines-core/common/src/Annotations.kt

@@ -31,6 +31,7 @@ public annotation class ExperimentalCoroutinesApi
 @MustBeDocumented
 @Retention(value = AnnotationRetention.BINARY)
 @Experimental(level = Experimental.Level.WARNING)
+@Target(AnnotationTarget.CLASS, AnnotationTarget.FUNCTION, AnnotationTarget.TYPEALIAS, AnnotationTarget.PROPERTY)
 public annotation class FlowPreview
 
 /**

kotlinx-coroutines-core/common/src/flow/Flow.kt

@@ -12,13 +12,17 @@ import kotlin.coroutines.*
  * A cold asynchronous data stream that sequentially emits values
  * and completes normally or with an exception.
  *
- * _Cold flow_ means that intermediate operators on a flow such as [map] and [filter] do not trigger its execution,
- * which is only done by terminal operators like [single]. By default, flows are _sequential_ and all flow
- * operations are executed sequentially in the same coroutine, see [buffer] for details.
- *
- * _Collecting the flow_ means executing all its operations.
- * Flow values can be collected in a suspending manner without actual blocking using the [collect] extension that
- * completes normally or exceptionally:
+ * _Intermediate operators_ on the flow such as [map], [filter], [take], [zip], etc are functions that are
+ * applied to the _upstream_ flow or flows and return a _downstream_ flow where further operators can be applied to.
+ * Intermediate operations do not execute any code in the flow and are not suspending functions themselves.
+ * They only set up a chain of operations for future execution and quickly return.
+ * This is known as a _cold flow_ property.
+ *
+ * _Terminal operators_ on the flow are suspending functions such as [collect], [single], [reduce], [toList], etc.
+ * They are applied to the upstream flow and trigger execution of all operations.
+ * Execution of the flow is also called _collecting the flow_  and is always performed in a suspending manner
+ * without actual blocking. Terminal operator complete normally or exceptionally depending on successful or failed
+ * execution of all the flow operations in the upstream. The most basic terminal operator is [collect], for example:
  *
  * ```
  * try {
@@ -30,10 +34,13 @@ import kotlin.coroutines.*
  * }
  * ```
  *
- * Additionally, the library provides a rich set of terminal operators such as [single], [reduce] and others.
+ * By default, flows are _sequential_ and all flow operations are executed sequentially in the same coroutine,
+ * with an exception for a few operations specifically designed to introduce concurrency into flow
+ * the execution such a [buffer] and [flatMapMerge]. See their documentation for details.
  *
- * Flows don't carry information whether they are cold streams (which can be collected repeatedly and
- * trigger their evaluation every time [collect] is executed) or hot ones, but, conventionally, they represent cold streams.
+ * Flow interface does not carry information whether a flow is a truly a cold stream that can be collected repeatedly and
+ * triggers execution of the same code every time it is collected or if it is a hot stream that emits different
+ * values from the same running source on each collection. However, conventionally flows represent cold streams.
  * Transitions between hot and cold streams are supported via channels and the corresponding API:
  * [channelFlow], [produceIn], [broadcastIn].
  *
@@ -48,7 +55,18 @@ import kotlin.coroutines.*
  * * [channelFlow { ... }][channelFlow] builder function to construct arbitrary flows from
  *   potentially concurrent calls to [send][kotlinx.coroutines.channels.SendChannel.send] function.
  *
- * ### Flow context
+ * ### Flow constraints
+ *
+ * All implementations of `Flow` interface must adhere to two key properties that are described in detail below:
+ *
+ * * Context preservation.
+ * * Exception transparency.
+ *
+ * These properties ensure the ability to perform local reasoning about the code with flows and modularize the code
+ * in such a way so that upstream flow emitters can be developed separately from downstream flow collectors.
+ * A user of the flow does not needs to know implementation details of the upstream flows it uses.
+ *
+ * ### Context preservation
  *
  * The flow has a context preservation property: it encapsulates its own execution context and never propagates or leaks
  * it downstream, thus making reasoning about the execution context of particular transformations or terminal
@@ -77,7 +95,7 @@ import kotlin.coroutines.*
  * }
  * ```
  *
- * From the implementation point of view it means that all flow implementations should
+ * From the implementation point of view, it means that all flow implementations should
  * emit only from the same coroutine.
  * This constraint is efficiently enforced by the default [flow] builder.
  * The [flow] builder should be used if flow implementation does not start any coroutines.
@@ -108,29 +126,57 @@ import kotlin.coroutines.*
  *  - Collecting another flow from a separate context is allowed, but it has the same effect as
  *    [flowOn] operator on that flow, which is more efficient.
  *
+ * ### Exception transparency
+ *
+ * Flow implementations never catch or handle exceptions that occur in downstream flows. From the implementation standpoint
+ * it means that calls to [emit][FlowCollector.emit] and [emitAll] shall never be wrapped into
+ * `try { ... } catch { ... }` blocks. Exception handling in flows shall be performed with
+ * [catch][Flow.catch] operator and it is designed to catch only exception coming from upstream flow while passing
+ * all the downstream exceptions. Similarly, terminal operators like [collect][Flow.collect]
+ * throw any unhandled exception that occurs in its code or in upstream flows, for example:
+ *
+ * ```
+ * flow { emitData() }
+ *     .map { computeOne(it) }
+ *     .catch { ... } // catches exceptions in emitData and computeOne
+ *     .map { computeTwo(it) }
+ *     .collect { process(it) } // throws exceptions from process and computeTwo
+ * ```
+ *
+ * Failure to adhere to the exception transparency requirement would result in strange behaviours that would make
+ * it hard to reason about the code because an exception in the `collect { ... }` could be somehow "caught"
+ * by the upstream flow, limiting the ability of local reasoning about the code.
+ *
+ * Currently, flow infrastructure does not enforce exception transparency contracts, however, it might be enforced
+ * in the future either at run time or at compile time.
+ *
+ * ### Reactive streams
+ *
  * Flow is [Reactive Streams](http://www.reactive-streams.org/) compliant, you can safely interop it with
- * reactive streams using [Flow.asPublisher] and [Publisher.asFlow] from kotlinx-coroutines-reactive module.
+ * reactive streams using [Flow.asPublisher] and [Publisher.asFlow] from `kotlinx-coroutines-reactive` module.
  */
 @ExperimentalCoroutinesApi
 public interface Flow<out T> {
-
     /**
      * Accepts the given [collector] and [emits][FlowCollector.emit] values into it.
      * This method should never be implemented or used directly.
      *
      * The only way to implement flow interface directly is to extend [AbstractFlow].
-     * To collect it into the specific collector, either `collector.emitAll(flow)` or `collect { }` extension should be used.
-     * Such limitation ensures that context preservation property is not violated and prevents most of the developer mistakes
-     * related to concurrency, inconsistent flow dispatchers and cancellation.
+     * To collect it into the specific collector, either `collector.emitAll(flow)` or `collect { ... }` extension
+     * should be used. Such limitation ensures that context preservation property is not violated and prevents most
+     * of the developer mistakes related to concurrency, inconsistent flow dispatchers and cancellation.
      */
     @InternalCoroutinesApi
     public suspend fun collect(collector: FlowCollector<T>)
 }
 
 /**
  * Base class to extend to have a stateful implementation of the flow.
- * It tracks all the properties required for context preservation and throws [IllegalStateException] if any of the properties are violated.
+ * It tracks all the properties required for context preservation and throws [IllegalStateException]
+ * if any of the properties are violated.
+ * 
  * Example of the implementation:
+ *
  * ```
  * // list.asFlow() + collect counter
  * class CountingListFlow(private val values: List<Int>) : AbstractFlow<Int>() {

kotlinx-coroutines-core/common/src/flow/Migration.kt

@@ -105,11 +105,18 @@ public fun PublishSubject(): Any = error("Should not be called")
 /** @suppress **/
 @Deprecated(
     level = DeprecationLevel.ERROR,
-    message = "Flow analogue is named onErrorCollect",
-    replaceWith = ReplaceWith("onErrorCollect(fallback)")
+    message = "Flow analogue of 'onErrorXxx' is 'catch'. Use 'catch { emitAll(fallback) }'",
+    replaceWith = ReplaceWith("catch { emitAll(fallback) }")
 )
 public fun <T> Flow<T>.onErrorResume(fallback: Flow<T>): Flow<T> = error("Should not be called")
 
+@Deprecated(
+    level = DeprecationLevel.ERROR,
+    message = "Flow analogue of 'onErrorXxx' is 'catch'. Use 'catch { emitAll(fallback) }'",
+    replaceWith = ReplaceWith("catch { emitAll(fallback) }")
+)
+public fun <T> Flow<T>.onErrorResumeNext(fallback: Flow<T>): Flow<T> = error("Should not be called")
+
 /**
  * Self-explanatory, the reason of deprecation is "context preservation" property (you can read more in [Flow] documentation)
  * @suppress
@@ -181,15 +188,15 @@ public fun <T, R> Flow<T>.concatMap(mapper: (T) -> Flow<R>): Flow<R> = error("Sh
  */
 @Deprecated(
     level = DeprecationLevel.ERROR,
-    message = "Flow analogue is named flattenConcat",
+    message = "Flow analogue of 'merge' is 'flattenConcat'",
     replaceWith = ReplaceWith("flattenConcat()")
 )
 public fun <T> Flow<Flow<T>>.merge(): Flow<T> = error("Should not be called")
 
 /** @suppress **/
 @Deprecated(
     level = DeprecationLevel.ERROR,
-    message = "Flow analogue is named flattenConcat",
+    message = "Flow analogue of 'flatten' is 'flattenConcat'",
     replaceWith = ReplaceWith("flattenConcat()")
 )
 public fun <T> Flow<Flow<T>>.flatten(): Flow<T> = error("Should not be called")
@@ -210,7 +217,7 @@ public fun <T> Flow<Flow<T>>.flatten(): Flow<T> = error("Should not be called")
  */
 @Deprecated(
     level = DeprecationLevel.ERROR,
-    message = "Kotlin analogue of compose is 'let'",
+    message = "Flow analogue of 'compose' is 'let'",
     replaceWith = ReplaceWith("let(transformer)")
 )
 public fun <T, R> Flow<T>.compose(transformer: Flow<T>.() -> Flow<R>): Flow<R> = error("Should not be called")
@@ -220,7 +227,7 @@ public fun <T, R> Flow<T>.compose(transformer: Flow<T>.() -> Flow<R>): Flow<R> =
  */
 @Deprecated(
     level = DeprecationLevel.ERROR,
-    message = "Kotlin analogue of 'skip' is 'drop'",
+    message = "Flow analogue of 'skip' is 'drop'",
     replaceWith = ReplaceWith("drop(count)")
 )
 public fun <T> Flow<T>.skip(count: Int): Flow<T> = error("Should not be called")
@@ -246,3 +253,23 @@ public fun <T> Flow<T>.forEach(action: suspend (value: T) -> Unit): Unit = error
     replaceWith = ReplaceWith("scan(initial, operation)")
 )
 public fun <T, R> Flow<T>.scanFold(initial: R, @BuilderInference operation: suspend (accumulator: R, value: T) -> R): Flow<R> = error("Should not be called")
+
+@Deprecated(
+    level = DeprecationLevel.ERROR,
+    message = "Flow analogue of 'onErrorXxx' is 'catch'. Use 'catch { emit(fallback) }'",
+    replaceWith = ReplaceWith("catch { emit(fallback) }")
+)
+// Note: this version without predicate gives better "replaceWith" action
+public fun <T> Flow<T>.onErrorReturn(fallback: T): Flow<T> = error("Should not be called")
+
+@Deprecated(
+    level = DeprecationLevel.ERROR,
+    message = "Flow analogue of 'onErrorXxx' is 'catch'. Use 'catch { e -> if (predicate(e)) emit(fallback) else throw e }'",
+    replaceWith = ReplaceWith("catch { e -> if (predicate(e)) emit(fallback) else throw e }")
+)
+public fun <T> Flow<T>.onErrorReturn(fallback: T, predicate: (Throwable) -> Boolean = { true }): Flow<T> =
+    catch { e ->
+        // Note: default value is for binary compatibility with preview version, that is why it has body
+        if (!predicate(e)) throw e
+        emit(fallback)
+    }

kotlinx-coroutines-core/common/src/flow/internal/ChannelFlow.kt

@@ -134,8 +134,8 @@ internal class ChannelFlowOperatorImpl<T>(
 // Now if the underlying collector was accepting concurrent emits, then this one is too
 // todo: we might need to generalize this pattern for "thread-safe" operators that can fuse with channels
 private fun <T> FlowCollector<T>.withUndispatchedContextCollector(emitContext: CoroutineContext): FlowCollector<T> = when (this) {
-    // SendingCollector does not care about the context at all so can be used as it
-    is SendingCollector -> this
+    // SendingCollector & NopCollector do not care about the context at all and can be used as is
+    is SendingCollector, is NopCollector -> this
     // Original collector is concurrent, so wrap into ConcurrentUndispatchedContextCollector (also concurrent)
     is ConcurrentFlowCollector -> ConcurrentUndispatchedContextCollector(this, emitContext)
     // Otherwise just wrap into UndispatchedContextCollector interface implementation

kotlinx-coroutines-core/common/src/flow/internal/NopCollector.kt

@@ -0,0 +1,13 @@
+/*
+ * Copyright 2016-2019 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license.
+ */
+
+package kotlinx.coroutines.flow.internal
+
+import kotlinx.coroutines.flow.*
+
+internal object NopCollector : ConcurrentFlowCollector<Any?> {
+    override suspend fun emit(value: Any?) {
+        // does nothing
+    }
+}

kotlinx-coroutines-core/common/src/flow/operators/Context.kt

@@ -151,7 +151,7 @@ public fun <T> Flow<T>.buffer(capacity: Int = BUFFERED): Flow<T> {
 public fun <T> Flow<T>.conflate(): Flow<T> = buffer(CONFLATED)
 
 /**
- * The operator that changes the context where this flow is executed to the given [context].
+ * Changes the context where this flow is executed to the given [context].
  * This operator is composable and affects only preceding operators that do not have its own context.
  * This operator is context preserving: [context] **does not** leak into the downstream flow.
  *