This tutorial shows the use of CompletionStage methods which compose this CompletionStage<T> with another CompletionStage<U>.

Combining vs Composing

Combining (last tutorial) is a process where a BiFunction/BiConsumer is used for combining the results from two stages into one, whereas composing is the process where the result of the first stage is fed into the second stage to get a composite result.

Composing methods

CompletionStage<U> thenCompose(Function<? super T, ? extends CompletionStage<U>> fn)

CompletionStage<U> thenComposeAsync(Function<? super T, ? extends CompletionStage<U>> fn)

CompletionStage<U> thenComposeAsync(Function<? super T, ?extends CompletionStage<U>> fn, Executor executor)

When this stage completes normally, the given function ('fn') is invoked with this stage's result (T) as the argument, returning another CompletionStage. When that stage completes normally, the CompletionStage returned by this method is completed with the same value.

In other words the result of this stage is 'mapped' to other value/type via a function returning the other stage instance. The other stage is used for the actual mapping.

Why do thenCompose(....) methods need a function? Why not use the other stage directly? The answer is; without a function, the result of the first completion stage will not be available to the second stage.

Analogous to Stream.flatMap()

What if, instead of using thenCompose() methods, we try to compose this stage with another stage via method like CompletionStage.thenApply(fn) (thenApply() tutorial). That will actually return CompletionStage of CompletionStage:

 CompletionStage<T> thisStage = ...;
 Function<T, CompletionStage<U>> fn = ...;
 //using CompletionStage<U> thenApply(Function<? super T, ? extends U> fn)
 CompletionStage<CompletionStage<U>> newStageOfStage = thisStage.thenApply(fn);
 //now using CompletionStage<U> thenCompose(Function<? super T, ? extends CompletionStage<U>> fn)
 CompletionStage<U> newStage = thisStage.thenCompose(fn);

In that sense thenCompose() methods are analogous to Stream.flatMap(....).
In similar situation, Stream.map() returns Stream<Stream<T>>, whereas, Stream.flatMap(....) method returns a flat stream.

Examples

In the following example, the first stage (the main stage) returns a random integer, the second stage uses another random integer and finds it's value (in BigDecimal) raised to the power of the first one.

public class ThenComposeExample {
  public static void main(String[] args) {
      CompletableFuture<Integer> rndNumCompletableFuture = CompletableFuture
              .supplyAsync(() -> {
                  int x = ThreadLocalRandom.current().nextInt(1, 5);
                  System.out.println("Main stage generated number: " + x);
                  return x;
              });

      CompletableFuture<BigDecimal> composedCompletableFuture =
              rndNumCompletableFuture.thenCompose(ThenComposeExample::getPowerCompletableFuture);
      BigDecimal power = composedCompletableFuture.join();
      System.out.println("power: " + power);
  }

  //the other function to be used as method reference.
  private static CompletableFuture<BigDecimal> getPowerCompletableFuture(Integer x) {
      CompletableFuture<BigDecimal> completableFuture = CompletableFuture
              .supplyAsync(() -> {
                  int y = ThreadLocalRandom.current().nextInt(5, 10);
                  System.out.println("other stage generated number: " + y);
                  return new BigDecimal(y).pow(x);
              });
      return completableFuture;
  }
}

Main stage generated number: 4
other stage generated number: 8
power: 4096

Following is another example which applies composition of three stages:

public class ThenComposeExample2 {

  public static void main(String[] args) {
      System.out.println("-- Main CompletableFuture: finding a random number x --");
      CompletableFuture<Integer> firstCompletableFuture = CompletableFuture
              .supplyAsync(() -> {
                  int x = ThreadLocalRandom.current().nextInt(1, 5);
                  System.out.println("x: " + x);
                  return x;
              });

      CompletableFuture<Long> composedCompletableFuture =
              firstCompletableFuture.thenComposeAsync(ThenComposeExample2::getSecondCompletableFuture);
      CompletableFuture<Double> secondComposedCompletableFuture =
              composedCompletableFuture.thenComposeAsync(ThenComposeExample2::getThirdCompletableFuture);
      Double sqrt = secondComposedCompletableFuture.join();
      System.out.println(sqrt);
  }

  private static CompletableFuture<Long> getSecondCompletableFuture(Integer x) {
      System.out.println("-- 2nd CompletableFuture: finding sum of random numbers raised to the power x --");
      return CompletableFuture
              .supplyAsync(() -> {
                  return IntStream.range(1, 5)
                                  .mapToLong(i -> ThreadLocalRandom.current().nextLong(5, 10))
                                  .peek(System.out::print)
                                  .map(y -> (long) Math.pow(y, x))
                                  .peek(z -> System.out.println(" - "+z))
                                  .sum();
              });
  }

  private static CompletableFuture<Double> getThirdCompletableFuture(Long sum) {
      System.out.printf("-- 3rd CompletableFuture: finding square root of sum: %s --%n", sum);
      return CompletableFuture.supplyAsync(() -> {
          return Math.sqrt(sum);
      });
  }
}

-- Main CompletableFuture: finding a random number x --
x: 3
-- 2nd CompletableFuture: finding sum of random numbers raised to the power x --
9 - 729
7 - 343
6 - 216
5 - 125
-- 3rd CompletableFuture: finding square root of sum: 1413 --
37.589892258425

Example Project

Dependencies and Technologies Used:

• JDK 11
• Maven 3.5.4