JDK8 docs, impl and tests improved

elizarov · elizarov · commit 489cac28f14d · 2017-05-17T12:18:28.000+03:00
diff --git a/integration/kotlinx-coroutines-jdk8/pom.xml b/integration/kotlinx-coroutines-jdk8/pom.xml
@@ -16,7 +16,6 @@
   -->
 
 <project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">
-
     <modelVersion>4.0.0</modelVersion>
 
     <parent>
@@ -35,12 +34,19 @@
     </build>
 
     <dependencies>
+        <!-- dependency on coroutines core -->
         <dependency>
             <groupId>org.jetbrains.kotlinx</groupId>
             <artifactId>kotlinx-coroutines-core</artifactId>
             <version>${project.version}</version>
-            <scope>compile</scope>
+        </dependency>
+        <!-- coroutines test framework dependency -->
+        <dependency>
+            <groupId>org.jetbrains.kotlinx</groupId>
+            <artifactId>kotlinx-coroutines-core</artifactId>
+            <version>${project.version}</version>
+            <classifier>tests</classifier>
+            <scope>test</scope>
         </dependency>
     </dependencies>
-
 </project>
diff --git a/integration/kotlinx-coroutines-jdk8/src/main/kotlin/kotlinx/coroutines/experimental/future/Future.kt b/integration/kotlinx-coroutines-jdk8/src/main/kotlin/kotlinx/coroutines/experimental/future/Future.kt
@@ -19,6 +19,7 @@ package kotlinx.coroutines.experimental.future
 import kotlinx.coroutines.experimental.*
 import java.util.concurrent.CompletableFuture
 import java.util.concurrent.CompletionStage
+import java.util.concurrent.ExecutionException
 import java.util.function.BiConsumer
 import kotlin.coroutines.experimental.Continuation
 import kotlin.coroutines.experimental.CoroutineContext
@@ -29,16 +30,16 @@ import kotlin.coroutines.experimental.suspendCoroutine
  * This coroutine builder uses [CommonPool] context by default and is conceptually similar to [CompletableFuture.supplyAsync].
  *
  * The running coroutine is cancelled when the resulting future is cancelled or otherwise completed.
- * If the [context] for the new coroutine is explicitly specified and does not include a coroutine interceptor,
- * then [CoroutineDispatcher] element.
- * See [CoroutineDispatcher] for the standard [context] implementations that are provided by `kotlinx.coroutines`.
- * The specified context is added to the context of the parent running coroutine (if any) inside which this function
- * is invoked. The [Job] of the resulting coroutine is a child of the job of the parent coroutine (if any).
+ * If the [context] for the new coroutine is omitted or is explicitly specified but does not include a
+ * coroutine interceptor, then [CommonPool] is used.
+ * See [CoroutineDispatcher] for other standard [context] implementations that are provided by `kotlinx.coroutines`.
+ * The [context][CoroutineScope.context] of the parent coroutine from its [scope][CoroutineScope] may be used,
+ * in which case the [Job] of the resulting coroutine is a child of the job of the parent coroutine.
  *
  * By default, the coroutine is immediately scheduled for execution.
  * Other options can be specified via `start` parameter. See [CoroutineStart] for details.
  * A value of [CoroutineStart.LAZY] is not supported
- * (since `ListenableFuture` framework does not provide the corresponding capability) and
+ * (since `CompletableFuture` framework does not provide the corresponding capability) and
  * produces [IllegalArgumentException].
  *
  * See [newCoroutineContext] for a description of debugging facilities that are available for newly created coroutine.
@@ -58,7 +59,7 @@ public fun <T> future(
     val future = CompletableFutureCoroutine<T>(newContext + job)
     job.cancelFutureOnCompletion(future)
     future.whenComplete { _, exception -> job.cancel(exception) }
-    start(block, receiver=future, completion=future)
+    start(block, receiver=future, completion=future) // use the specified start strategy
     return future
 }
 
