CompletableFutures { static <F, T> Function<F, CompletableFuture<Stream<T>>> fanOut(Collection<Function<F, T>> mappings) { return fanOut(mappings, ForkJoinPool.commonPool()); } private  CompletableFutures(); static BinaryOperator<CompletableFuture<T>> combineUsing(BiFunction<T, T, T> combiningFunction); static BinaryOperator<CompletableFuture<T>> combineUsing(BiFunction<T, T, T> combiningFunction, ExecutorService executorService);  }
@Test public void shouldExecuteAllFunctionsFromMappings() throws Exception { CompletableFuture<String> future = CompletableFuture.completedFuture("SampleText"); CompletableFuture<Stream<Integer>> actualFuture = future.thenCompose( fanOut(ImmutableList.of( text -> 1, text -> 2, text -> 4 )) ); CompletableFuture<Integer> actualSum = actualFuture.thenApply( stream -> stream.reduce((a, b) -> a + b).orElse(-1) ); assertThat(actualSum.get(), is(equalTo(7))); }
@Test public void shouldExecuteLongRunningOutliers() throws Exception { CompletableFuture<String> future = CompletableFuture.completedFuture("SampleText"); CompletableFuture<Stream<Integer>> actualFuture = future.thenCompose( fanOut(ImmutableList.of( text -> 1, text -> 2, text -> { waitThreeSecond(); return 4; } )) ); CompletableFuture<Integer> actualSum = actualFuture.thenApply( stream -> stream.reduce((a, b) -> a + b).orElse(-1) ); assertThat(actualSum.get(), is(equalTo(7))); }
@Test public void shouldExecuteAllFunctionsAtTheSameTime() throws Exception { CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> "SampleText"); ExecutorService pool = Executors.newFixedThreadPool(4); CompletableFuture<Stream<Integer>> actualFuture = future.thenComposeAsync( fanOut(ImmutableList.of( text -> { waitThreeSecond(); return 1; }, text -> { waitThreeSecond(); return 2; }, text -> { waitThreeSecond(); return 4; }) ) ); try { actualFuture.get(4, TimeUnit.SECONDS); } catch (TimeoutException e) { fail("Should have finished in less than four seconds"); } }