Public API for AbstractCoroutine

Author: elizarov

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

@@ -16,43 +16,107 @@
 
 package kotlinx.coroutines.experimental
 
-import kotlin.coroutines.experimental.Continuation
-import kotlin.coroutines.experimental.CoroutineContext
+import kotlin.coroutines.experimental.*
 
 /**
- * Abstract class for coroutines.
+ * Abstract base class for implementation of coroutines in coroutine builders.
  *
  *  * Coroutines implement completion [Continuation], [Job], and [CoroutineScope] interfaces.
  *  * Coroutine stores the result of continuation in the state of the job.
  *  * Coroutine waits for children coroutines to finish before completing.
  *  * Coroutines are cancelled through an intermediate _cancelling_ state.
  *
- * @param active when `true` coroutine is created in _active_ state, when `false` in _new_ state. See [Job] for details.
- * @suppress **This is unstable API and it is subject to change.**
+ * @param parentContext context of the parent coroutine.
+ * @param active when `true` (by default) coroutine is created in _active_ state, when `false` in _new_ state.
+ *               See [Job] for details.
  */
+@Suppress("EXPOSED_SUPER_CLASS")
 public abstract class AbstractCoroutine<in T>(
     private val parentContext: CoroutineContext,
-    active: Boolean
-) : JobSupport(active), Continuation<T>, CoroutineScope {
+    active: Boolean = true
+) : JobSupport(active), Job, Continuation<T>, CoroutineScope {
     @Suppress("LeakingThis")
     public final override val context: CoroutineContext = parentContext + this
     public final override val coroutineContext: CoroutineContext get() = context
 
-    protected open val defaultResumeMode: Int get() = MODE_ATOMIC_DEFAULT
+    /**
+     * Initializes parent job from the `parentContext` of this coroutine that was passed to it during construction.
+     * It shall be invoked at most once after construction after all other initialization.
+     * 
+     * Invocation of this function may cause this coroutine to become cancelled if parent is already cancelled,
+     * in which case it synchronously invokes all the corresponding handlers.
+     */
+    public fun initParentJob() {
+        initParentJobInternal(parentContext[Job])
+    }
+
+    /**
+     * This function is invoked once when non-active coroutine (constructed with `active` set to `false)
+     * is [started][start].
+     */
+    protected open fun onStart() {}
+
+    internal final override fun onStartInternal() {
+        onStart()
+    }
+
+    /**
+     * This function is invoked once when this coroutine is cancelled or is completed,
+     * similarly to [invokeOnCompletion] with `onCancelling` set to `true`.
+     *
+     * @param cause the cause that was passed to [Job.cancel] function or `null` if coroutine was cancelled
+     *              without cause or is completing normally.
+     */
+    protected open fun onCancellation(cause: Throwable?) {}
+
+    internal final override fun onCancellationInternal(exceptionally: CompletedExceptionally?) {
+        onCancellation(exceptionally?.cause)
+    }
+
+    /**
+     * This function is invoked once when job is completed normally with the specified [value].
+     */
+    protected open fun onCompleted(value: T) {}
+
+    /**
+     * This function is invoked once when job is completed exceptionally with the specified [exception].
+     */
+    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) {
+        if (state is CompletedExceptionally)
+            onCompletedExceptionally(state.exception)
+        else
+            onCompleted(state as T)
+    }
+
+    internal open val defaultResumeMode: Int get() = MODE_ATOMIC_DEFAULT
 
