Merge pull request #97 from thiml/supplyComposeAsync

Supply compose async

README.md

@@ -182,6 +182,16 @@ CompletionStage<CompletionStage<String>> wrapped = completedFuture(completedFutu
 CompletionStage<String> unwrapped = CompletableFutures.dereference(wrapped);
 ```
 
+#### supplyAsyncCompose
+
+Like `CompletableFuture.supplyAsync` but unwraps a `CompletionStage<CompletionStage<T>>` to a plain `CompletionStage<T>` when
+the `Supplier` returns a `CompletionStage`.
+
+```java
+CompletionStage<String> suppliedStage = completedFuture("hello").thenApply(stage -> stage + "-chained");
+CompletionStage<String> outputStage = CompletableFutures.supplyAsyncCompose(suppliedStage);
+```
+
 #### exceptionallyCompletedFuture
 
 Creates a new future that is already exceptionally completed with the given exception.

src/main/java/com/spotify/futures/CompletableFutures.java

@@ -34,6 +34,7 @@
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.CompletionException;
 import java.util.concurrent.CompletionStage;
+import java.util.concurrent.Executor;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.ScheduledFuture;
 import java.util.concurrent.TimeUnit;
@@ -354,6 +355,42 @@ public static <T> CompletionStage<T> exceptionallyCompose(
     return dereference(wrap(stage).exceptionally(fn));
   }
 
+  /**
+   * This allows for a stage to be the return type of supplier when using supplyAsync and returns a plain stage.
+   *
+   * <p>This differs from {@link
+   * java.util.concurrent.CompletableFuture#supplyAsync(Supplier)} in that the
+   * function should return a {@link java.util.concurrent.CompletionStage} rather than the {@link CompletionStage} of
+   * a {@link CompletionStage} of a value when the return value of the {@link Supplier} is a {@link CompletionStage}.
+   *
+   * @param supplier a {@link Supplier} with a return value of {@link CompletionStage}
+   * @param <U> the type of the supplied stage's value
+   * @return the new {@link CompletionStage}
+   * @since 0.3.6
+   */
+  public static <U> CompletionStage<U> supplyAsyncCompose(Supplier<CompletionStage<U>> supplier) {
+    return dereference(CompletableFuture.supplyAsync(supplier));
+  }
+
+  /**
+   * This allows for a stage to be the return type of supplier when using supplyAsync and returns a plain stage.
+   *
+   * <p>This differs from {@link
+   * java.util.concurrent.CompletableFuture#supplyAsync(Supplier, Executor)} in that the
+   * function should return a {@link java.util.concurrent.CompletionStage} rather than the {@link CompletionStage} of
+   * a {@link CompletionStage} of a value when the return value of the {@link Supplier} is a {@link CompletionStage}.
+   *
+   * @param supplier a {@link Supplier} with a return value of {@link CompletionStage}
+   * @param executor a {@link Executor} the executor to use for asynchronous execution
+   * @param <U> the type of the supplied stage's value
+   * @return the new {@link CompletionStage}
+   * @since 0.3.6
+   */
+  public static <U> CompletionStage<U> supplyAsyncCompose(
+          Supplier<CompletionStage<U>> supplier, Executor executor) {
+    return dereference(CompletableFuture.supplyAsync(supplier, executor));
+  }
+
   /**
    * This takes a stage of a stage of a value and returns a plain stage of a value.
    *

src/test/java/com/spotify/futures/CompletableFuturesTest.java

@@ -514,6 +514,33 @@ public void exceptionallyCompose_returnsNull() {
     assertThat(e.getCause(), is(instanceOf(NullPointerException.class)));
   }
 
+  @Test
+  public void supplyAsyncCompose_completed() {
+    final CompletionStage<String> future = completedFuture("hello");
+    final CompletionStage<String> composed = CompletableFutures.supplyAsyncCompose(() -> future);
+
+    assertThat(composed, completesTo("hello"));
+  }
+
+  @Test
+  public void supplyAsyncCompose_chain_completed() {
+    final CompletionStage<String> future = completedFuture("hello").thenApply(previous -> previous + "-chained");
+    final CompletionStage<String> composed = CompletableFutures.supplyAsyncCompose(() -> future);
+
+    assertThat(composed, completesTo("hello-chained"));
+  }
+
+  @Test
+  public void supplyAsyncCompose_failure() {
+    final IllegalStateException ex = new IllegalStateException();
+    final CompletionStage<String> future = exceptionallyCompletedFuture(ex);
+    final CompletionStage<String> composed = CompletableFutures.supplyAsyncCompose(() -> future);
+
+    final CompletionException e =
+            assertThrows(CompletionException.class, () -> getCompleted(composed));
+    assertThat(e.getCause(), is(equalTo(ex)));
+  }
+
   @Test
   public void handleCompose_completed() {
     final CompletionStage<String> future = exceptionallyCompletedFuture(new Exception("boom"));