~rename currentContext to currentCoroutineContext, add fusion

Author: qwwdfsad

kotlinx-coroutines-core/api/kotlinx-coroutines-core.api

@@ -214,7 +214,7 @@ public final class kotlinx/coroutines/CoroutineScopeKt {
 	public static synthetic fun cancel$default (Lkotlinx/coroutines/CoroutineScope;Ljava/lang/String;Ljava/lang/Throwable;ILjava/lang/Object;)V
 	public static synthetic fun cancel$default (Lkotlinx/coroutines/CoroutineScope;Ljava/util/concurrent/CancellationException;ILjava/lang/Object;)V
 	public static final fun coroutineScope (Lkotlin/jvm/functions/Function2;Lkotlin/coroutines/Continuation;)Ljava/lang/Object;
-	public static final fun currentContext (Lkotlin/coroutines/Continuation;)Ljava/lang/Object;
+	public static final fun currentCoroutineContext (Lkotlin/coroutines/Continuation;)Ljava/lang/Object;
 	public static final fun ensureActive (Lkotlinx/coroutines/CoroutineScope;)V
 	public static final fun isActive (Lkotlinx/coroutines/CoroutineScope;)Z
 	public static final fun plus (Lkotlinx/coroutines/CoroutineScope;Lkotlin/coroutines/CoroutineContext;)Lkotlinx/coroutines/CoroutineScope;

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

@@ -242,10 +242,10 @@ public fun CoroutineScope.ensureActive(): Unit = coroutineContext.ensureActive()
  * ```
  * launch { // this: CoroutineScope
  *     val flow = flow<Unit> {
- *         coroutineContext // Resolves into the context of outer launch, which is incorrect, see KT
- *         currentContext() // Retrieves actual context whe the flow is collected
+ *         coroutineContext // Resolves into the context of outer launch, which is incorrect, see KT-38033
+ *         currentCoroutineContext() // Retrieves actual context where the flow is collected
  *     }
  * }
  * ```
  */
-public suspend inline fun currentContext(): CoroutineContext = coroutineContext
+public suspend inline fun currentCoroutineContext(): CoroutineContext = coroutineContext

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

@@ -50,14 +50,9 @@ import kotlinx.coroutines.flow.internal.unsafeFlow as flow
 public fun <T> flow(@BuilderInference block: suspend FlowCollector<T>.() -> Unit): Flow<T> = SafeFlow(block)
 
 // Named anonymous object
-private class SafeFlow<T>(private val block: suspend FlowCollector<T>.() -> Unit) : Flow<T> {
-    override suspend fun collect(collector: FlowCollector<T>) {
-        val safeCollector = SafeCollector(collector, coroutineContext)
-        try {
-            safeCollector.block()
-        } finally {
-            safeCollector.releaseIntercepted()
-        }
+private class SafeFlow<T>(private val block: suspend FlowCollector<T>.() -> Unit) : AbstractFlow<T>() {
+    override suspend fun collectSafely(collector: FlowCollector<T>) {
+        collector.block()
     }
 }
 

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

@@ -213,10 +213,13 @@ public fun <T> Flow<T>.flowOn(context: CoroutineContext): Flow<T> {
  * the corresponding cancellation cause if flow collector was cancelled.
  * Note that [flow] builder is [cancellable] by default.
  */
-public fun <T> Flow<T>.cancellable(): Flow<T> = unsafeFlow {
-    collect {
-        currentContext().ensureActive()
-        emit(it)
+public fun <T> Flow<T>.cancellable(): Flow<T> {
+    if (this is AbstractFlow<*>) return this // Fast-path, already cancellable
+    return unsafeFlow {
+        collect {
+            currentCoroutineContext().ensureActive()
+            emit(it)
+        }
     }
 }
 
@@ -268,7 +271,7 @@ public fun <T, R> Flow<T>.flowWith(
          * All builders are written using scoping and no global coroutines are launched, so it is safe not to provide explicit Job.
          * It is also necessary not to mess with cancellation if multiple flowWith are used.
          */
-        val originalContext = currentContext().minusKey(Job)
+        val originalContext = currentCoroutineContext().minusKey(Job)
         val prepared = source.flowOn(originalContext).buffer(bufferSize)
         builder(prepared).flowOn(flowContext).buffer(bufferSize).collect { value ->
             return@collect emit(value)

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

@@ -10,7 +10,6 @@ package kotlinx.coroutines.flow
 
 import kotlinx.coroutines.*
 import kotlinx.coroutines.flow.internal.*
-import kotlin.coroutines.*
 import kotlin.jvm.*
 
 // ------------------ WARNING ------------------
@@ -70,7 +69,7 @@ internal inline fun <T, R> Flow<T>.unsafeTransform(
 public fun <T> Flow<T>.onStart(
     action: suspend FlowCollector<T>.() -> Unit
 ): Flow<T> = unsafeFlow { // Note: unsafe flow is used here, but safe collector is used to invoke start action
-    val safeCollector = SafeCollector<T>(this, currentContext())
+    val safeCollector = SafeCollector<T>(this, currentCoroutineContext())
     try {
         safeCollector.action()
     } finally {
@@ -153,7 +152,7 @@ public fun <T> Flow<T>.onCompletion(
         throw e
     }
     // Normal completion
-    SafeCollector(this, currentContext()).invokeSafely(action, null)
+    SafeCollector(this, currentCoroutineContext()).invokeSafely(action, null)
 }
 
 /**
@@ -178,7 +177,7 @@ public fun <T> Flow<T>.onEmpty(
         emit(it)
     }
     if (isEmpty) {
-        val collector = SafeCollector(this, currentContext())
+        val collector = SafeCollector(this, currentCoroutineContext())
         try {
             collector.action()
         } finally {

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

@@ -45,28 +45,28 @@ public fun <T> StateFlow<T>.distinctUntilChanged(): Flow<T> = noImpl()
 
 @Deprecated(
     message = "isActive is resolved into the extension of outer CoroutineScope which is likely to be an error." +
-        "Use currentContext().isActive or cancellable() operator instead " +
+        "Use currentCoroutineContext().isActive or cancellable() operator instead " +
         "or specify the receiver of isActive explicitly. " +
         "Additionally, flow {} builder emissions are cancellable by default.",
     level = DeprecationLevel.WARNING, // ERROR in 1.4
-    replaceWith = ReplaceWith("currentContext().isActive")
+    replaceWith = ReplaceWith("currentCoroutineContext().isActive")
 )
 public val FlowCollector<*>.isActive: Boolean
     get() = noImpl()
 
 @Deprecated(
     message = "cancel() is resolved into the extension of outer CoroutineScope which is likely to be an error." +
-        "Use currentContext().cancel() instead or specify the receiver of cancel() explicitly",
+        "Use currentCoroutineContext().cancel() instead or specify the receiver of cancel() explicitly",
     level = DeprecationLevel.WARNING, // ERROR in 1.4
-    replaceWith = ReplaceWith("currentContext().cancel(cause)")
+    replaceWith = ReplaceWith("currentCoroutineContext().cancel(cause)")
 )
 public fun FlowCollector<*>.cancel(cause: CancellationException? = null): Unit = noImpl()
 
 @Deprecated(
     message = "coroutineContext is resolved into the property of outer CoroutineScope which is likely to be an error." +
-        "Use currentContext() instead or specify the receiver of coroutineContext explicitly",
+        "Use currentCoroutineContext() instead or specify the receiver of coroutineContext explicitly",
     level = DeprecationLevel.WARNING, // ERROR in 1.4
-    replaceWith = ReplaceWith("currentContext()")
+    replaceWith = ReplaceWith("currentCoroutineContext()")
 )
 public val FlowCollector<*>.coroutineContext: CoroutineContext
     get() = noImpl()

kotlinx-coroutines-core/common/test/flow/operators/CancellableTest.kt

@@ -15,7 +15,7 @@ class CancellableTest : TestBase() {
         var sum = 0
         val flow = (0..1000).asFlow()
             .onEach {
-                if (it != 0) currentContext().cancel()
+                if (it != 0) currentCoroutineContext().cancel()
                 sum += it
             }
 
@@ -26,4 +26,13 @@ class CancellableTest : TestBase() {
         flow.cancellable().launchIn(this).join()
         assertEquals(1, sum)
     }
-}
+
+    @Test
+    fun testFastPath() {
+        val flow = listOf(1).asFlow()
+        assertNotSame(flow, flow.cancellable())
+
+        val cancellableFlow = flow { emit(42) }
+        assertSame(cancellableFlow, cancellableFlow.cancellable())
+    }
+}

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

@@ -18,7 +18,7 @@ internal actual class SafeCollector<T> actual constructor(
     private var lastEmissionContext: CoroutineContext? = null
 
     override suspend fun emit(value: T) {
-        val currentContext = currentContext()
+        val currentContext = currentCoroutineContext()
         currentContext.ensureActive()
         if (lastEmissionContext !== currentContext) {
             checkContext(currentContext)

kotlinx-coroutines-core/jvm/test/flow/FlowCancellationTest.kt

@@ -33,7 +33,7 @@ class FlowCancellationTest : TestBase() {
         val job = flow<Unit> {
             expect(1)
             latch.send(Unit)
-            while (currentContext().isActive) {
+            while (currentCoroutineContext().isActive) {
                 // Do nothing
             }
         }.launchIn(this + Dispatchers.Default)

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

@@ -18,7 +18,7 @@ internal actual class SafeCollector<T> actual constructor(
     private var lastEmissionContext: CoroutineContext? = null
 
     override suspend fun emit(value: T) {
-        val currentContext = currentContext()
+        val currentContext = currentCoroutineContext()
         currentContext.ensureActive()
         if (lastEmissionContext !== currentContext) {
             checkContext(currentContext)

reactive/kotlinx-coroutines-reactor/test/BackpressureTest.kt

@@ -41,13 +41,13 @@ class BackpressureTest : TestBase() {
     @Test
     fun testCooperativeCancellation() = runTest {
         val flow = Flux.fromIterable((0L..Long.MAX_VALUE)).asFlow()
-        flow.onEach { if (it > 10) currentContext().cancel() }.launchIn(this + Dispatchers.Default).join()
+        flow.onEach { if (it > 10) currentCoroutineContext().cancel() }.launchIn(this + Dispatchers.Default).join()
     }
 
     @Test
     fun testCooperativeCancellationForBuffered() = runTest(expected = { it is CancellationException }) {
         val flow = Flux.fromIterable((0L..Long.MAX_VALUE)).asFlow()
-        val channel = flow.onEach { if (it > 10) currentContext().cancel() }.produceIn(this + Dispatchers.Default)
+        val channel = flow.onEach { if (it > 10) currentCoroutineContext().cancel() }.produceIn(this + Dispatchers.Default)
         channel.consumeEach { /* Do nothing, just consume elements */ }
     }
 }