139

Java 8 introduces CompletableFuture, a new implementation of Future that is composable (includes a bunch of thenXxx methods). I'd like to use this exclusively, but many of the libraries I want to use return only non-composable Future instances.

Is there a way to wrap up a returned Future instances inside of a CompleteableFuture so that I can compose it?

9 Answers 9

77

If the library you want to use also offers a callback style method in addition to the Future style, you can provide it a handler that completes the CompletableFuture without any extra thread blocking. Like so:

    AsynchronousFileChannel open = AsynchronousFileChannel.open(Paths.get("/some/file"));
    // ... 
    CompletableFuture<ByteBuffer> completableFuture = new CompletableFuture<ByteBuffer>();
    open.read(buffer, position, null, new CompletionHandler<Integer, Void>() {
        @Override
        public void completed(Integer result, Void attachment) {
            completableFuture.complete(buffer);
        }

        @Override
        public void failed(Throwable exc, Void attachment) {
            completableFuture.completeExceptionally(exc);
        }
    });
    completableFuture.thenApply(...)

Without the callback the only other way I see solving this is to use a polling loop that puts all your Future.isDone() checks on a single thread and then invoking complete whenever a Future is gettable.

1
  • 1
    I am using Apache Http async library which accepts FutureCallback. It made my life easy :) Commented Aug 22, 2017 at 19:10
73

If your Future is the result of a call to an ExecutorService method (e.g. submit()), the easiest would be to use the CompletableFuture.runAsync(Runnable, Executor) method instead.

From

Runnbale myTask = ... ;
Future<?> future = myExecutor.submit(myTask);

to

Runnbale myTask = ... ;
CompletableFuture<?> future = CompletableFuture.runAsync(myTask, myExecutor);

The CompletableFuture is then created "natively".

EDIT: Pursuing comments by @SamMefford corrected by @MartinAndersson, if you want to pass a Callable, you need to call supplyAsync(), converting the Callable<T> into a Supplier<T>, e.g. with:

CompletableFuture.supplyAsync(() -> {
    try { return myCallable.call(); }
    catch (Exception ex) { throw new CompletionException(ex); } // Or return default value
}, myExecutor);

Because T Callable.call() throws Exception; throws an exception and T Supplier.get(); doesn't, you have to catch the exception so prototypes are compatible.

A note on exception handling

The get() method doesn't specify a throws, which means it should not throw a checked exception. However, unchecked exception can be used. The code in CompletableFuture shows that CompletionException is used and is unchecked (i.e. is a RuntimeException), hence the catch/throw wrapping any exception into a CompletionException.

Also, as @WeGa indicated, you can use the handle() method to deal with exceptions potentially being thrown by the result:

CompletableFuture<T> future = CompletableFuture.supplyAsync(...);
future.handle((ex,res) -> {
        if (ex != null) {
            // An exception occurred ...
        } else {
            // No exception was thrown, 'res' is valid and can be handled here
        }
    });
5
  • 2
    Or, if you're using Callable<T> rather than Runnable, instead try supplyAsync: CompletableFuture<T> future = CompletableFuture.supplyAsync(myCallable, myExecutor); Commented Apr 28, 2020 at 22:38
  • 1
    supplyAsync receives a Supplier. The code will not compile if you attempt to pass in a Callable. Commented Aug 6, 2020 at 12:26
  • 1
    You can also implement exception handling with java.util.concurrent.CompletableFuture#handle, chained right after CompletableFuture: CompletableFuture.supplyAsync(...).handle((response, throwable) -> ...)
    – WeGa
    Commented Apr 29, 2021 at 6:33
  • What if the API you're dealing with returns a future already?
    – Dragas
    Commented Aug 21 at 10:21
  • @Dragas you can check the other answer
    – Matthieu
    Commented Aug 21 at 20:02
61

There is a way, but you won't like it. The following method transforms a Future<T> into a CompletableFuture<T>:

public static <T> CompletableFuture<T> makeCompletableFuture(Future<T> future) {
  if (future.isDone())
    return transformDoneFuture(future);
  return CompletableFuture.supplyAsync(() -> {
    try {
      if (!future.isDone())
        awaitFutureIsDoneInForkJoinPool(future);
      return future.get();
    } catch (ExecutionException e) {
      throw new RuntimeException(e);
    } catch (InterruptedException e) {
      // Normally, this should never happen inside ForkJoinPool
      Thread.currentThread().interrupt();
      // Add the following statement if the future doesn't have side effects
      // future.cancel(true);
      throw new RuntimeException(e);
    }
  });
}

