Fixed aggregated error handling in Publish & test for it

Author: elizarov

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

@@ -199,7 +199,9 @@ private class PublisherCoroutine<in T>(
         }
     }
 
-    override fun onCancellation(cause: Throwable?) {
+    override fun onCompletedExceptionally(exception: Throwable) = onCompleted(Unit)
+
+    override fun onCompleted(value: Unit) {
         while (true) { // lock-free loop for nRequested
             val cur = _nRequested.value
             if (cur == SIGNALLED) return // some other thread holding lock already signalled cancellation/completion

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

@@ -90,4 +90,44 @@ class PublishTest : TestBase() {
             throw RuntimeException("OK")
         }.openSubscription()
     }
+
+    @Test
+    fun testHandleFailureAfterCancel() = runTest(
+        unhandled = listOf({ it -> it is RuntimeException && it.message == "FAILED" })
+    ){
+        expect(1)
+        // Exception should be delivered to CoroutineExceptionHandler, because we create publisher
+        // with the NonCancellable parent
+        val publisher = publish<Unit>(NonCancellable + Dispatchers.Unconfined) {
+            try {
+                expect(3)
+                delay(10000)
+            } finally {
+                expect(5)
+                throw RuntimeException("FAILED") // crash after cancel
+            }
+        }
+        var sub: Subscription? = null
+        publisher.subscribe(object : Subscriber<Unit> {
+            override fun onComplete() {
+                expectUnreached()
+            }
+
+            override fun onSubscribe(s: Subscription) {
+                expect(2)
+                sub = s
+            }
+
+            override fun onNext(t: Unit?) {
+                expectUnreached()
+            }
+
+            override fun onError(t: Throwable?) {
+                expectUnreached()
+            }
+        })
+        expect(4)
+        sub!!.cancel()
+        finish(6)
+    }
 }