SameExecutorCompletionStage implements CompletionStage<T> { @Override public final <U> CompletionStage<U> thenCompose(Function<? super T, ? extends CompletionStage<U>> fn) { return of(delegate.thenCompose(fn), defaultExecutor); } SameExecutorCompletionStage(CompletionStage<T> delegate, Executor defaultExecutor); @Override final CompletionStage<U> thenApply(Function<? super T, ? extends U> fn); @Override final CompletionStage<U> thenApplyAsync(Function<? super T, ? extends U> fn); @Override final CompletionStage<U> thenApplyAsync(Function<? super T, ? extends U> fn, Executor executor); @Override final CompletionStage<Void> thenAccept(Consumer<? super T> action); @Override final CompletionStage<Void> thenAcceptAsync(Consumer<? super T> action); @Override final CompletionStage<Void> thenAcceptAsync(Consumer<? super T> action, Executor executor); @Override final CompletionStage<Void> thenRun(Runnable action); @Override final CompletionStage<Void> thenRunAsync(Runnable action); @Override final CompletionStage<Void> thenRunAsync(Runnable action, Executor executor); @Override final CompletionStage<V> thenCombine(CompletionStage<? extends U> other,
            BiFunction<? super T, ? super U, ? extends V> fn); @Override final CompletionStage<V> thenCombineAsync(CompletionStage<? extends U> other,
            BiFunction<? super T, ? super U, ? extends V> fn); @Override final CompletionStage<V> thenCombineAsync(CompletionStage<? extends U> other,
            BiFunction<? super T, ? super U, ? extends V> fn, Executor executor); @Override final CompletionStage<Void> thenAcceptBoth(CompletionStage<? extends U> other,
            BiConsumer<? super T, ? super U> action); @Override final CompletionStage<Void> thenAcceptBothAsync(CompletionStage<? extends U> other,
            BiConsumer<? super T, ? super U> action); @Override final CompletionStage<Void> thenAcceptBothAsync(CompletionStage<? extends U> other,
            BiConsumer<? super T, ? super U> action, Executor executor); @Override final CompletionStage<Void> runAfterBoth(CompletionStage<?> other, Runnable action); @Override final CompletionStage<Void> runAfterBothAsync(CompletionStage<?> other, Runnable action); @Override final CompletionStage<Void> runAfterBothAsync(CompletionStage<?> other, Runnable action, Executor executor); @Override final CompletionStage<U> applyToEither(CompletionStage<? extends T> other, Function<? super T, U> fn); @Override final CompletionStage<U> applyToEitherAsync(CompletionStage<? extends T> other, Function<? super T, U> fn); @Override final CompletionStage<U> applyToEitherAsync(CompletionStage<? extends T> other, Function<? super T, U> fn,
            Executor executor); @Override final CompletionStage<Void> acceptEither(CompletionStage<? extends T> other, Consumer<? super T> action); @Override final CompletionStage<Void> acceptEitherAsync(CompletionStage<? extends T> other, Consumer<? super T> action); @Override final CompletionStage<Void> acceptEitherAsync(CompletionStage<? extends T> other, Consumer<? super T> action,
            Executor executor); @Override final CompletionStage<Void> runAfterEither(CompletionStage<?> other, Runnable action); @Override final CompletionStage<Void> runAfterEitherAsync(CompletionStage<?> other, Runnable action); @Override final CompletionStage<Void> runAfterEitherAsync(CompletionStage<?> other, Runnable action, Executor executor); @Override final CompletionStage<U> thenCompose(Function<? super T, ? extends CompletionStage<U>> fn); @Override final CompletionStage<U> thenComposeAsync(Function<? super T, ? extends CompletionStage<U>> fn); @Override final CompletionStage<U> thenComposeAsync(Function<? super T, ? extends CompletionStage<U>> fn,
            Executor executor); @Override final CompletionStage<T> exceptionally(Function<Throwable, ? extends T> fn); @Override final CompletionStage<T> whenComplete(BiConsumer<? super T, ? super Throwable> action); @Override final CompletionStage<T> whenCompleteAsync(BiConsumer<? super T, ? super Throwable> action); @Override final CompletionStage<T> whenCompleteAsync(BiConsumer<? super T, ? super Throwable> action, Executor executor); @Override final CompletionStage<U> handle(BiFunction<? super T, Throwable, ? extends U> fn); @Override final CompletionStage<U> handleAsync(BiFunction<? super T, Throwable, ? extends U> fn); @Override final CompletionStage<U> handleAsync(BiFunction<? super T, Throwable, ? extends U> fn, Executor executor); @Override final CompletableFuture<T> toCompletableFuture();  }
@Test public void testThenCompose() throws InterruptedException { CompletionStage<Void> firstCompletionStage = Async.supplyAsync(latchSupplier(latch1), executorA); latch1.countDown(); firstCompletionStage.toCompletableFuture().join(); assertHits(executorA, 1); Function<Void, CompletableFuture<Void>> fn = x -> CompletableFuture.completedFuture(x); CompletionStage<Void> secondCompletionStage = firstCompletionStage.thenCompose(fn); secondCompletionStage.toCompletableFuture().join(); assertHits(executorA, 1); Function<Void, CompletableFuture<Void>> fn2 = Unchecked.function(x -> { latch2.await(); return CompletableFuture.completedFuture(x); }); CompletionStage<Void> thirdCompletionStage = secondCompletionStage.thenComposeAsync(fn2); latch2.countDown(); thirdCompletionStage.toCompletableFuture().join(); assertHits(executorA, 2); Function<Void, CompletableFuture<Void>> fn3 = Unchecked.function(x -> { latch3.await(); return CompletableFuture.completedFuture(x); }); CompletionStage<Void> fourthCompletionStage = thirdCompletionStage.thenComposeAsync(fn3, executorB); latch3.countDown(); fourthCompletionStage.toCompletableFuture().join(); assertHits(executorA, 2); assertHits(executorB, 1); }