private static <T> CompletableFuture<T> transformDoneFuture(Future<T> future) {
  CompletableFuture<T> cf = new CompletableFuture<>();
  T result;
  try {
    result = future.get();
  } catch (Throwable ex) {
    cf.completeExceptionally(ex);
    return cf;
  }
  cf.complete(result);
  return cf;
}

private static void awaitFutureIsDoneInForkJoinPool(Future<?> future)
    throws InterruptedException {
  ForkJoinPool.managedBlock(new ForkJoinPool.ManagedBlocker() {
    @Override public boolean block() throws InterruptedException {
      try {
        future.get();
      } catch (ExecutionException e) {
        throw new RuntimeException(e);
      }
      return true;
    }
    @Override public boolean isReleasable() {
      return future.isDone();
    }
  });
}

Obviously, the problem with this approach is, that for each Future, a thread will be blocked to wait for the result of the Future--contradicting the idea of futures. In some cases, it might be possible to do better. However, in general, there is no solution without actively wait for the result of the Future.

8
  • 1
    Ha, that's exactly what I wrote before thinking that there must be a better way. But, I guess not Commented Apr 25, 2014 at 20:24
  • 16
    Hmmm... doesn't this solution eat one of the threads of the "common pool", just for waiting? Those "common pool" threads should never block... hmmmm...
    – Peti
    Commented Dec 3, 2015 at 13:16
  • 4
    It might not be perfect, but using CompletableFuture.supplyAsync(supplier, new SinglethreadExecutor()) would at least not block the common pool threads.
    – MikeFHay
    Commented Jan 27, 2017 at 17:22
  • 8
    Please, just never do that
    – Laymain
    Commented Jul 31, 2018 at 11:27
  • 2
    @MikeFHay I think employing FJP.ManagedBlocker will result in better utilization than CompletableFuture.supplyAsync(supplier, t -> newSingleThreadExecutor()). I've updated the code in the answer to reflect that.
    – leventov
    Commented Oct 30, 2019 at 9:40
15

I published a little futurity project that tries to do better than the straightforward way in the answer.

The main idea is to use only one thread (and of course with not just a spin loop) to check all Futures states inside, which helps to avoid blocking a thread from a pool for each Future -> CompletableFuture transformation.

Usage example:

Future oldFuture = ...;
CompletableFuture profit = Futurity.shift(oldFuture);
2
  • 1
    This looks interesting. Is it using a timer thread ? How come this is not the accepted answer ?
    – Kira
    Commented Mar 20, 2018 at 9:49
  • 1
    @Kira Yeah, it basically uses one timer thread to wait on all submitted futures. Commented Mar 20, 2018 at 22:01
9

Suggestion:

https://gabfssilva.github.io/old-java-future-to-completable-future/

But, basically:

public class CompletablePromiseContext {
    private static final ScheduledExecutorService SERVICE = Executors.newSingleThreadScheduledExecutor();

    public static void schedule(Runnable r) {
        SERVICE.schedule(r, 1, TimeUnit.MILLISECONDS);
    }
}

And, the CompletablePromise:

public class CompletablePromise<V> extends CompletableFuture<V> {
    private Future<V> future;

    public CompletablePromise(Future<V> future) {
        this.future = future;
        CompletablePromiseContext.schedule(this::tryToComplete);
    }

    private void tryToComplete() {
        if (future.isDone()) {
            try {
                complete(future.get());
            } catch (InterruptedException e) {
                completeExceptionally(e);
            } catch (ExecutionException e) {
                completeExceptionally(e.getCause());
            }
            return;
        }

        if (future.isCancelled()) {
            cancel(true);
            return;
        }

        CompletablePromiseContext.schedule(this::tryToComplete);
    }
}

Example:

public class Main {
    public static void main(String[] args) {
        final ExecutorService service = Executors.newSingleThreadExecutor();
        final Future<String> stringFuture = service.submit(() -> "success");
        final CompletableFuture<String> completableFuture = new CompletablePromise<>(stringFuture);

        completableFuture.whenComplete((result, failure) -> {
            System.out.println(result);
        });
    }
}
1
  • 1
    this is quite simple to reason about & elegant & fits most use cases. I would make the CompletablePromiseContext not-static and take param for the check interval (which is set to 1 ms here) then overload the CompletablePromise<V> constructor to be able to provide your own CompletablePromiseContext with a possibly different (longer) check interval for long-running Future<V> where you don't have to absolutely be able to run callback (or compose) immediately upon finishing, and you can also have an instance of CompletablePromiseContext to watch a set of Future (in case you have many) Commented Apr 13, 2019 at 11:51
