Add conversion of CompletionStage to Deferred

Fixes #262

Author: Jonathan Cornaz

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

@@ -20,6 +20,7 @@ import kotlinx.coroutines.experimental.*
 import java.util.concurrent.CompletableFuture
 import java.util.concurrent.CompletionStage
 import java.util.concurrent.ExecutionException
+import java.util.concurrent.Future
 import java.util.function.BiConsumer
 import kotlin.coroutines.experimental.Continuation
 import kotlin.coroutines.experimental.ContinuationInterceptor
@@ -114,6 +115,34 @@ public suspend fun <T> CompletionStage<T>.await(): T = suspendCoroutine { cont:
     whenComplete(ContinuationConsumer(cont))
 }
 
+/**
+ * Converts this future to an instance of [Deferred].
+ */
+public fun <T> CompletionStage<T>.asDeferred(): Deferred<T> {
+
+    // Fast path if already completed
+    if (this is Future<*> && isDone()){
+        return try {
+            @Suppress("UNCHECKED_CAST")
+            CompletableDeferred(get() as T)
+        } catch (t: Throwable) {
+            CompletableDeferred<T>().also { it.completeExceptionally(t) }
+        }
+    }
+
+    val result = CompletableDeferred<T>()
+
+    whenComplete { value, exception ->
+        if (exception == null) {
+            result.complete(value)
+        } else {
+            result.completeExceptionally(exception)
+        }
+    }
+
+    return result
+}
+
 /**
  * Awaits for completion of the future without blocking a thread.
  *

integration/kotlinx-coroutines-jdk8/src/test/kotlin/kotlinx/coroutines/experimental/future/FutureTest.kt

@@ -25,6 +25,8 @@ import java.util.concurrent.CompletableFuture
 import java.util.concurrent.CompletionStage
 import java.util.concurrent.ExecutionException
 import java.util.concurrent.atomic.AtomicInteger
+import java.util.concurrent.locks.ReentrantLock
+import kotlin.concurrent.withLock
 import kotlin.coroutines.experimental.CoroutineContext
 
 class FutureTest : TestBase() {
@@ -240,13 +242,13 @@ class FutureTest : TestBase() {
             assertEquals("Part before first suspension must be wrapped", 1, depth.get())
             val result =
                     CompletableFuture.supplyAsync {
-                        while (depth.get() > 0) ;
+                        while (depth.get() > 0);
                         assertEquals("Part inside suspension point should not be wrapped", 0, depth.get())
                         "OK"
                     }.await()
             assertEquals("Part after first suspension should be wrapped", 1, depth.get())
             CompletableFuture.supplyAsync {
-                while (depth.get() > 0) ;
+                while (depth.get() > 0);
                 assertEquals("Part inside suspension point should not be wrapped", 0, depth.get())
                 "ignored"
             }.await()
@@ -255,6 +257,63 @@ class FutureTest : TestBase() {
         assertThat(future.get(), IsEqual("OK"))
     }
 
+    @Test
+    fun testCompletableFutureStageAsDeferred() = runBlocking {
+        val lock = ReentrantLock().apply { lock() }
+
+        val deferred: Deferred<Int> = CompletableFuture.supplyAsync {
+            lock.withLock { 42 }
+        }.asDeferred()
+
+        assertFalse(deferred.isCompleted)
+        lock.unlock()
+
+        assertEquals(42, deferred.await())
+        assertTrue(deferred.isCompleted)
+    }
+
+    @Test
+    fun testCompletedFutureAsDeferred() = runBlocking {
+        val deferred: Deferred<Int> = CompletableFuture.completedFuture(42).asDeferred()
+        assertEquals(42, deferred.await())
+    }
+
+    @Test
+    fun testFailedFutureAsDeferred() = runBlocking {
+        val future = CompletableFuture<Int>().apply { completeExceptionally(Exception("something went wrong")) }
+        val deferred = future.asDeferred()
+
+        assertTrue(deferred.isCompletedExceptionally)
+        assertEquals("something went wrong", deferred.getCompletionExceptionOrNull()!!.cause!!.message)
+
+        try {
+            deferred.await()
+            fail("deferred.await() should throw an exception")
+        } catch (e: Exception) {
+            assertEquals("something went wrong", e.cause!!.message)
+        }
+    }
+
+    @Test
+    fun testCompletableFutureWithExceptionAsDeferred() = runBlocking {
+        val lock = ReentrantLock().apply { lock() }
+
+        val deferred: Deferred<Int> = CompletableFuture.supplyAsync {
+            lock.withLock { throw Exception("something went wrong") }
+        }.asDeferred()
+
+        assertFalse(deferred.isCompleted)
+        lock.unlock()
+
+        try {
+            deferred.await()
+            fail("deferred.await() should throw an exception")
+        } catch (e: Exception) {
+            assertTrue(deferred.isCompletedExceptionally)
+            assertEquals("something went wrong", e.cause!!.message)
+        }
+    }
+
     private fun wrapContinuation(wrapper: (() -> Unit) -> Unit): CoroutineDispatcher = object : CoroutineDispatcher() {
         override fun dispatch(context: CoroutineContext, block: Runnable) {
             wrapper {