CODEWITHSUNDEEP
javajava-8reactive-programmingvert.xcompletable-future
zafar142003
zafar142003

Reputation: 2149

How do I run a CompletableFuture handler on the Vertx event loop thread?

I have a library xyz that gives me a CompletableFuture which I want to process on my Vertx (v3.5) event loop. Currently I am using CompletableFuture.handle(BiFunction) (see below) but I would like to use CompletableFuture.handleAsync(BiFunction, Executor), but I cannot figure out how to provide the vertx event loop thread to the second parameter in this method call.

I tried executing this whole code in Vertx.runOnContext() but the calls inside handleAsync still executed on Java's ForkJoin pool which I want to avoid.

CompletableFuture<Void> f = xyz.someMethod();
f.handle((v, th) -> { //Want to run this in handleAsync()
   if (th == null) {
     future.complete(null);
   } else {
     future.completeExceptionally(th);
   }
   return null;
});

Upvotes: 4

Views: 4851

Answers (2)

ledniov
ledniov

Reputation: 2382

You can achieve this by simply using vertx.nettyEventLoopGroup() as the second parameter, so your code will be something like this:

CompletableFuture<Void> f = xyz.someMethod();
f.handleAsync((v, th) -> { 

  if (th == null) {
    future.complete(null);
  } else {
    future.completeExceptionally(th);
  }

  return null;
}, vertx.nettyEventLoopGroup());

Note: the above code will may run the callback code not on the same thread the Vertx future was running.

To preserve the threading model of the Vertx you need to use the following code:

CompletableFuture<String> f = xyz.someMethod();
Context context = vertx.getOrCreateContext();
f.handleAsync((v, th) -> {
    context.runOnContext(event -> {
        if (th == null) {
            future.complete(null);
        } else {
            future.completeExceptionally(th);
        }
    });
    return null;
});

Upvotes: 8

zafar142003
zafar142003

Reputation: 2149

This can be done the following way. It is a workaround, not an optimal solution. Initially I placed the runOnContext call outside handle method, which is why it did not work.

CompletableFuture<Void> f = xyz.someMethod(); f.handle((v, th) -> { vertx.runOnContext(e->{ if (th == null) { future.complete(null); } else { future.completeExceptionally(th); } }); return null; });

This will make one extra context switch (from ForkJoin pool to vertx event loop) which is one disadvantage of this approach.

Upvotes: 0

Related Questions