Improve code style and test coverage

Author: qwwdfsad

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

@@ -160,10 +160,10 @@ private class PublisherCoroutine<in T>(
 
     private fun unlockAndCheckCompleted() {
        /*
-         There is no sense to check completion before doing `unlock`, because completion might
-         happen after this check and before `unlock` (see `signalCompleted` that does not do anything
-         if it fails to acquire the lock that we are still holding).
-         We have to recheck `isCompleted` after `unlock` anyway.
+        * There is no sense to check completion before doing `unlock`, because completion might
+        * happen after this check and before `unlock` (see `signalCompleted` that does not do anything
+        * if it fails to acquire the lock that we are still holding).
+        * We have to recheck `isCompleted` after `unlock` anyway.
         */
         mutex.unlock()
         // check isCompleted and and try to regain lock to signal completion
@@ -180,13 +180,17 @@ private class PublisherCoroutine<in T>(
                 if (cancelled) {
                     // If the parent had failed to handle our exception (handleJobException was invoked), then
                     // we must not loose this exception
-                    if (shouldHandleException && cause != null) handleExceptionViaHandler(parentContext, cause)
+                    if (shouldHandleException && cause != null) {
+                        handleExceptionViaHandler(parentContext, cause)
+                    }
                 } else {
                     try {
-                        if (cause != null && cause !is CancellationException)
+                        if (cause != null && cause !is CancellationException) {
                             subscriber.onError(cause)
-                        else
+                        }
+                        else {
                             subscriber.onComplete()
+                        }
                     } catch (e: Throwable) {
                         handleExceptionViaHandler(parentContext, e)
                     }
@@ -261,4 +265,4 @@ private class PublisherCoroutine<in T>(
         cancelled = true
         super.cancel()
     }
-}
+}

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

@@ -131,6 +131,51 @@ class PublishTest : TestBase() {
         finish(6)
     }
 
+    @Test
+    fun testParentHandlesFailure() = runTest {
+        expect(1)
+        val deferred = CompletableDeferred<Unit>()
+        val publisher = publish<Unit>(deferred + Dispatchers.Unconfined) {
+            try {
+                expect(3)
+                delay(10000)
+            } finally {
+                expect(5)
+                throw TestException("FAILED")
+            }
+        }
+        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()
+
+        try {
+            deferred.await()
+            expectUnreached()
+        } catch (e: TestException) {
+            expect(6)
+        }
+
+        finish(7)
+    }
+
     @Test
     fun testPublishFailureCancelsParent() = runTest(
         expected = { it is TestException }
@@ -207,6 +252,4 @@ class PublishTest : TestBase() {
         latch.await()
         finish(8)
     }
-
-    private class TestException : Exception()
 }