Deferred.getCompleted() and an efficient implementation of Deferred.toCompletableFuture() extension

Author: elizarov

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

@@ -59,7 +59,7 @@ private class StandaloneCoroutine(
     val parentContext: CoroutineContext
 ) : JobContinuation<Unit>(parentContext) {
     override fun afterCompletion(state: Any?, closeException: Throwable?) {
-        if (closeException != null) handleCoroutineException(context, closeException)
+        super.afterCompletion(state, closeException) // handle closeException
         // note the use of the parent context below!
         if (state is CompletedExceptionally) handleCoroutineException(parentContext, state.exception)
     }
@@ -69,7 +69,7 @@ private class BlockingCoroutine<T>(parentContext: CoroutineContext) : JobContinu
     val blockedThread: Thread = Thread.currentThread()
 
     override fun afterCompletion(state: Any?, closeException: Throwable?) {
-        if (closeException != null) handleCoroutineException(context, closeException)
+        super.afterCompletion(state, closeException) // handle closeException
         LockSupport.unpark(blockedThread)
     }
 

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

@@ -47,7 +47,7 @@ internal class SafeCancellableContinuation<in T>(
 
     @Suppress("UNCHECKED_CAST")
     override fun afterCompletion(state: Any?, closeException: Throwable?) {
-        if (closeException != null) handleCoroutineException(context, closeException)
+        super.afterCompletion(state, closeException) // handle closeException
         if (suspendedThread === Thread.currentThread()) {
             // cancelled during suspendCancellableCoroutine in its thread
             suspendedThread = null

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

@@ -12,9 +12,17 @@ public interface Deferred<out T> : Job {
      * Awaits for completion of this value without blocking a thread and resumes when deferred computation is complete.
      * This suspending function is cancellable.
      * If the [Job] of the current coroutine is completed while this suspending function is waiting, this function
-     * immediately resumes with [CancellationException] .
+     * immediately resumes with [CancellationException].
      */
     public suspend fun await(): T
+
+    /**
+     * Returns *completed* result or throws [IllegalStateException] if this deferred value is still [isActive].
+     * It throws the corresponding exception if this deferred has completed exceptionally.
+     * This function is designed to be used from [onCompletion] handlers, when there is an absolute certainty that
+     * the value is already complete.
+     */
+    public fun getCompleted(): T
 }
 
 /**
@@ -54,7 +62,11 @@ private class DeferredCoroutine<T>(
         })
     }
 
-    override fun afterCompletion(state: Any?, closeException: Throwable?) {
-        if (closeException != null) handleCoroutineException(context, closeException)
+    @Suppress("UNCHECKED_CAST")
+    override fun getCompleted(): T {
+        val state = getState()
+        check(state !is Active) { "This deferred value is still active" }
+        if (state is CompletedExceptionally) throw state.exception
+        return state as T
     }
 }

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

@@ -35,4 +35,8 @@ internal open class JobContinuation<in T>(
             }
         }
     }
+
+    override fun afterCompletion(state: Any?, closeException: Throwable?) {
+        if (closeException != null) handleCoroutineException(context, closeException)
+    }
 }

kotlinx-coroutines-jdk8/src/main/kotlin/kotlinx/coroutines/experimental/future/Future.kt

@@ -35,9 +35,8 @@ public fun <T> Deferred<T>.toCompletableFuture(): CompletableFuture<T> {
     val future = CompletableFuture<T>()
     future.whenComplete { _, exception -> cancel(exception) }
     onCompletion {
-        // todo: write better (more efficient) implementation, because we know that await will not suspend
         try {
-            future.complete(runBlocking(Job() + Here) { await() })
+            future.complete(getCompleted())
         } catch (exception: Exception) {
             future.completeExceptionally(exception)
         }