Java 8 - Process list of elements in CompleteableFuture's thenCompose method

user_3340573 0
Viewed 795

Please find below a made-up example of my actual code. This example has been oversimplified to explain what I am trying to achieve.

public class TestClass {

ForkJoinPool forkJoinPool = new ForkJoinPool(3);

@Test 
public void testSample(){
    List<String> testStrings = Arrays.asList("Hello", "World", "Cobra", "Eagle", "Sam");

    //This doesn't compile
    List<CompletableFuture<Double>> result =
            testStrings.stream().map(each -> CompletableFuture.supplyAsync(() -> getIndividualCharacters(each), forkJoinPool)
                    .thenComposeAsync(listOfChars -> listOfChars.stream()
                            .map(character -> CompletableFuture.supplyAsync(() -> getDoubleString(character)))
                            .collect(Collectors.toList())));

}

public List<String> getIndividualCharacters(String name){
    List<String> result = new ArrayList<>();
    for(int i =0; i < name.length(); i++){
        result.add(Character.toString(name.charAt(i)));
    }
    return result;
}

public Double getDoubleString(String singleCharacter){
    return Math.random();
}

}

My getIndividualCharacters method returns a list of results (Asynchronously). I use the individual result and process it further to return another result (Asynchronously). What I want as the end result is a List<Completeable<final result>> in this case List<Completeable<Double>> which I can use inside CompleteablFuture.allOf

I want to use chaining of CompleteableFuture if possible. I haven't managed to find a way to do it nor any examples mention it. Any help or pointers on how to achieve this will be really helpful.

PS: I have managed to solve the problem using two separate CompleteableFuture streams, however, I want to use chaining thenCompose

List<String> testStrings = Arrays.asList("Hello", "World", "Cobra", "Eagle", "Sam");
        List<CompletableFuture<List<String>>> firstResult = testStrings.stream()
                .map(each -> CompletableFuture.supplyAsync(() -> getIndividualCharacters(each), forkJoinPool))
                .collect(Collectors.toList());
        CompletableFuture.allOf(firstResult.toArray(new CompletableFuture[firstResult.size()])).join();
        List<CompletableFuture<Double>> secondResult = firstResult.stream()
                .flatMap(res -> res.join().stream())
                .map(ea -> CompletableFuture.supplyAsync(() -> getDoubleString(ea), forkJoinPool))
                .collect(Collectors.toList());
        List<Double> finalResult = secondResult.stream().map(res-> res.join()).collect(Collectors.toList());
        System.out.println("finalResult " + finalResult);

Regards.

completable-future java-8 asynchronous lambda java
2 Answers

I guess something like this?

List<CompletableFuture<Double>> result =
  testStrings.stream()
             .map(x -> CompletableFuture.supplyAsync(() -> getIndividualCharacters(x)))
             .map(x -> x.thenApplyAsync(y -> y.stream().map(z -> CompletableFuture.supplyAsync(() -> getDoubleString(z)))))
             .flatMap(CompletableFuture::join)
             .collect(toList());

Just notice that this will block because of join.

user_10593720

I'll add another answer, may be this can be of a solution to you. You could delay the execution of join from the previous answer only until you would need it. For example via:

 Stream<CompletableFuture<Double>> result = 
 testStrings.stream()
            .map(each -> supplyAsync(() -> getIndividualCharacters(each)))
            .map(x -> x.thenCompose(y -> supplyAsync(() -> y.stream().map(z -> supplyAsync(() -> getDoubleString(z))))))
            .flatMap(x -> Stream.of(x).map(CompletableFuture::join))
            .flatMap(Function.identity());

But this changes the result to a Stream, at the same time - the stream is only evaluated when a terminal operation is called.

user_10593720
Related