Report fatal errors to both onError (to comply the spec fully) and to coroutine context (to eagerly fail as fatal exceptions are not recoverable anyway), cleanup tests after scopeless reactive

Author: qwwdfsad

kotlinx-coroutines-core/common/test/flow/terminal/SingleTest.kt

@@ -7,7 +7,7 @@ package kotlinx.coroutines.flow
 import kotlinx.coroutines.*
 import kotlin.test.*
 
-class SingleTest : TestBase() {
+class SingleTest : TestBase() { 
 
     @Test
     fun testSingle() = runTest {

reactive/kotlinx-coroutines-reactive/src/Publish.kt

@@ -163,13 +163,13 @@ private class PublisherCoroutine<in T>(
             val current = _nRequested.value
             if (current < 0) break // closed from inside onNext => unlock
             if (current == Long.MAX_VALUE) break // no back-pressure => unlock
-            val upd = current - 1
-            if (_nRequested.compareAndSet(current, upd)) {
-                if (upd == 0L) {
+            val updated = current - 1
+            if (_nRequested.compareAndSet(current, updated)) {
+                if (updated == 0L) {
                     // return to keep locked due to back-pressure
                     return
                 }
-                break // unlock if upd > 0
+                break // unlock if updated > 0
             }
         }
         unlockAndCheckCompleted()
@@ -206,18 +206,17 @@ private class PublisherCoroutine<in T>(
                         /*
                          * Reactive frameworks have two types of exceptions: regular and fatal.
                          * Regular are passed to onError.
-                         * Fatal can be passed to onError, but implementation **is free to swallow it** (e.g. see #1297).
+                         * Fatal can be passed to onError, but even the standard implementations **can just swallow it** (e.g. see #1297).
                          * Such behaviour is inconsistent, leads to silent failures and we can't possibly know whether
                          * the cause will be handled by onError (and moreover, it depends on whether a fatal exception was
                          * thrown by subscriber or upstream).
                          * To make behaviour consistent and least surprising, we always handle fatal exceptions
                          * by coroutines machinery, anyway, they should not be present in regular program flow,
                          * thus our goal here is just to expose it as soon as possible.
                          */
-                        if (cause.isFatal()) {
-                            if (!handled) handleCoroutineException(context, cause)
-                        } else {
-                            subscriber.onError(cause)
+                        subscriber.onError(cause)
+                        if (!handled && cause.isFatal()) {
+                            handleCoroutineException(context, cause)
                         }
                     } else {
                         subscriber.onComplete()

reactive/kotlinx-coroutines-rx2/src/RxObservable.kt

@@ -7,6 +7,7 @@
 package kotlinx.coroutines.rx2
 
 import io.reactivex.*
+import io.reactivex.exceptions.*
 import kotlinx.atomicfu.*
 import kotlinx.coroutines.*
 import kotlinx.coroutines.channels.*
@@ -159,20 +160,19 @@ private class RxObservableCoroutine<T: Any>(
                 try {
                     if (cause != null && cause !is CancellationException) {
                         /*
-                        * Reactive frameworks have two types of exceptions: regular and fatal.
-                        * Regular are passed to onError.
-                        * Fatal can be passed to onError, but implementation **is free to swallow it** (e.g. see #1297).
-                        * Such behaviour is inconsistent, leads to silent failures and we can't possibly know whether
-                        * the cause will be handled by onError (and moreover, it depends on whether a fatal exception was
-                        * thrown by subscriber or upstream).
-                        * To make behaviour consistent and least surprising, we always handle fatal exceptions
-                        * by coroutines machinery, anyway, they should not be present in regular program flow,
-                        * thus our goal here is just to expose it as soon as possible.
-                        */
-                        if (cause.isFatal()) {
-                            if (!handled) handleCoroutineException(context, cause)
-                        } else {
-                            subscriber.onError(cause)
+                         * Reactive frameworks have two types of exceptions: regular and fatal.
+                         * Regular are passed to onError.
+                         * Fatal can be passed to onError, but even the standard implementations **can just swallow it** (e.g. see #1297).
+                         * Such behaviour is inconsistent, leads to silent failures and we can't possibly know whether
+                         * the cause will be handled by onError (and moreover, it depends on whether a fatal exception was
+                         * thrown by subscriber or upstream).
+                         * To make behaviour consistent and least surprising, we always handle fatal exceptions
+                         * by coroutines machinery, anyway, they should not be present in regular program flow,
+                         * thus our goal here is just to expose it as soon as possible.
+                         */
+                        subscriber.onError(cause)
+                        if (!handled && cause.isFatal()) {
+                            handleCoroutineException(context, cause)
                         }
                     }
                     else {
@@ -203,4 +203,9 @@ private class RxObservableCoroutine<T: Any>(
     }
 }
 
-internal fun Throwable.isFatal() = this is VirtualMachineError || this is ThreadDeath || this is LinkageError
+internal fun Throwable.isFatal() = try {
+    Exceptions.throwIfFatal(this) // Rx-consistent behaviour without hardcode
+    false
+} catch (e: Throwable) {
+    true
+}

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

@@ -143,7 +143,7 @@ class ConvertTest : TestBase() {
         val single = rxSingle(Dispatchers.Unconfined) {
             var result = ""
             try {
-                observable.consumeEach { result += it }
+                observable.collect { result += it }
             } catch(e: Throwable) {
                 check(e is TestException)
                 result += e.message

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

@@ -16,96 +16,42 @@ class FlowableExceptionHandlingTest : TestBase() {
         ignoreLostThreads("RxComputationThreadPool-", "RxCachedWorkerPoolEvictor-", "RxSchedulerPurge-")
     }
 
-    @Test
-    fun testException() = runTest(expected = { it is TestException }) {
-        rxFlowable<Int>(Dispatchers.Unconfined) {
-            expect(1)
-            throw TestException()
-        }.subscribe({
-            expectUnreached()
-        }, {
-            expect(2) // Reported to onError
-        })
-        finish(3)
+    private inline fun <reified T : Throwable> ceh(expect: Int) = CoroutineExceptionHandler { _, t ->
+        assertTrue(t is T)
+        expect(expect)
     }
 
-    @Test
-    fun testFatalException() = runTest(expected = { it is LinkageError }) {
-        rxFlowable<Int>(Dispatchers.Unconfined + CoroutineExceptionHandler { _, _ -> expectUnreached() }) {
-            expect(1)
-            throw LinkageError()
-        }.subscribe({
-            expectUnreached()
-        }, {
-            expectUnreached() // Fatal exception is not reported in onError
-        })
-        finish(2)
-    }
+    private fun cehUnreached() = CoroutineExceptionHandler { _, _ -> expectUnreached() }
 
     @Test
-    fun testExceptionWithoutParent() = runTest {
-        GlobalScope.rxFlowable<Int>(Dispatchers.Unconfined + CoroutineExceptionHandler { _, _ -> expectUnreached() }) {
+    fun testException() = runTest {
+        rxFlowable<Int>(Dispatchers.Unconfined + cehUnreached()) {
             expect(1)
             throw TestException()
         }.subscribe({
             expectUnreached()
         }, {
-            assertTrue(it is TestException)
             expect(2) // Reported to onError
         })
         finish(3)
     }
 
     @Test
-    fun testFatalExceptionWithoutParent() = runTest {
-        GlobalScope.rxFlowable<Int>(Dispatchers.Unconfined + CoroutineExceptionHandler { _, e ->
-            assertTrue(e is LinkageError); expect(
-            2
-        )
-        }) {
+    fun testFatalException() = runTest {
+        rxFlowable<Int>(Dispatchers.Unconfined + ceh<LinkageError>(3)) {
             expect(1)
             throw LinkageError()
         }.subscribe({
             expectUnreached()
         }, {
-            expectUnreached() // Fatal exception is not reported in onError
+            expect(2) // Fatal exception is reported to both onError and CEH
         })
-        finish(3)
-    }
-
-    @Test
-    fun testExceptionAsynchronous() = runTest(expected = { it is TestException }) {
-        rxFlowable<Int>(Dispatchers.Unconfined) {
-            expect(1)
-            throw TestException()
-        }.publish()
-            .refCount()
-            .subscribe({
-                expectUnreached()
-            }, {
-                expect(2) // Reported to onError
-            })
-        finish(3)
-    }
-
-    @Test
-    fun testFatalExceptionAsynchronous() = runTest(expected = { it is LinkageError }) {
-        rxFlowable<Int>(Dispatchers.Unconfined + CoroutineExceptionHandler { _, _ -> expectUnreached() }) {
-            expect(1)
-            throw LinkageError()
-        }.publish()
-            .refCount()
-            .subscribe({
-                expectUnreached()
-            }, {
-                expectUnreached() // Fatal exception is not reported in onError
-            })
-        finish(2)
+        finish(4)
     }
 
     @Test
-    fun testExceptionAsynchronousWithoutParent() = runTest {
-        GlobalScope.rxFlowable<Int>(Dispatchers.Unconfined + CoroutineExceptionHandler { _, _ -> expectUnreached() }) {
+    fun testExceptionAsynchronous() = runTest {
+        rxFlowable<Int>(Dispatchers.Unconfined + cehUnreached()) {
             expect(1)
             throw TestException()
         }.publish()
@@ -119,37 +65,35 @@ class FlowableExceptionHandlingTest : TestBase() {
     }
 
     @Test
-    fun testFatalExceptionAsynchronousWithoutParent() = runTest {
-        GlobalScope.rxFlowable<Int>(Dispatchers.Unconfined + CoroutineExceptionHandler { _, e ->
-            assertTrue(e is LinkageError); expect(2)
-        }) {
+    fun testFatalExceptionAsynchronous() = runTest {
+        rxFlowable<Int>(Dispatchers.Unconfined + ceh<LinkageError>(3)) {
             expect(1)
             throw LinkageError()
         }.publish()
             .refCount()
             .subscribe({
                 expectUnreached()
             }, {
-                expectUnreached() // Fatal exception is not reported in onError
+                expect(2)
             })
-        finish(3)
+        finish(4)
     }
 
     @Test
-    fun testFatalExceptionFromSubscribe() = runTest(expected = { it is LinkageError }) {
-        rxFlowable(Dispatchers.Unconfined) {
+    fun testFatalExceptionFromSubscribe() = runTest {
+        rxFlowable(Dispatchers.Unconfined + ceh<LinkageError>(4)) {
             expect(1)
             send(Unit)
         }.subscribe({
             expect(2)
             throw LinkageError()
-        }, { expectUnreached() }) // Unreached because fatal errors are rethrown
-        finish(3)
+        }, { expect(3) }) // Fatal exception is reported to both onError and CEH
+        finish(5)
     }
 
     @Test
     fun testExceptionFromSubscribe() = runTest {
-        rxFlowable(Dispatchers.Unconfined) {
+        rxFlowable(Dispatchers.Unconfined + cehUnreached()) {
             expect(1)
             send(Unit)
         }.subscribe({
@@ -161,7 +105,7 @@ class FlowableExceptionHandlingTest : TestBase() {
 
     @Test
     fun testAsynchronousExceptionFromSubscribe() = runTest {
-        rxFlowable(Dispatchers.Unconfined) {
+        rxFlowable(Dispatchers.Unconfined + cehUnreached()) {
             expect(1)
             send(Unit)
         }.publish()
@@ -174,8 +118,8 @@ class FlowableExceptionHandlingTest : TestBase() {
     }
 
     @Test
-    fun testAsynchronousFatalExceptionFromSubscribe() = runTest(expected = { it is LinkageError }) {
-        rxFlowable(Dispatchers.Unconfined) {
+    fun testAsynchronousFatalExceptionFromSubscribe() = runTest {
+        rxFlowable(Dispatchers.Unconfined + ceh<LinkageError>(3)) {
             expect(1)
             send(Unit)
         }.publish()
@@ -184,38 +128,6 @@ class FlowableExceptionHandlingTest : TestBase() {
                 expect(2)
                 throw LinkageError()
             }, { expectUnreached() })
-        finish(3)
+        finish(4)
     }
-
-    @Test
-    fun testAsynchronousExceptionFromSubscribeWithoutParent() =
-        runTest {
-            GlobalScope.rxFlowable(Dispatchers.Unconfined) {
-                expect(1)
-                send(Unit)
-            }.publish()
-                .refCount()
-                .subscribe({
-                    expect(2)
-                    throw RuntimeException()
-                }, { expect(3) })
-            finish(4)
-        }
-
-    @Test
-    fun testAsynchronousFatalExceptionFromSubscribeWithoutParent() =
-        runTest {
-            GlobalScope.rxFlowable(Dispatchers.Unconfined + CoroutineExceptionHandler { _, e ->
-                assertTrue(e is LinkageError); expect(3)
-            }) {
-                expect(1)
-                send(Unit)
-            }.publish()
-                .refCount()
-                .subscribe({
-                    expect(2)
-                    throw LinkageError()
-                }, { expectUnreached() })
-            finish(4)
-        }
 }