Rework purity concept:

   * Purity is a way too polluted word and it is hard to tell what it means in Flow
   * Rename "purity" to "context preservation"
   * Prohibit changing the context, ignore only Job and CoroutineId (for debug mode) in SafeCollector
   * Reword documentation, add more samples
   * Add explanation for deprecated Rx-like methods

Author: qwwdfsad

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

@@ -864,7 +864,7 @@ public final class kotlinx/coroutines/flow/MigrationKt {
 }
 
 public final class kotlinx/coroutines/flow/internal/SafeCollector : kotlinx/coroutines/flow/FlowCollector {
-	public fun <init> (Lkotlinx/coroutines/flow/FlowCollector;Lkotlin/coroutines/ContinuationInterceptor;)V
+	public fun <init> (Lkotlinx/coroutines/flow/FlowCollector;Lkotlin/coroutines/CoroutineContext;)V
 	public fun emit (Ljava/lang/Object;Lkotlin/coroutines/Continuation;)Ljava/lang/Object;
 }
 

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

@@ -20,4 +20,5 @@ internal expect val DefaultDelay: Delay
 // countOrElement -- pre-cached value for ThreadContext.kt
 internal expect inline fun <T> withCoroutineContext(context: CoroutineContext, countOrElement: Any?, block: () -> T): T
 internal expect fun Continuation<*>.toDebugString(): String
-internal expect val CoroutineContext.coroutineName: String?
+internal expect val CoroutineContext.coroutineName: String?
+internal expect fun CoroutineContext.minusId(): CoroutineContext

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

@@ -19,26 +19,26 @@ import kotlin.jvm.*
  * Example of usage:
  * ```
  * fun fibonacci(): Flow<Long> = flow {
- *   emit(1L)
- *   var f1 = 1L
- *   var f2 = 1L
- *   repeat(100) {
- *     var tmp = f1
- *     f1 = f2
- *     f2 += tmp
- *     emit(f1)
- *   }
+ *     emit(1L)
+ *     var f1 = 1L
+ *     var f2 = 1L
+ *     repeat(100) {
+ *         var tmp = f1
+ *         f1 = f2
+ *         f2 += tmp
+ *         emit(f1)
+ *     }
  * }
  * ```
  *
- * `emit` should happen strictly in the dispatchers of the [block] in order to preserve flow purity.
+ * `emit` should happen strictly in the dispatchers of the [block] in order to preserve flow context.
  * For example, the following code will produce [IllegalStateException]:
  * ```
  * flow {
- *   emit(1) // Ok
- *   withContext(Dispatcher.IO) {
- *       emit(2) // Will fail with ISE
- *   }
+ *     emit(1) // Ok
+ *     withContext(Dispatcher.IO) {
+ *         emit(2) // Will fail with ISE
+ *     }
  * }
  * ```
  * If you want to switch the context where this flow is executed use [flowOn] operator.
@@ -47,7 +47,7 @@ import kotlin.jvm.*
 public fun <T> flow(@BuilderInference block: suspend FlowCollector<in T>.() -> Unit): Flow<T> {
     return object : Flow<T> {
         override suspend fun collect(collector: FlowCollector<in T>) {
-            SafeCollector(collector, coroutineContext[ContinuationInterceptor]).block()
+            SafeCollector(collector, coroutineContext).block()
         }
     }
 }

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