-    final override fun resume(value: T) {
+    /**
+     * Completes execution of this coroutine normally with the specified [value].
+     */
+    public final override fun resume(value: T) {
         makeCompletingOnce(value, defaultResumeMode)
     }
 
-    final override fun resumeWithException(exception: Throwable) {
+    /**
+     * Completes execution of this with coroutine exceptionally with the specified [exception].
+     */
+    public final override fun resumeWithException(exception: Throwable) {
         makeCompletingOnce(CompletedExceptionally(exception), defaultResumeMode)
     }
 
-    final override fun handleException(exception: Throwable) {
+    internal final override fun handleException(exception: Throwable) {
         handleCoroutineException(parentContext, exception)
     }
 
-    override fun nameString(): String {
+    internal override fun nameString(): String {
         val coroutineName = context.coroutineName ?: return super.nameString()
         return "\"$coroutineName\":${super.nameString()}"
     }

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

@@ -62,7 +62,7 @@ public actual fun launch(
     val coroutine = if (start.isLazy)
         LazyStandaloneCoroutine(newContext, block) else
         StandaloneCoroutine(newContext, active = true)
-    coroutine.initParentJob(newContext[Job])
+    coroutine.initParentJob()
     start(block, coroutine, coroutine)
     return coroutine
 }
@@ -126,7 +126,7 @@ public actual suspend fun <T> withContext(
         context = newContext,
         delegate = cont,
         resumeMode = if (start == CoroutineStart.ATOMIC) MODE_ATOMIC_DEFAULT else MODE_CANCELLABLE)
-    completion.initParentJob(newContext[Job]) // attach to job
+    completion.initParentJobInternal(newContext[Job]) // attach to job
     start(block, completion)
     completion.getResult()
 }
@@ -169,7 +169,7 @@ public fun <T> runBlocking(context: CoroutineContext = EmptyCoroutineContext, bl
     val eventLoop = if (context[ContinuationInterceptor] == null) BlockingEventLoop(currentThread) else null
     val newContext = newCoroutineContext(context + (eventLoop ?: EmptyCoroutineContext))
     val coroutine = BlockingCoroutine<T>(newContext, currentThread, privateEventLoop = eventLoop != null)
-    coroutine.initParentJob(newContext[Job])
+    coroutine.initParentJob()
     block.startCoroutine(coroutine, coroutine)
     return coroutine.joinBlocking()
 }
@@ -180,9 +180,9 @@ private open class StandaloneCoroutine(
     private val parentContext: CoroutineContext,
     active: Boolean
 ) : AbstractCoroutine<Unit>(parentContext, active) {
-    override fun onCancellation(exceptionally: CompletedExceptionally?) {
+    override fun onCancellation(cause: Throwable?) {
         // note the use of the parent's job context below!
-        if (exceptionally != null) handleCoroutineException(parentContext, exceptionally.exception)
+        if (cause != null) handleCoroutineException(parentContext, cause)
     }
 }
 

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

@@ -222,7 +222,7 @@ internal class CancellableContinuationImpl<in T>(
         get() = _context ?: (delegate.context + this).also { _context = it }
 
     override fun initCancellability() {
-        initParentJob(delegate.context[Job])
+        initParentJobInternal(delegate.context[Job])
     }
 
     override val onCancelMode: Int get() = ON_CANCEL_MAKE_CANCELLED

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

@@ -16,7 +16,7 @@
 
 package kotlinx.coroutines.experimental
 
-import kotlinx.coroutines.experimental.selects.SelectClause1
+import kotlinx.coroutines.experimental.selects.*
 
 /**
  * A [Deferred] that can be completed via public functions
@@ -71,18 +71,17 @@ public actual fun <T> CompletableDeferred(value: T): CompletableDeferred<T> = Co
 @Suppress("UNCHECKED_CAST")
 private class CompletableDeferredImpl<T>(
     parent: Job?
-) : JobSupport(true), CompletableDeferred<T> {
-    init { initParentJob(parent) }
+) : JobSupport(true), CompletableDeferred<T>, SelectClause1<T> {
+    init { initParentJobInternal(parent) }
     override val onCancelMode: Int get() = ON_CANCEL_MAKE_COMPLETING
-
     override fun getCompleted(): T = getCompletedInternal() as T
     override suspend fun await(): T = awaitInternal() as T
-    override val onAwait: SelectClause1<T>
-        get() = this as SelectClause1<T>
+    override val onAwait: SelectClause1<T> get() = this
+    override fun <R> registerSelectClause1(select: SelectInstance<R>, block: suspend (T) -> R) =
+        registerSelectClause1Internal(select, block)
 
     override fun complete(value: T): Boolean =
         makeCompleting(value)
-
     override fun completeExceptionally(exception: Throwable): Boolean =
         makeCompleting(CompletedExceptionally(exception))
 }

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

@@ -16,8 +16,7 @@
 
 package kotlinx.coroutines.experimental
 
-import kotlinx.coroutines.experimental.selects.SelectClause1
-import kotlinx.coroutines.experimental.selects.select
+import kotlinx.coroutines.experimental.selects.*
 import kotlin.coroutines.experimental.ContinuationInterceptor
 import kotlin.coroutines.experimental.CoroutineContext
 
@@ -168,7 +167,7 @@ public actual fun <T> async(
     val coroutine = if (start.isLazy)
         LazyDeferredCoroutine(newContext, block) else
         DeferredCoroutine<T>(newContext, active = true)
-    coroutine.initParentJob(newContext[Job])
+    coroutine.initParentJob()
     start(block, coroutine, coroutine)
     return coroutine
 }
@@ -202,11 +201,12 @@ public fun <T> defer(context: CoroutineContext, block: suspend CoroutineScope.()
 private open class DeferredCoroutine<T>(
     parentContext: CoroutineContext,
     active: Boolean
-) : AbstractCoroutine<T>(parentContext, active), Deferred<T> {
+) : AbstractCoroutine<T>(parentContext, active), Deferred<T>, SelectClause1<T> {
     override fun getCompleted(): T = getCompletedInternal() as T
     override suspend fun await(): T = awaitInternal() as T
-    override val onAwait: SelectClause1<T>
-        get() = this as SelectClause1<T>
+    override val onAwait: SelectClause1<T> get() = this
+    override fun <R> registerSelectClause1(select: SelectInstance<R>, block: suspend (T) -> R) =
+        registerSelectClause1Internal(select, block)
 }
 
 private class LazyDeferredCoroutine<T>(