Merge pull request #3278 from tsegismont/completed-futures-context

Completed futures should operate on provided context

src/main/java/io/vertx/core/impl/FailedFuture.java

@@ -52,7 +52,11 @@ public class FailedFuture<T> implements Future<T> {
 
   @Override
   public Future<T> onComplete(Handler<AsyncResult<T>> handler) {
-    handler.handle(this);
+    if (context != null) {
+      context.dispatch(this, handler);
+    } else {
+      handler.handle(this);
+    }
     return this;
   }
 

src/main/java/io/vertx/core/impl/SucceededFuture.java

@@ -45,7 +45,11 @@ class SucceededFuture<T> implements Future<T> {
 
   @Override
   public Future<T> onComplete(Handler<AsyncResult<T>> handler) {
-    handler.handle(this);
+    if (context != null) {
+      context.dispatch(this, handler);
+    } else {
+      handler.handle(this);
+    }
     return this;
   }
 

src/test/java/io/vertx/core/FutureTest.java

@@ -21,7 +21,10 @@ import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collections;
 import java.util.List;
-import java.util.concurrent.*;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ForkJoinPool;
+import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicReference;
@@ -1570,4 +1573,45 @@ public class FutureTest extends VertxTestBase {
 
     await();
   }
+
+  @Test
+  public void testCompletedFuturesContext() throws Exception {
+    waitFor(4);
+
+    Thread testThread = Thread.currentThread();
+    ContextInternal context = (ContextInternal) vertx.getOrCreateContext();
+
+
+    CompletableFuture<Thread> cf = new CompletableFuture<>();
+    context.runOnContext(v -> cf.complete(Thread.currentThread()));
+    Thread contextThread = cf.get();
+
+    Future.succeededFuture().onSuccess(v -> {
+      assertSame(testThread, Thread.currentThread());
+      assertNull(Vertx.currentContext());
+      complete();
+    });
+
+    context.succeededFuture().onSuccess(v -> {
+      assertNotSame(testThread, Thread.currentThread());
+      assertSame(context, Vertx.currentContext());
+      assertSame(contextThread, Thread.currentThread());
+      complete();
+    });
+
+    Future.failedFuture(new Exception()).onFailure(v -> {
+      assertSame(testThread, Thread.currentThread());
+      assertNull(Vertx.currentContext());
+      complete();
+    });
+
+    context.failedFuture(new Exception()).onFailure(v -> {
+      assertNotSame(testThread, Thread.currentThread());
+      assertSame(context, Vertx.currentContext());
+      assertSame(contextThread, Thread.currentThread());
+      complete();
+    });
+
+    await();
+  }
 }