Publish: Proper handling of onNext exception and test for it

Author: elizarov

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

@@ -132,7 +132,15 @@ private class PublisherCoroutine<in T>(
             throw getCancellationException()
         }
         // notify subscriber
-        subscriber.onNext(elem)
+        try {
+            subscriber.onNext(elem)
+        } catch (e: Throwable) {
+            // If onNext fails with exception, then we cancel coroutine (with this exception) and then rethrow it
+            // to abort the corresponding send/offer invocation
+            cancel(e)
+            unlockAndCheckCompleted()
+            throw e
+        }
         // now update nRequested
         while (true) { // lock-free loop on nRequested
             val cur = _nRequested.value

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

@@ -168,5 +168,45 @@ class PublishTest : TestBase() {
         expectUnreached()
     }
 
+    @Test
+    fun testOnNextError() = runTest {
+        expect(1)
+        val publisher = publish<String>(NonCancellable) {
+            expect(4)
+            try {
+                send("OK")
+            } catch(e: Throwable) {
+                expect(6)
+                assert(e is TestException)
+            }
+        }
+        expect(2)
+        val latch = CompletableDeferred<Unit>()
+        publisher.subscribe(object : Subscriber<String> {
+            override fun onComplete() {
+                expectUnreached()
+            }
+
+            override fun onSubscribe(s: Subscription) {
+                expect(3)
+                s.request(1)
+            }
+
+            override fun onNext(t: String) {
+                expect(5)
+                assertEquals("OK", t)
+                throw TestException()
+            }
+
+            override fun onError(t: Throwable) {
+                expect(7)
+                assert(t is TestException)
+                latch.complete(Unit)
+            }
+        })
+        latch.await()
+        finish(8)
+    }
+
     private class TestException : Exception()
 }