CompletableFuture Loop API Calls and Map to Object

user_19776202 0
Viewed 61

I am having some trouble while experimenting with CompletableFutures

What I am trying to do is break up a list of Objects into partitions (smaller Object lists) using Google Guava, then asynchronously spin up new threads to call an API with each subset list of Objects. I want to wait until all these futures are completed, then combine these results into a single return object. Here is a non-working example of what I'm trying to do. This example does not compile due to this line:

return CompletableFuture.allOf(completableFutures.toArray(new CompletableFuture[completableFutures.size()])).join();

The method expects a Position return type but CompletableFuture.allOf() returns Void. However, I thought that adding .join() should give me the desired Position return type. This is where I am stuck.

public Position fetchPosition(List<UnitDTO> unitDTOList) throws ExecutionException, InterruptedException {
    List<List<UnitDTO>> subsets = Lists.partition(unitDTOList, 2);
    List<CompletableFuture<Position>> completableFutures = new ArrayList<>();

    for(List<UnitDTO> subset : subsets){
      completableFutures.add(getPositionDetailsAsync(subset));
    }
    return CompletableFuture.allOf(completableFutures.toArray(new CompletableFuture[completableFutures.size()])).join();
}

public CompletableFuture<Position> getPositionDetailsAsync(List<UnitDTO> unitDTOList){
    return CompletableFuture.supplyAsync(() -> {
      Position position = new Position();
      try {
        position = myApiClient.getPosition();
      } catch (URISyntaxException e) {
        e.printStackTrace();
      }
      return position;
    });

}

I know I'm not giving much to work with here but basically I am not sure if this is the proper approach or if I am on the right track. Hopefully my issue makes sense. I'm happy to provide more information if needed.

completable-future java
1 Answers

You need to think of how you want to reduce the list of Position values into a single Position. Then you can use Stream.reduce to convert a Stream<CompletableFuture<Position>> into a CompletableFuture<Position> by lifting that reduction function into the stream:

static <T> CompletableFuture<T> reduce(
        Stream<CompletableFuture<T>> futures,
        T identity,
        BinaryOperator<T> accumulator)
{
    return futures.reduce(
        completedFuture(identity),
        (f1, f2) ->
            f1.thenCompose(s1 ->
            f2.thenApply(s2 ->
                accumulator.apply(s1, s2))));
}

For example, to concatenate some strings wrapped in CompletableFutures you would do

Stream<CompletableFuture<String>> futures = Stream.of(
    completedFuture("A"),
    completedFuture("mouse"),
    completedFuture("took"),
    completedFuture("a"),
    completedFuture("stroll"));

System.out.println(
    reduce(futures, "", (s1, s2) -> s1 + " " + s2).get());

Which prints "A mouse took a stroll"

user_4024280
Related