6

Let me suggest another (hopefully, better) option: https://github.com/vsilaev/java-async-await/tree/master/com.farata.lang.async.examples/src/main/java/com/farata/concurrent

Briefly, the idea is the following:

  1. Introduce CompletableTask<V> interface -- the union of the CompletionStage<V> + RunnableFuture<V>
  2. Warp ExecutorService to return CompletableTask from submit(...) methods (instead of Future<V>)
  3. Done, we have runnable AND composable Futures.

Implementation uses an alternative CompletionStage implementation (pay attention, CompletionStage rather than CompletableFuture):

Usage:

J8ExecutorService exec = J8Executors.newCachedThreadPool();
CompletionStage<String> = exec
   .submit( someCallableA )
   .thenCombineAsync( exec.submit(someCallableB), (a, b) -> a + " " + b)
   .thenCombine( exec.submit(someCallableC), (ab, b) -> ab + " " + c); 
1
4
public static <T> CompletableFuture<T> fromFuture(Future<T> f) {
    return CompletableFuture.completedFuture(null).thenCompose(avoid -> {
        try {
            return CompletableFuture.completedFuture(f.get());
        } catch (InterruptedException e) {
            return CompletableFuture.failedFuture(e);
        } catch (ExecutionException e) {
            return CompletableFuture.failedFuture(e.getCause());
        }
    });
}
0

The main idea goes like this:

Future<?> future = null;
return CompletableFuture.supplyAsync(future::get);

However, you will receive some warnings from your compiler.

So, here is the first option.

Future<?> future = null;
return CompletableFuture.supplyAsync(
        ()->{
            try {
                return future.get();
            } catch (Exception e) {
                throw new RuntimeException(e);
            }
        });

Second Option, hide the try...catch via casting the functional interface.

    @FunctionalInterface
    public interface MySupplier<T> extends Supplier<T> {
        @Override
        default T get() {
            try {
                return getInternal();
            } catch (Exception e) {
                throw new RuntimeException(e);
            }
        }

        T getInternal() throws Exception;
    }


    public static void main(String[] args) {
        Future<?> future = null;
        return CompletableFuture.supplyAsync((MySupplier<?>) future::get);

    }


Third Option, find out some 3rd party lib which has provided such a functional interface.

See Also: Java 8 Lambda function that throws exception?

0

Combining some of the existing answers to this question, while updating and refining them, here's a newer version using an Executor to wait for the Future if it's pending. This implementation makes sure to consider the interrupted flag and to have proper cancellation handling.

A good Executor to use would be Thread::startVirtualThread, but if you don't mind just blocking and waiting for the Future to complete then Runnable::run can also be used as a same-thread Executor.

For older versions of Java see this Gist (with the Java 8 version including some fiddly logic to still properly support cancellations or using a direct/same-thread Executor).

Note:
The passed in Executor should NOT be ForkJoinPool.commonPool(), since that risks exhausting it. If you want to use ForkJoinPool, then instead consider a solution using ForkJoinPool.managedBlock.

public static <T> CompletableFuture<T> toCompletableFuture(Future<T> future, Executor executor) {
    if (future instanceof CompletableFuture<T> completableFuture) {
        return completableFuture;
    }
    return switch (future.state()) {
        case SUCCESS -> CompletableFuture.completedFuture(future.resultNow());
        case FAILED -> CompletableFuture.failedFuture(future.exceptionNow());
        case CANCELLED -> {
            CompletableFuture<T> cf = new CompletableFuture<>();
            cf.cancel(false);
            yield cf;
        }
        case RUNNING -> {
            CompletableFuture<T> cf = new CompletableFuture<>();
            yield cf.completeAsync(() -> {
                try {
                    return future.get();
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                    throw new CompletionException(e);
                } catch (ExecutionException e) {
                    throw new CompletionException(e.getCause());
                } catch (CancellationException e) {
                    // Do this, rather than call cancel, for a cleaner stacktrace
                    cf.completeExceptionally(e);
                    throw e;
                }
            }, executor);
        }
    };
}

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge you have read our privacy policy.

Not the answer you're looking for? Browse other questions tagged or ask your own question.