AbstractCoroutines notifies onXXX before all the installed handlers;

so launch handles uncaught exceptions before "joining" coroutines wakeup.

Fixes #208

Author: elizarov

common/kotlinx-coroutines-core-common/src/main/kotlin/kotlinx/coroutines/experimental/AbstractCoroutine.kt

@@ -80,7 +80,7 @@ public abstract class AbstractCoroutine<in T>(
      */
     protected open fun onCancellation(cause: Throwable?) {}
 
-    internal final override fun onCancellationInternal(exceptionally: CompletedExceptionally?) {
+    internal override fun onCancellationInternal(exceptionally: CompletedExceptionally?) {
         onCancellation(exceptionally?.cause)
     }
 
@@ -94,13 +94,8 @@ public abstract class AbstractCoroutine<in T>(
      */
     protected open fun onCompletedExceptionally(exception: Throwable) {}
 
-    /**
-     * Override for post-completion actions that need to do something with the state.
-     * @param mode completion mode.
-     * @suppress **This is unstable API and it is subject to change.**
-     */
     @Suppress("UNCHECKED_CAST")
-    internal override fun afterCompletion(state: Any?, mode: Int) {
+    internal override fun onCompletionInternal(state: Any?, mode: Int) {
         if (state is CompletedExceptionally)
             onCompletedExceptionally(state.exception)
         else

common/kotlinx-coroutines-core-common/src/main/kotlin/kotlinx/coroutines/experimental/CommonJob.kt

@@ -87,7 +87,7 @@ internal expect open class JobSupport(active: Boolean) : Job {
 
     internal fun initParentJobInternal(parent: Job?)
     internal fun makeCompletingOnce(proposedUpdate: Any?, mode: Int): Boolean
-    internal open fun afterCompletion(state: Any?, mode: Int)
+    internal open fun onCompletionInternal(state: Any?, mode: Int)
     internal open fun onStartInternal()
     internal open fun onCancellationInternal(exceptionally: CompletedExceptionally?)
     internal open fun nameString(): String

common/kotlinx-coroutines-core-common/src/test/kotlin/kotlinx/coroutines/experimental/AbstractCoroutineTest.kt

@@ -0,0 +1,92 @@
+/*
+ * Copyright 2016-2017 JetBrains s.r.o.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package kotlinx.coroutines.experimental
+
+import kotlin.test.*
+
+class AbstractCoroutineTest : TestBase() {
+    @Test
+    fun testNotifications() = runTest {
+        expect(1)
+        val coroutine = object : AbstractCoroutine<String>(coroutineContext, false) {
+            override fun onStart() {
+                expect(3)
+            }
+
+            override fun onCancellation(cause: Throwable?) {
+                assertTrue(cause == null)
+                expect(5)
+            }
+
+            override fun onCompleted(value: String) {
+                assertEquals("OK", value)
+                expect(6)
+            }
+
+            override fun onCompletedExceptionally(exception: Throwable) {
+                expectUnreached()
+            }
+        }
+        coroutine.invokeOnCompletion {
+            assertTrue(it == null)
+            expect(7)
+        }
+        expect(2)
+        coroutine.start()
+        expect(4)
+        coroutine.resume("OK")
+        finish(8)
+    }
+
+    @Test
+    fun testNotificationsWithException() = runTest {
+        expect(1)
+        val coroutine = object : AbstractCoroutine<String>(coroutineContext, false) {
+            override fun onStart() {
+                expect(3)
+            }
+
+            override fun onCancellation(cause: Throwable?) {
+                assertTrue(cause is TestException0)
+                expect(5)
+            }
+
+            override fun onCompleted(value: String) {
+                expectUnreached()
+            }
+
+            override fun onCompletedExceptionally(exception: Throwable) {
+                assertTrue(exception is TestException1)
+                expect(7)
+            }
+        }
+        coroutine.invokeOnCompletion {
+            assertTrue(it is TestException1)
+            expect(8)
+        }
+        expect(2)
+        coroutine.start()
+        expect(4)
+        coroutine.cancel(TestException0())
+        expect(6)
+        coroutine.resumeWithException(TestException1())
+        finish(9)
+    }
+
+    private class TestException0 : Throwable()
+    private class TestException1 : Throwable()
+}

core/kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/AbstractContinuation.kt

@@ -74,13 +74,13 @@ internal abstract class AbstractContinuation<in T>(
     @PublishedApi
     internal fun getResult(): Any? {
         if (trySuspend()) return COROUTINE_SUSPENDED
-        // otherwise, afterCompletion was already invoked & invoked tryResume, and the result is in the state
+        // otherwise, onCompletionInternal was already invoked & invoked tryResume, and the result is in the state
         val state = this.state
         if (state is CompletedExceptionally) throw state.exception
         return getSuccessfulResult(state)
     }
 
-    override fun afterCompletion(state: Any?, mode: Int) {
+    internal final override fun onCompletionInternal(state: Any?, mode: Int) {
         if (tryResume()) return // completed before getResult invocation -- bail out
         // otherwise, getResult has already commenced, i.e. completed later or in other thread
         dispatch(mode)

core/kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/Builders.kt

@@ -218,7 +218,7 @@ private class BlockingCoroutine<T>(
         if (privateEventLoop) require(eventLoop is BlockingEventLoop)
     }
 
-    override fun afterCompletion(state: Any?, mode: Int) {
+    override fun onCancellationInternal(exceptionally: CompletedExceptionally?) {
         // wake up blocked thread
         if (Thread.currentThread() != blockedThread)
             LockSupport.unpark(blockedThread)

core/kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/Job.kt

@@ -695,8 +695,11 @@ internal actual open class JobSupport actual constructor(active: Boolean) : Job,
      * @suppress **This is unstable API and it is subject to change.**
      */
     internal fun completeUpdateState(expect: Incomplete, update: Any?, mode: Int) {
-        // Invoke completion handlers
         val exceptionally = update as? CompletedExceptionally
+        // Do overridable processing before completion handlers
+        if (!expect.isCancelling) onCancellationInternal(exceptionally) // only notify when was not cancelling before
+        onCompletionInternal(update, mode)
+        // Invoke completion handlers
         val cause = exceptionally?.cause
         if (expect is JobNode<*>) { // SINGLE/SINGLE+ state -- one completion handler (common case)
             try {
@@ -707,9 +710,6 @@ internal actual open class JobSupport actual constructor(active: Boolean) : Job,
         } else {
             expect.list?.notifyCompletion(cause)
         }
-        // Do overridable processing after completion handlers
-        if (!expect.isCancelling) onCancellationInternal(exceptionally) // only notify when was not cancelling before
-        afterCompletion(update, mode)
     }
 
     private inline fun <reified T: JobNode<*>> notifyHandlers(list: NodeList, cause: Throwable?) {
@@ -1162,7 +1162,7 @@ internal actual open class JobSupport actual constructor(active: Boolean) : Job,
      * @param mode completion mode.
      * @suppress **This is unstable API and it is subject to change.**
      */
-    internal actual open fun afterCompletion(state: Any?, mode: Int) {}
+    internal actual open fun onCompletionInternal(state: Any?, mode: Int) {}
 
     // for nicer debugging
     public override fun toString(): String =

core/kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/Scheduled.kt

@@ -101,7 +101,7 @@ private open class TimeoutCoroutine<U, in T: U>(
     }
 
     @Suppress("UNCHECKED_CAST")
-    override fun afterCompletion(state: Any?, mode: Int) {
+    internal override fun onCompletionInternal(state: Any?, mode: Int) {
         if (state is CompletedExceptionally)
             cont.resumeWithExceptionMode(state.exception, mode)
         else
@@ -171,7 +171,7 @@ private class TimeoutOrNullCoroutine<T>(
     cont: Continuation<T?>
 ) : TimeoutCoroutine<T?, T>(time, unit, cont) {
     @Suppress("UNCHECKED_CAST")
-    override fun afterCompletion(state: Any?, mode: Int) {
+    internal override fun onCompletionInternal(state: Any?, mode: Int) {
         if (state is CompletedExceptionally) {
             val exception = state.exception
             if (exception is TimeoutCancellationException && exception.coroutine === this)

js/kotlinx-coroutines-core-js/src/main/kotlin/kotlinx/coroutines/experimental/AbstractContinuation.kt

@@ -64,13 +64,13 @@ internal abstract class AbstractContinuation<in T>(
     @PublishedApi
     internal fun getResult(): Any? {
         if (trySuspend()) return COROUTINE_SUSPENDED
-        // otherwise, afterCompletion was already invoked & invoked tryResume, and the result is in the state
+        // otherwise, onCompletionInternal was already invoked & invoked tryResume, and the result is in the state
         val state = this.state
         if (state is CompletedExceptionally) throw state.exception
         return getSuccessfulResult(state)
     }
 
-    override fun afterCompletion(state: Any?, mode: Int) {
+    internal final override fun onCompletionInternal(state: Any?, mode: Int) {
         if (tryResume()) return // completed before getResult invocation -- bail out
         // otherwise, getResult has already commenced, i.e. completed later or in other thread
         dispatch(mode)

js/kotlinx-coroutines-core-js/src/main/kotlin/kotlinx/coroutines/experimental/Job.kt

@@ -461,8 +461,11 @@ internal actual open class JobSupport actual constructor(active: Boolean) : Job
      * @suppress **This is unstable API and it is subject to change.**
      */
     internal fun completeUpdateState(expect: Incomplete, update: Any?, mode: Int) {
-        // Invoke completion handlers
         val exceptionally = update as? CompletedExceptionally
+        // Do overridable processing before completion handlers
+        if (!expect.isCancelling) onCancellationInternal(exceptionally) // only notify when was not cancelling before
+        onCompletionInternal(update, mode)
+        // Invoke completion handlers
         val cause = exceptionally?.cause
         if (expect is JobNode<*>) { // SINGLE/SINGLE+ state -- one completion handler (common case)
             try {
@@ -473,9 +476,6 @@ internal actual open class JobSupport actual constructor(active: Boolean) : Job
         } else {
             expect.list?.notifyCompletion(cause)
         }
-        // Do overridable processing after completion handlers
-        if (!expect.isCancelling) onCancellationInternal(exceptionally) // only notify when was not cancelling before
-        afterCompletion(update, mode)
     }
 
     private inline fun <reified T: JobNode<*>> notifyHandlers(list: NodeList, cause: Throwable?) {
@@ -861,11 +861,11 @@ internal actual open class JobSupport actual constructor(active: Boolean) : Job
     internal actual open fun onCancellationInternal(exceptionally: CompletedExceptionally?) {}
 
     /**
-     * Override for post-completion actions that need to do something with the state.
+     * Override for completion actions that need to do something with the state.
      * @param mode completion mode.
      * @suppress **This is unstable API and it is subject to change.**
      */
-    internal actual open fun afterCompletion(state: Any?, mode: Int) {}
+    internal actual open fun onCompletionInternal(state: Any?, mode: Int) {}
 
     // for nicer debugging
     public override fun toString(): String =

js/kotlinx-coroutines-core-js/src/main/kotlin/kotlinx/coroutines/experimental/Scheduled.kt

@@ -71,7 +71,7 @@ private open class TimeoutCoroutine<U, in T: U>(
     }
 
     @Suppress("UNCHECKED_CAST")
-    override fun afterCompletion(state: Any?, mode: Int) {
+    internal override fun onCompletionInternal(state: Any?, mode: Int) {
         if (state is CompletedExceptionally)
             cont.resumeWithExceptionMode(state.exception, mode)
         else
@@ -111,7 +111,7 @@ private class TimeoutOrNullCoroutine<T>(
     cont: Continuation<T?>
 ) : TimeoutCoroutine<T?, T>(time, cont) {
     @Suppress("UNCHECKED_CAST")
-    override fun afterCompletion(state: Any?, mode: Int) {
+    internal override fun onCompletionInternal(state: Any?, mode: Int) {
         if (state is CompletedExceptionally) {
             val exception = state.exception
             if (exception is TimeoutCancellationException && exception.coroutine === this)