Convert from List to CompletableFuture

by

Use CompletableFuture.allOf(...):

static<T> CompletableFuture<List<T>> sequence(List<CompletableFuture<T>> com) {
    return CompletableFuture.allOf(com.toArray(new CompletableFuture<?>[0]))
            .thenApply(v -> com.stream()
                .map(CompletableFuture::join)
                .collect(Collectors.toList())
            );
}

A few comments on your implementation:

Your use of .thenComposeAsync, .thenApplyAsync and .thenCombineAsync is likely not doing what you expect. These ...Async methods run the function supplied to them in a separate thread. So, in your case, you are causing the addition of the new item to the list to run in the supplied executor. There is no need to stuff light-weight operations into a cached thread executor. Do not use thenXXXXAsync methods without a good reason.

Additionally, reduce should not be used to accumulate into mutable containers. Even though it might work correctly when the stream is sequential, it will fail if the stream were to be made parallel. To perform mutable reduction, use .collect instead.

If you want to complete the entire computation exceptionally immediately after the first failure, do the following in your sequence method:

CompletableFuture<List<T>> result = CompletableFuture.allOf(com.toArray(new CompletableFuture<?>[0]))
        .thenApply(v -> com.stream()
                .map(CompletableFuture::join)
                .collect(Collectors.toList())
        );

com.forEach(f -> f.whenComplete((t, ex) -> {
    if (ex != null) {
        result.completeExceptionally(ex);
    }
}));

return result;

If, additionally, you want to cancel the remaining operations on first failure, add exec.shutdownNow(); right after result.completeExceptionally(ex);. This, of course, assumes that exec only exist for this one computation. If it doesn’t, you’ll have to loop over and cancel each remaining Future individually.

More Related Contents:

  1. Custom thread pool in Java 8 parallel stream
  2. Parallel streams, collectors and thread safety
  3. Why should I use concurrent characteristic in parallel stream with collect?
  4. CompletableFuture recoverWith equivalent? i.e. exceptionally but return CompletableFuture
  5. Nested Java 8 parallel forEach loop perform poor. Is this behavior expected?
  6. Will a chain of method calls (CompletableFuture API) execute asynchronously if the first method in a chain is asynchronous?
  7. How to combine 3 or more CompletionStages?
  8. Java 8 Supplier Exception handling with CompletableFuture
  9. How to interrupt underlying execution of CompletableFuture
  10. How to wait for all threads to finish, using ExecutorService?
  11. What is the difference between Thread.start() and Thread.run()?
  12. Explicitly calling a default method in Java
  13. Swing rendering appears broken in JDK 1.8, correct in JDK 1.7
  14. Why use a ReentrantLock if one can use synchronized(this)?
  15. Java8: ambiguity with lambdas and overloaded methods
  16. Servlet seems to handle multiple concurrent browser requests synchronously
  17. Is this a bug in Files.lines(), or am I misunderstanding something about parallel streams?
  18. Can a Collector’s combiner function ever be used on sequential streams?
  19. String Deduplication feature of Java 8
  20. How can CopyOnWriteArrayList be thread-safe?
  21. Memory barriers and coding style over a Java VM
  22. Java8 Lambdas and Exceptions
  23. Why can method reference use non-final variables?
  24. stream on JPA lazy list
  25. Uninitialized object leaked to another thread despite no code explicitly leaking it?
  26. Does Java 8 provide a good way to repeat a value or function?
  27. How to short-circuit a reduce() operation on a Stream?
  28. Want to create a stream of characters from char array in java
  29. Creating String representation of lambda expression [duplicate]
  30. Spring 3.2.x with Java 8

Leave a Comment