) AsyncResult.super.otherwiseEmpty();
}
+ /**
+ * Bridges this Vert.x future to a {@link CompletionStage} instance.
+ *
+ * The {@link CompletionStage} handling methods will be called from the thread that resolves this future.
+ *
+ * @return a {@link CompletionStage} that completes when this future resolves
+ */
+ @GenIgnore(GenIgnore.PERMITTED_TYPE)
+ default CompletionStage toCompletionStage() {
+ CompletableFuture completableFuture = new CompletableFuture<>();
+ this.setHandler(ar -> {
+ if (ar.succeeded()) {
+ completableFuture.complete(ar.result());
+ } else {
+ completableFuture.completeExceptionally(ar.cause());
+ }
+ });
+ return completableFuture;
+ }
+
+ /**
+ * Bridges a {@link CompletionStage} object to a Vert.x future instance.
+ *
+ * The Vert.x future handling methods will be called from the thread that completes {@code completionStage}.
+ *
+ * @param completionStage a completion stage
+ * @param the result type
+ * @return a Vert.x future that resolves when {@code completionStage} resolves
+ */
+ @GenIgnore(GenIgnore.PERMITTED_TYPE)
+ static Future fromCompletionStage(CompletionStage completionStage) {
+ Promise promise = Promise.promise();
+ completionStage.whenComplete((value, err) -> {
+ if (err != null) {
+ promise.fail(err);
+ } else {
+ promise.complete(value);
+ }
+ });
+ return promise.future();
+ }
+
+ /**
+ * Bridges a {@link CompletionStage} object to a Vert.x future instance.
+ *
+ * The Vert.x future handling methods will be called on the provided {@code context}.
+ *
+ * @param completionStage a completion stage
+ * @param context a Vert.x context to dispatch to
+ * @param the result type
+ * @return a Vert.x future that resolves when {@code completionStage} resolves
+ */
+ @GenIgnore(GenIgnore.PERMITTED_TYPE)
+ static Future fromCompletionStage(CompletionStage completionStage, Context context) {
+ Promise promise = ((ContextInternal) context).promise();
+ completionStage.whenComplete((value, err) -> {
+ if (err != null) {
+ promise.fail(err);
+ } else {
+ promise.complete(value);
+ }
+ });
+ return promise.future();
+ }
+
@GenIgnore
FutureFactory factory = ServiceHelper.loadFactory(FutureFactory.class);
diff --git a/src/main/java/io/vertx/core/impl/ContextImpl.java b/src/main/java/io/vertx/core/impl/ContextImpl.java
index 025e0a513..d10462c39 100644
--- a/src/main/java/io/vertx/core/impl/ContextImpl.java
+++ b/src/main/java/io/vertx/core/impl/ContextImpl.java
@@ -29,7 +29,7 @@ import java.util.concurrent.RejectedExecutionException;
* @author Tim Fox
*/
abstract class ContextImpl extends AbstractContext {
-
+
/**
* Execute the {@code task} disabling the thread-local association for the duration
* of the execution. {@link Vertx#currentContext()} will return {@code null},
diff --git a/src/test/java/io/vertx/core/FutureTest.java b/src/test/java/io/vertx/core/FutureTest.java
index b7cbc7d7a..c97608827 100644
--- a/src/test/java/io/vertx/core/FutureTest.java
+++ b/src/test/java/io/vertx/core/FutureTest.java
@@ -21,9 +21,7 @@ import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.List;
-import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.CountDownLatch;
-import java.util.concurrent.TimeUnit;
+import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;
@@ -1424,4 +1422,133 @@ public class FutureTest extends VertxTestBase {
promise.fail(failure);
await();
}
+
+ @Test
+ public void testToCompletionStageTrampolining() {
+ waitFor(2);
+ Thread mainThread = Thread.currentThread();
+ Future success = Future.succeededFuture("Yo");
+ success.toCompletionStage()
+ .thenAccept(str -> {
+ assertEquals("Yo", str);
+ assertSame(mainThread, Thread.currentThread());
+ complete();
+ });
+ Future failed = Future.failedFuture(new RuntimeException("Woops"));
+ failed.toCompletionStage()
+ .whenComplete((str, err) -> {
+ assertNull(str);
+ assertTrue(err instanceof RuntimeException);
+ assertEquals("Woops", err.getMessage());
+ assertSame(mainThread, Thread.currentThread());
+ complete();
+ });
+ await();
+ }
+
+ @Test
+ public void testToCompletionStageDelayedCompletion() {
+ waitFor(2);
+ Thread mainThread = Thread.currentThread();
+ Promise willSucceed = Promise.promise();
+ Promise willFail = Promise.promise();
+
+ willSucceed.future().toCompletionStage().whenComplete((str, err) -> {
+ assertEquals("Yo", str);
+ assertNull(err);
+ assertNotSame(mainThread, Thread.currentThread());
+ complete();
+ });
+
+ willFail.future().toCompletionStage().whenComplete((str, err) -> {
+ assertNull(str);
+ assertTrue(err instanceof RuntimeException);
+ assertEquals("Woops", err.getMessage());
+ assertNotSame(mainThread, Thread.currentThread());
+ complete();
+ });
+
+ disableThreadChecks();
+ new Thread(() -> willSucceed.complete("Yo")).start();
+ new Thread(() -> willFail.fail(new RuntimeException("Woops"))).start();
+ await();
+ }
+
+ @Test
+ public void testFromCompletionStageTrampolining() {
+ waitFor(2);
+ disableThreadChecks();
+
+ AtomicReference successSupplierThread = new AtomicReference<>();
+ CompletableFuture willSucceed = new CompletableFuture<>();
+
+ AtomicReference failureSupplierThread = new AtomicReference<>();
+ CompletableFuture willFail = new CompletableFuture<>();
+
+ Future.fromCompletionStage(willSucceed).onSuccess(str -> {
+ assertEquals("Ok", str);
+ assertSame(successSupplierThread.get(), Thread.currentThread());
+ complete();
+ });
+
+ Future.fromCompletionStage(willFail).onFailure(err -> {
+ assertTrue(err instanceof RuntimeException);
+ assertEquals("Woops", err.getMessage());
+ assertSame(failureSupplierThread.get(), Thread.currentThread());
+ complete();
+ });
+
+ ForkJoinPool fjp = ForkJoinPool.commonPool();
+ fjp.execute(() -> {
+ successSupplierThread.set(Thread.currentThread());
+ willSucceed.complete("Ok");
+ });
+ fjp.execute(() -> {
+ failureSupplierThread.set(Thread.currentThread());
+ willFail.completeExceptionally(new RuntimeException("Woops"));
+ });
+
+ await();
+ }
+
+ @Test
+ public void testFromCompletionStageWithContext() {
+ waitFor(2);
+ Context context = vertx.getOrCreateContext();
+
+ AtomicReference successSupplierThread = new AtomicReference<>();
+ CompletableFuture willSucceed = new CompletableFuture<>();
+
+ AtomicReference failureSupplierThread = new AtomicReference<>();
+ CompletableFuture willFail = new CompletableFuture<>();
+
+ Future.fromCompletionStage(willSucceed, context).onSuccess(str -> {
+ assertEquals("Ok", str);
+ assertNotSame(successSupplierThread.get(), Thread.currentThread());
+ assertEquals(context, vertx.getOrCreateContext());
+ assertTrue(Thread.currentThread().getName().startsWith("vert.x-eventloop-thread"));
+ complete();
+ });
+
+ Future.fromCompletionStage(willFail, context).onFailure(err -> {
+ assertTrue(err instanceof RuntimeException);
+ assertEquals("Woops", err.getMessage());
+ assertNotSame(failureSupplierThread.get(), Thread.currentThread());
+ assertEquals(context, vertx.getOrCreateContext());
+ assertTrue(Thread.currentThread().getName().startsWith("vert.x-eventloop-thread"));
+ complete();
+ });
+
+ ForkJoinPool fjp = ForkJoinPool.commonPool();
+ fjp.execute(() -> {
+ successSupplierThread.set(Thread.currentThread());
+ willSucceed.complete("Ok");
+ });
+ fjp.execute(() -> {
+ failureSupplierThread.set(Thread.currentThread());
+ willFail.completeExceptionally(new RuntimeException("Woops"));
+ });
+
+ await();
+ }
}