@@ -91,7 +92,7 @@ public fun <T> Deferred<T>.asCompletableFuture(): CompletableFuture<T> {
  * Awaits for completion of the completion stage without blocking a thread.
  *
  * This suspending function is not cancellable, because there is no way to cancel a `CompletionStage`.
- * Use `CompletableFuture.await()` for cancellation support.
+ * Use `CompletableFuture.await()` for cancellable wait.
  */
 public suspend fun <T> CompletionStage<T>.await(): T = suspendCoroutine { cont: Continuation<T> ->
     whenComplete(ContinuationConsumer(cont))
@@ -103,35 +104,36 @@ public suspend fun <T> CompletionStage<T>.await(): T = suspendCoroutine { cont:
  * This suspending function is cancellable.
  * If the [Job] of the current coroutine is completed while this suspending function is waiting, this function
  * stops waiting for the future and immediately resumes with [CancellationException].
+ *
+ * Note, that `CompletableFuture` does not support prompt removal of installed listeners, so on cancellation of this wait
+ * a few small objects will remain in the `CompletableFuture` stack of completion actions until the future completes.
+ * However, the care is taken to clear the reference to the waiting coroutine itself, so that its memory can be
+ * released even if the future never completes.
  */
 public suspend fun <T> CompletableFuture<T>.await(): T {
-    if (isDone) { // fast path when CompletableFuture is already done (does not suspend)
-        // then only way to get unwrapped exception from the CompletableFuture is via whenComplete anyway
-        var result: T? = null
-        var exception: Throwable? = null
-        whenComplete { r, e ->
-            result = r
-            exception = e
+    // fast path when CompletableFuture is already done (does not suspend)
+    if (isDone) {
+        try {
+            return get()
+        } catch (e: ExecutionException) {
+            throw e.cause ?: e // unwrap original cause from ExecutionException
         }
-        if (exception != null) throw exception!!
-        return result as T
     }
     // slow path -- suspend
     return suspendCancellableCoroutine { cont: CancellableContinuation<T> ->
         val consumer = ContinuationConsumer(cont)
-        val completionFuture = whenComplete(consumer)
+        whenComplete(consumer)
         cont.invokeOnCompletion {
-            completionFuture.cancel(false) // cancel future
-            consumer.cont = null // shall clear reference to continuation, because CompletableFuture continues to keep it
+            consumer.cont = null // shall clear reference to continuation
         }
     }
 }
 
 private class ContinuationConsumer<T>(
-    @JvmField var cont: Continuation<T>?
+    @Volatile @JvmField var cont: Continuation<T>?
 ) : BiConsumer<T?, Throwable?> {
     override fun accept(result: T?, exception: Throwable?) {
-        val cont = this.cont ?: return // atomically read current value unless null, benign data race when it is set to null
+        val cont = this.cont ?: return // atomically read current value unless null
         if (exception == null) // the future has been completed normally
             cont.resume(result as T)
         else // the future has completed with an exception
@@ -152,6 +154,7 @@ private class ContinuationConsumer<T>(
     replaceWith = ReplaceWith("asCompletableFuture()"))
 public fun <T> Deferred<T>.toCompletableFuture(): CompletableFuture<T> = asCompletableFuture()
 
+@Suppress("DeprecatedCallableAddReplaceWith") // todo: the warning is incorrectly shown, see KT-17917
 @Deprecated("Use the other version. This one is for binary compatibility only.", level=DeprecationLevel.HIDDEN)
 public fun <T> future(
     context: CoroutineContext = CommonPool,
diff --git a/integration/kotlinx-coroutines-jdk8/src/test/kotlin/kotlinx/coroutines/experimental/future/FutureTest.kt b/integration/kotlinx-coroutines-jdk8/src/test/kotlin/kotlinx/coroutines/experimental/future/FutureTest.kt
@@ -16,24 +16,46 @@
 
 package kotlinx.coroutines.experimental.future
 
-import kotlinx.coroutines.experimental.CoroutineDispatcher
+import kotlinx.coroutines.experimental.*
+import org.hamcrest.core.IsEqual
+import org.junit.Assert.*
 import org.junit.Test
 import java.util.concurrent.CompletableFuture
+import java.util.concurrent.CompletionStage
 import java.util.concurrent.ExecutionException
 import java.util.concurrent.atomic.AtomicInteger
 import kotlin.coroutines.experimental.CoroutineContext
-import org.junit.Assert.*
-import java.util.concurrent.CompletionStage
 
-class FutureTest {
+class FutureTest : TestBase() {
     @Test
-    fun testSimple() {
+    fun testSimpleAwait() {
         val future = future {
             CompletableFuture.supplyAsync {
                 "O"
             }.await() + "K"
         }
-        assertEquals("OK", future.get())
+        assertThat(future.get(), IsEqual("OK"))
+    }
+
+    @Test
+    fun testCompletedFuture() {
+        val toAwait = CompletableFuture<String>()
+        toAwait.complete("O")
+        val future = future {
+            toAwait.await() + "K"
+        }
+        assertThat(future.get(), IsEqual("OK"))
+    }
+
+    @Test
+    fun testCompletedCompletionStage() {
+        val completable = CompletableFuture<String>()
+        completable.complete("O")
+        val toAwait: CompletionStage<String> = completable
+        val future = future {
+            toAwait.await() + "K"
+        }
+        assertThat(future.get(), IsEqual("OK"))
     }
 
     @Test
@@ -44,7 +66,7 @@ class FutureTest {
         }
         assertFalse(future.isDone)
         toAwait.complete("O")
-        assertEquals("OK", future.get())
+        assertThat(future.get(), IsEqual("OK"))
     }
 
     @Test
@@ -56,11 +78,11 @@ class FutureTest {
         }
         assertFalse(future.isDone)
         completable.complete("O")
-        assertEquals("OK", future.get())
+        assertThat(future.get(), IsEqual("OK"))
     }
 
     @Test
-    fun testDoneFutureExceptionally() {
+    fun testCompletedFutureExceptionally() {
         val toAwait = CompletableFuture<String>()
         toAwait.completeExceptionally(RuntimeException("O"))
         val future = future<String> {
@@ -70,11 +92,11 @@ class FutureTest {
                 e.message!!
             } + "K"
         }
-        assertEquals("OK", future.get())
+        assertThat(future.get(), IsEqual("OK"))
     }
 
     @Test
-    fun testDoneCompletionStageExceptionally() {
+    fun testCompletedCompletionStageExceptionally() {
         val completable = CompletableFuture<String>()
         val toAwait: CompletionStage<String> = completable
         completable.completeExceptionally(RuntimeException("O"))
@@ -85,11 +107,11 @@ class FutureTest {
                 e.message!!
             } + "K"
         }
-        assertEquals("OK", future.get())
+        assertThat(future.get(), IsEqual("OK"))
     }
 
     @Test
-    fun testAwaitedFutureCompletedExceptionally() {
+    fun testWaitForFutureWithException() {
         val toAwait = CompletableFuture<String>()
         val future = future<String> {
             try {
@@ -100,11 +122,11 @@ class FutureTest {
         }
         assertFalse(future.isDone)
         toAwait.completeExceptionally(RuntimeException("O"))
-        assertEquals("OK", future.get())
+        assertThat(future.get(), IsEqual("OK"))
     }
 
     @Test
-    fun testAwaitedCompletionStageCompletedExceptionally() {
+    fun testWaitForCompletionStageWithException() {
         val completable = CompletableFuture<String>()
         val toAwait: CompletionStage<String> = completable
         val future = future<String> {
@@ -116,7 +138,7 @@ class FutureTest {
         }
         assertFalse(future.isDone)
         completable.completeExceptionally(RuntimeException("O"))
-        assertEquals("OK", future.get())
+        assertThat(future.get(), IsEqual("OK"))
     }
 
     @Test
@@ -125,15 +147,80 @@ class FutureTest {
             if (CompletableFuture.supplyAsync { true }.await()) {
                 throw IllegalStateException("OK")
             }
-            CompletableFuture.supplyAsync { "fail" }.await()
+            "fail"
         }
         try {
             future.get()
             fail("'get' should've throw an exception")
         } catch (e: ExecutionException) {
             assertTrue(e.cause is IllegalStateException)
-            assertEquals("OK", e.cause!!.message)
+            assertThat(e.cause!!.message, IsEqual("OK"))
+        }
+    }
+
+    @Test
+    fun testCompletedDeferredAsCompletableFuture() = runBlocking {
+        expect(1)
+        val deferred = async(context, CoroutineStart.UNDISPATCHED) {
+            expect(2) // completed right away
+            "OK"
+        }
+        expect(3)
+        val future = deferred.asCompletableFuture()
+        assertThat(future.await(), IsEqual("OK"))
+        finish(4)
+    }
+
+    @Test
+    fun testWaitForDeferredAsCompletableFuture() = runBlocking {
+        expect(1)
+        val deferred = async(context) {
+            expect(3) // will complete later
+            "OK"
+        }
+        expect(2)
+        val future = deferred.asCompletableFuture()
+        assertThat(future.await(), IsEqual("OK")) // await yields main thread to deferred coroutine
+        finish(4)
+    }
+
+    @Test
+    fun testCancellableAwaitFuture() = runBlocking {
+        expect(1)
+        val toAwait = CompletableFuture<String>()
+        val job = launch(context, CoroutineStart.UNDISPATCHED) {
+            expect(2)
+            try {
+                toAwait.await() // suspends
+            } catch (e: CancellationException) {
+                expect(5) // should throw cancellation exception
+                throw e
+            }
+        }
+        expect(3)
+        job.cancel() // cancel the job
+        toAwait.complete("fail") // too late, the waiting job was already cancelled
+        expect(4) // job processing of cancellation was scheduled, not executed yet
+        yield() // yield main thread to job
+        finish(6)
+    }
+
+    @Test
+    fun testNonCancellableAwaitCompletionStage() = runBlocking {
+        expect(1)
+        val completable = CompletableFuture<String>()
+        val toAwait: CompletionStage<String> = completable
+        val job = launch(context, CoroutineStart.UNDISPATCHED) {
+            expect(2)
+            assertThat(toAwait.await(), IsEqual("OK")) // suspends
+            expect(5)
         }
+        expect(3)
+        job.cancel() // cancel the job
+        completable.complete("OK") // ok, because await on completion stage is not cancellable
+        expect(4) // job processing of was scheduled, not executed yet
+        yield() // yield main thread to job
+        finish(6)
     }
 
     @Test
@@ -159,7 +246,7 @@ class FutureTest {
             }.await()
             result
         }
-        assertEquals("OK", future.get())
+        assertThat(future.get(), IsEqual("OK"))
     }
 
     private fun wrapContinuation(wrapper: (() -> Unit) -> Unit): CoroutineDispatcher = object : CoroutineDispatcher() {
