Report exception on cancellation in AbstractContinuation, do not swallow exceptions on cancellation in rx1

Author: qwwdfsad

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

@@ -277,12 +277,16 @@ internal abstract class AbstractContinuation<in T>(
                 }
                 is CancelledContinuation -> {
                     /*
-                     * If continuation was cancelled, then all further updates (resumes or exceptions) must be
-                     * ignored, because cancellation is asynchronous and may race with resume/resumeWithException.
-                     * This race is normal.
+                     * If continuation was cancelled, then all further resumes must be
+                     * ignored, because cancellation is asynchronous and may race with resume.
+                     * Racy exception are reported so no exceptions are lost
                      *
                      * :todo: we should somehow remember the attempt to invoke resume and fail on the second attempt.
                      */
+                    if (proposedUpdate is CompletedExceptionally) {
+                        handleException(proposedUpdate.cause)
+                    }
+
                     return
                 }
                 else -> error("Already resumed, but proposed with update $proposedUpdate")

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

@@ -1,6 +1,7 @@
 /*
  * Copyright 2016-2018 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license.
  */
+@file:Suppress("NAMED_ARGUMENTS_NOT_ALLOWED") // KT-21913
 
 package kotlinx.coroutines.experimental
 
@@ -71,7 +72,7 @@ class CancellableContinuationTest : TestBase() {
      * should be ignored. Here suspended coroutine is cancelled but then resumed with exception.
      */
     @Test
-    fun testCancelAndResumeWithException() = runTest {
+    fun testCancelAndResumeWithException() = runTest(unhandled = listOf({e -> e is TestException})) {
         var continuation: Continuation<Unit>? = null
         val job = launch(coroutineContext) {
             try {

reactive/kotlinx-coroutines-reactive/test/IntegrationTest.kt

@@ -32,7 +32,7 @@ class IntegrationTest(
         @JvmStatic
         fun params(): Collection<Array<Any>> = Ctx.values().flatMap { ctx ->
             listOf(false, true).map { delay ->
-                arrayOf<Any>(ctx, delay)
+                arrayOf(ctx, delay)
             }
         }
     }
@@ -95,6 +95,34 @@ class IntegrationTest(
         channel.cancel()
     }
 
+    @Test
+    fun testCancelWithoutValue() = runTest {
+        val job = launch(coroutineContext, parent = Job(), start = CoroutineStart.UNDISPATCHED) {
+            publish<String>(coroutineContext) {
+                yield()
+                expectUnreached()
+            }.awaitFirst()
+        }
+
+        job.cancel()
+        job.join()
+    }
+
+    @Test
+    fun testEmptySingle() = runTest(unhandled = listOf({e -> e is NoSuchElementException})) {
+        expect(1)
+        val job = launch(coroutineContext, parent = Job(), start = CoroutineStart.UNDISPATCHED) {
+            publish<String>(coroutineContext) {
+                yield()
+                expect(2)
+                // Nothing to emit
+            }.awaitFirst()
+        }
+
+        job.join()
+        finish(3)
+    }
+
     private suspend fun checkNumbers(n: Int, pub: Publisher<Int>) {
         var last = 0
         pub.consumeEach {

reactive/kotlinx-coroutines-rx1/src/RxAwait.kt

@@ -120,7 +120,17 @@ private suspend fun <T> Observable<T>.awaitOne(): T = suspendCancellableCoroutin
         override fun onStart() { request(1) }
         override fun onNext(t: T) { cont.resume(t) }
         override fun onCompleted() { if (cont.isActive) cont.resumeWithException(IllegalStateException("Should have invoked onNext")) }
-        override fun onError(e: Throwable) { cont.resumeWithException(e) }
+        override fun onError(e: Throwable) {
+            /*
+             * Rx1 observable throws NoSuchElementException if cancellation happened before
+             * element emission. To mitigate this we try to atomically resume continuation with exception:
+             * if resume failed, then we know that continuation successfully cancelled itself
+             */
+            val token = cont.tryResumeWithException(e)
+            if (token != null) {
+                cont.completeResume(token)
+            }
+        }
     }))
 }
 

reactive/kotlinx-coroutines-rx1/test/IntegrationTest.kt

@@ -32,7 +32,7 @@ class IntegrationTest(
         @JvmStatic
         fun params(): Collection<Array<Any>> = Ctx.values().flatMap { ctx ->
             listOf(false, true).map { delay ->
-                arrayOf<Any>(ctx, delay)
+                arrayOf(ctx, delay)
             }
         }
     }
@@ -117,6 +117,34 @@ class IntegrationTest(
         channel.cancel()
     }
 
+    @Test
+    fun testCancelWithoutValue() = runTest {
+        val job = launch(coroutineContext, parent = Job(), start = CoroutineStart.UNDISPATCHED) {
+            rxObservable<String>(coroutineContext) {
+                yield()
+                expectUnreached()
+            }.awaitFirst()
+        }
+
+        job.cancel()
+        job.join()
+    }
+
+    @Test
+    fun testEmptySingle() = runTest(unhandled = listOf({e -> e is NoSuchElementException})) {
+        expect(1)
+        val job = launch(coroutineContext, parent = Job(), start = CoroutineStart.UNDISPATCHED) {
+            rxObservable<String>(coroutineContext) {
+                yield()
+                expect(2)
+                // Nothing to emit
+            }.awaitFirst()
+        }
+
+        job.join()
+        finish(3)
+    }
+
     private suspend fun checkNumbers(n: Int, observable: Observable<Int>) {
         var last = 0
         observable.consumeEach {

reactive/kotlinx-coroutines-rx2/test/IntegrationTest.kt

@@ -32,7 +32,7 @@ class IntegrationTest(
         @JvmStatic
         fun params(): Collection<Array<Any>> = Ctx.values().flatMap { ctx ->
             listOf(false, true).map { delay ->
-                arrayOf<Any>(ctx, delay)
+                arrayOf(ctx, delay)
             }
         }
     }
@@ -97,6 +97,34 @@ class IntegrationTest(
         channel.cancel()
     }
 
+    @Test
+    fun testCancelWithoutValue() = runTest {
+        val job = launch(coroutineContext, parent = Job(), start = CoroutineStart.UNDISPATCHED) {
+            rxObservable<String>(coroutineContext) {
+                yield()
+                expectUnreached()
+            }.awaitFirst()
+        }
+
+        job.cancel()
+        job.join()
+    }
+
+    @Test
+    fun testEmptySingle() = runTest(unhandled = listOf({e -> e is NoSuchElementException})) {
+        expect(1)
+        val job = launch(coroutineContext, parent = Job(), start = CoroutineStart.UNDISPATCHED) {
+            rxObservable<String>(coroutineContext) {
+                yield()
+                expect(2)
+                // Nothing to emit
+            }.awaitFirst()
+        }
+
+        job.join()
+        finish(3)
+    }
+
     private suspend fun checkNumbers(n: Int, observable: Observable<Int>) {
         var last = 0
         observable.consumeEach {