@@ -7,30 +7,29 @@ package kotlinx.coroutines.flow
 import kotlinx.coroutines.*
 
 /**
- * A cold asynchronous stream of the data, that emits from zero to N (where N can be unbounded)
- * values and completes normally or with an exception.
+ * A cold asynchronous stream of the data, that emits from zero to N (where N can be unbounded) values and completes normally or with an exception.
  *
- * All transformations on the flow, such as [map] and [filter] do not trigger flow collection or execution, only
- * terminal operators (e.g. [single])do trigger it.
+ * All transformations on the flow, such as [map] and [filter] do not trigger flow collection or execution, only terminal operators (e.g. [single]) do trigger it.
  *
- * Flow can be collected in a suspending manner, without actual blocking using [collect] extension that will complete normally or exceptionally:
+ * Flow can be collected in a suspending manner, without actual blocking, using [collect] extension that will complete normally or exceptionally:
  * ```
  * try {
- *   flow.collect { value ->
- *     println("Received $value")
- *   }
+ *     flow.collect { value ->
+ *         println("Received $value")
+ *     }
  * } catch (e: Exception) {
- *    println("Flow has thrown an exception: $e")
+ *     println("Flow has thrown an exception: $e")
  * }
  * ```
  * Additionally, the library provides a rich set of terminal operators such as [single], [reduce] and others.
  *
  * Flow does not carry information whether it is a cold stream (that can be collected multiple times and
- * triggers its evaluation every time collection is executed) or hot one, but conventionally flow represents a cold stream.
- * Transitions between hot and cold streams are supported via channels and corresponding API: [flowViaChannel], [broadcastIn], [produceIn].
+ * triggers its evaluation every time [collect] is executed) or a hot one, but conventionally flow represents a cold stream.
+ * Transitions between hot and cold streams are supported via channels and the corresponding API: [flowViaChannel], [broadcastIn], [produceIn].
  *
- * Flow is a **pure** concept: it encapsulates its own execution context and never propagates it to the downstream, thus making
+ * Flow has a context preserving property: it encapsulates its own execution context and never propagates or leaks it to the downstream, thus making
  * reasoning about execution context of particular transformations or terminal operations trivial.
+ *
  * There are two ways of changing the flow's context: [flowOn][Flow.flowOn] and [flowWith][Flow.flowWith].
  * The former changes the upstream context ("everything above the flowOn operator") while the latter
  * changes the context of the flow within [flowWith] body. For additional information refer to these operators documentation.
@@ -41,10 +40,10 @@ import kotlinx.coroutines.*
  *     .map { it + 1 } // Will be executed in ctx_1
  *     .flowOn(ctx_1) // Changes upstream context: flowOf and map
  *
- * // Now we have flow that is pure: it is executed somewhere but this information is encapsulated in the flow itself
+ * // Now we have flow that is context-preserving: it is executed somewhere but this information is encapsulated in the flow itself
  *
- * val filtered = flow
- *  .filter { it == 3 } // Pure operator without a context
+ * val filtered = flow // ctx_1 is inaccessible
+ *    .filter { it == 3 } // Pure operator without a context yet
  *
  * withContext(Dispatchers.Main) {
  *     // All not encapsulated operators will be executed in Main: filter and single
@@ -53,6 +52,29 @@ import kotlinx.coroutines.*
  * }
  * ```
  *
+ * From the implementation point of view it means that all intermediate operators on [Flow] should use the following constraint:
+ * If one wants to separate collection or emission into multiple coroutines, it should use [coroutineScope] or [supervisorScope] and
+ * is not allowed to modify coroutines context:
+ * ```
+ * fun <T> Flow<T>.buffer(bufferSize: Int): Flow<T> = flow {
+ *     coroutineScope { // coroutine scope is necessary, withContext is prohibited
+ *         val channel = Channel<T>(bufferSize)
+ *         // GlobalScope.launch { is prohibited
+ *         // launch(Dispatchers.IO) { is prohibited
+ *         launch { // is OK
+ *             collect { value ->
+ *                 channel.send(value)
+ *             }
+ *             channel.close()
+ *         }
+ *
+ *         for (i in channel) {
+ *             emit(i)
+ *         }
+ *     }
+ * }
+ * ```
+ *
  * 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.
  */

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

@@ -11,18 +11,98 @@ import kotlin.coroutines.*
  * search for their favourite operators and/or patterns that are missing or renamed in Flow.
  */
 
