Completed futures should operate on provided context

SucceededFuture and FailedFuture, like other futures, take a context as constructor param.
If not null, this context should be used to dispatch the execution of future handlers.

Signed-off-by: Thomas Segismont <tsegismont@gmail.com>

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,38 @@ public class FutureTest extends VertxTestBase {
 
     await();
   }
+
+  @Test
+  public void testCompletedFuturesContext() {
+    waitFor(4);
+
+    Thread testThread = Thread.currentThread();
+    ContextInternal context = (ContextInternal) vertx.getOrCreateContext();
+
+    Future.succeededFuture().onSuccess(v -> {
+      assertSame(testThread, Thread.currentThread());
+      assertNull(Vertx.currentContext());
+      complete();
+    });
+
+    context.succeededFuture().onSuccess(v -> {
+      assertNotSame(testThread, Thread.currentThread());
+      assertSame(context, Vertx.currentContext());
+      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());
+      complete();
+    });
+
+    await();
+  }
 }