Fix ensureAction to work in the empty context case (#2119)

Fixes #2044

Author: elizarov

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

@@ -587,10 +587,10 @@ public fun Job.ensureActive(): Unit {
 
 /**
  * Ensures that job in the current context is [active][Job.isActive].
- * Throws [IllegalStateException] if the context does not have a job in it.
  *
  * If the job is no longer active, throws [CancellationException].
  * If the job was cancelled, thrown exception contains the original cancellation cause.
+ * This function does not do anything if there is no [Job] in the context, since such a coroutine cannot be cancelled.
  *
  * This method is a drop-in replacement for the following code, but with more precise exception:
  * ```
@@ -599,9 +599,8 @@ public fun Job.ensureActive(): Unit {
  * }
  * ```
  */
-public fun CoroutineContext.ensureActive(): Unit {
-    val job = get(Job) ?: error("Context cannot be checked for liveness because it does not have a job: $this")
-    job.ensureActive()
+public fun CoroutineContext.ensureActive() {
+    get(Job)?.ensureActive()
 }
 
 /**

kotlinx-coroutines-core/common/test/EmptyContext.kt

@@ -0,0 +1,16 @@
+/*
+ * Copyright 2016-2020 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license.
+ */
+
+package kotlinx.coroutines
+
+import kotlinx.coroutines.intrinsics.*
+import kotlin.coroutines.*
+
+suspend fun <T> withEmptyContext(block: suspend () -> T): T {
+    val baseline = Result.failure<T>(IllegalStateException("Block was suspended"))
+    var result: Result<T> = baseline
+    block.startCoroutineUnintercepted(Continuation(EmptyCoroutineContext) { result = it })
+    while (result == baseline) yield()
+    return result.getOrThrow()
+}

kotlinx-coroutines-core/common/test/EnsureActiveTest.kt

@@ -68,6 +68,13 @@ class EnsureActiveTest : TestBase() {
         finish(4)
     }
 
+    @Test
+    fun testEnsureActiveWithEmptyContext() = runTest {
+        withEmptyContext {
+            ensureActive() // should not do anything
+        }
+    }
+
     private inline fun checkException(block: () -> Unit) {
         val result = runCatching(block)
         val exception = result.exceptionOrNull() ?: fail()

kotlinx-coroutines-core/common/test/flow/FlowInvariantsTest.kt

@@ -6,7 +6,6 @@ package kotlinx.coroutines.flow
 
 import kotlinx.coroutines.*
 import kotlinx.coroutines.channels.*
-import kotlinx.coroutines.intrinsics.*
 import kotlin.coroutines.*
 import kotlin.reflect.*
 import kotlin.test.*
@@ -243,16 +242,8 @@ class FlowInvariantsTest : TestBase() {
             return result
         }
 
-        val result = runSuspendFun { collector() }
+        val result = withEmptyContext { collector() }
         assertEquals(2, result)
         finish(3)
     }
-
-    private suspend fun runSuspendFun(block: suspend () -> Int): Int {
-        val baseline = Result.failure<Int>(IllegalStateException("Block was suspended"))
-        var result: Result<Int> = baseline
-        block.startCoroutineUnintercepted(Continuation(EmptyCoroutineContext) { result = it })
-        while (result == baseline) yield()
-        return result.getOrThrow()
-    }
 }

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

@@ -43,4 +43,20 @@ class FlowCancellationTest : TestBase() {
         job.cancelAndJoin()
         finish(3)
     }
+
+    @Test
+    fun testFlowWithEmptyContext() = runTest {
+        expect(1)
+        withEmptyContext {
+            val flow = flow {
+                expect(2)
+                emit("OK")
+            }
+            flow.collect {
+                expect(3)
+                assertEquals("OK", it)
+            }
+        }
+        finish(4)
+    }
 }