-/** @suppress **/
-@Deprecated(message = "Use flowWith or flowOn instead", level = DeprecationLevel.ERROR)
-public fun <T> Flow<T>.subscribeOn(context: CoroutineContext): Flow<T> = error("Should not be called")
-
-/** @suppress **/
-@Deprecated(message = "Use flowWith or flowOn instead", level = DeprecationLevel.ERROR)
+/**
+ * `observeOn` has no direct match in [Flow] API because all terminal flow operators are suspending and
+ * thus use the context of the caller.
+ *
+ * For example, the following code:
+ * ```
+ * flowable
+ *     .observeOn(Schedulers.io())
+ *     .doOnEach { value -> println("Received $value") }
+ *     .subscribe()
+ * ```
+ *
+ *  has the following Flow equivalent:
+ * ```
+ * withContext(Dispatchers.IO) {
+ *     flow.collect { value -> println("Received $value") }
+ * }
+ *
+ * ```
+ * @suppress
+ */
+@Deprecated(message = "Collect flow in the desired context instead", level = DeprecationLevel.ERROR)
 public fun <T> Flow<T>.observeOn(context: CoroutineContext): Flow<T> = error("Should not be called")
 
-/** @suppress **/
-@Deprecated(message = "Use flowWith or flowOn instead", level = DeprecationLevel.ERROR)
+/**
+ * `publishOn` has no direct match in [Flow] API because all terminal flow operators are suspending and
+ * thus use the context of the caller.
+ *
+ * For example, the following code:
+ * ```
+ * flux
+ *     .publishOn(Schedulers.io())
+ *     .doOnEach { value -> println("Received $value") }
+ *     .subscribe()
+ * ```
+ *
+ *  has the following Flow equivalent:
+ * ```
+ * withContext(Dispatchers.IO) {
+ *     flow.collect { value -> println("Received $value") }
+ * }
+ *
+ * ```
+ * @suppress
+ */
+@Deprecated(message = "Collect flow in the desired context instead", level = DeprecationLevel.ERROR)
 public fun <T> Flow<T>.publishOn(context: CoroutineContext): Flow<T> = error("Should not be called")
 
+/**
+ * `subscribeOn` has no direct match in [Flow] API because [Flow] preserves its context and does not leak it.
+ *
+ * For example, the following code:
+ * ```
+ * flowable
+ *     .map { value -> println("Doing map in IO"); value }
+ *     .subscribeOn(Schedulers.io())
+ *     .observeOn(Schedulers.computation())
+ *     .doOnEach { value -> println("Processing $value in computation")
+ *     .subscribe()
+ * ```
+ * has the following Flow equivalents:
+ * ```
+ * withContext(Dispatchers.Default) {
+ *     flow
+ *        .map { value -> println("Doing map in IO"); value }
+ *        .flowOn(Dispatchers.IO) // Works upstream, doesn't change downstream
+ *        .collect { value ->
+ *             println("Processing $value in computation")
+ *        }
+ * }
+ * ```
+ * or
+ *
+ * ```
+ *  withContext(Dispatchers.Default) {
+ *     flow
+ *        .flowWith(Dispatchers.IO) { map { value -> println("Doing map in IO"); value } }
+ *        .collect { value ->
+ *             println("Processing $value in computation")
+ *        }
+ * }
+ * ```
+ *
+ * The difference is that [flowWith] encapsulates ("preserves") the context within its lambda
+ * while [flowOn] changes the context of all preceding operators.
+ * Opposed to subscribeOn, it it **possible** to use multiple `flowOn` operators in the one flow.
+ *
+ * @suppress
+ */
+@Deprecated(message = "Use flowWith or flowOn instead", level = DeprecationLevel.ERROR)
+public fun <T> Flow<T>.subscribeOn(context: CoroutineContext): Flow<T> = error("Should not be called")
+
 /** @suppress **/
 @Deprecated(message = "Use BroadcastChannel.asFlow()", level = DeprecationLevel.ERROR)
 public fun BehaviourSubject(): Any = error("Should not be called")
@@ -45,14 +125,40 @@ public fun PublishSubject(): Any = error("Should not be called")
 )
 public fun <T> Flow<T>.onErrorResume(fallback: Flow<T>): Flow<T> = error("Should not be called")
 
-
-/** @suppress **/
+/**
+ * Self-explanatory, the reason of deprecation is "context preservation" property (you can read more in [Flow] documentation)
+ * @suppress
+ **/
 @Suppress("UNUSED_PARAMETER", "UNUSED", "DeprecatedCallableAddReplaceWith")
 @Deprecated(message = "withContext in flow body is deprecated, use flowOn instead", level = DeprecationLevel.ERROR)
 public fun <T, R> FlowCollector<T>.withContext(context: CoroutineContext, block: suspend () -> R): Unit = error("Should not be called")
 
 
-/** @suppress **/
+/**
+ * `subscribe` is Rx-specific API that has no direct match in flows.
+ * One can use `launch` instead, for example the following:
+ * ```
+ * flowable
+ *     .observeOn(Schedulers.io())
+ *     .subscribe({ println("Received $it") }, { println("Exception $it happened") }, { println("Flowable is completed successfully") }
+ * ```
+ *
+ * has the following Flow equivalent:
+ * ```
+ * launch(Dispatchers.IO) {
+ *     try {
+ *         flow.collect { value ->
+ *             println("Received $value")
+ *         }
+ *         println("Flow is completed successfully")
+ *     } catch (e: Throwable) {
+ *         println("Exception $e happened")
+ *     }
+ * }
+ * ```
+ * But most of the time it is better to use terminal operators like [single] instead of [collect].
+ * @suppress
+ */
 @Deprecated(message = "Use launch + collect instead", level = DeprecationLevel.ERROR)
 public fun <T> Flow<T>.subscribe(): Unit = error("Should not be called")
 
@@ -64,7 +170,6 @@ public fun <T> Flow<T>.subscribe(onEach: (T) -> Unit): Unit = error("Should not
 @Deprecated(message = "Use launch + collect instead", level = DeprecationLevel.ERROR)
 public fun <T> Flow<T>.subscribe(onEach: (T) -> Unit, onError: (Throwable) -> Unit): Unit = error("Should not be called")
 
-
 /**
  * Note that this replacement is sequential (`concat`) by default.
  * For concurrent flatMap [flatMapMerge] can be used instead.
@@ -85,7 +190,6 @@ public fun <T, R> Flow<T>.flatMap(mapper: suspend (T) -> Flow<R>): Flow<R> = err
 )
 public fun <T, R> Flow<T>.concatMap(mapper: (T) -> Flow<R>): Flow<R> = error("Should not be called")
 
-
 /**
  * Note that this replacement is sequential (`concat`) by default.
  * For concurrent flatMap [flattenMerge] can be used instead.

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

@@ -4,22 +4,25 @@
 
 package kotlinx.coroutines.flow.internal
 
+import kotlinx.coroutines.*
 import kotlinx.coroutines.flow.*
 import kotlinx.coroutines.internal.*
 import kotlin.coroutines.*
 
 @PublishedApi
 internal class SafeCollector<T>(
     private val collector: FlowCollector<T>,
-    private val interceptor: ContinuationInterceptor?
+    collectContext: CoroutineContext
 ) : FlowCollector<T>, SynchronizedObject() {
 
+    private val collectContext = collectContext.minusKey(Job).minusId()
+
     override suspend fun emit(value: T)  {
-        if (interceptor != coroutineContext[ContinuationInterceptor]) {
+        val emitContext = coroutineContext.minusKey(Job).minusId()
+        if (emitContext != collectContext) {
             error(
-                "Flow invariant is violated: flow was collected in $interceptor, but emission happened in ${coroutineContext[ContinuationInterceptor]}. " +
-                        "Please refer to 'flow' documentation or use 'flowOn' instead"
-            )
+                "Flow invariant is violated: flow was collected in $collectContext, but emission happened in $emitContext. " +
+                        "Please refer to 'flow' documentation or use 'flowOn' instead")
         }
         collector.emit(value)
     }