Merge 20b8d21 into 2232323

pom.xml

@@ -16,18 +16,20 @@
     <profile>
       <id>jdk7</id>
       <activation>
-        <jdk>[1.6,1.7)</jdk>
+        <jdk>[1.6,1.8)</jdk>
       </activation>
       <properties>
+        <junitincludes>**/jdk7/*Test.class</junitincludes>
         <javadocparam />
       </properties>
     </profile>
     <profile>
       <id>jdk8</id>
       <activation>
-        <jdk>1.8</jdk>
+        <jdk>[1.8,)</jdk>
       </activation>
       <properties>
+        <junitincludes>**/*Test.class</junitincludes>
         <javadocparam>-Xdoclint:none</javadocparam>
       </properties>
     </profile>
@@ -95,7 +97,7 @@
     <dependency>
       <groupId>junit</groupId>
       <artifactId>junit</artifactId>
-      <version>4.11</version>
+      <version>4.12</version>
       <scope>test</scope>
     </dependency>
     <dependency>
@@ -161,6 +163,16 @@
           </execution>
         </executions>
       </plugin>
+      <plugin>
+        <groupId>org.apache.maven.plugins</groupId>
+        <artifactId>maven-surefire-plugin</artifactId>
+        <version>2.19</version>
+        <configuration>
+          <includes>
+            <include>${junitincludes}</include>
+          </includes>
+        </configuration>
+      </plugin>
       <plugin>
         <groupId>org.apache.maven.plugins</groupId>
         <artifactId>maven-javadoc-plugin</artifactId>

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

@@ -15,9 +15,14 @@
  */
 package com.spotify.futures;
 
+import com.google.common.base.Throwables;
 import com.google.common.util.concurrent.ListenableFuture;
 
 import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CompletionException;
+import java.util.concurrent.CompletionStage;
+import java.util.function.BiFunction;
+import java.util.function.Function;
 
 public class CompletableFuturesExtra {
 
@@ -55,4 +60,132 @@ public static <V> ListenableFuture<V> toListenableFuture(
     }
     return new CompletableToListenableFutureWrapper<V>(future);
   }
+
+  /**
+   * Returns a new CompletableFuture that is already exceptionally completed with
+   * the given exception.
+   *
+   * @param throwable the exception
+   * @return the exceptionally completed CompletableFuture
+   */
+  public static <T> CompletableFuture<T> exceptionallyCompletedFuture(Throwable throwable) {
+    final CompletableFuture<T> future = new CompletableFuture<T>();
+    future.completeExceptionally(throwable);
+    return future;
+  }
+
+  /**
+   * Returns a new CompletionStage that, when this stage completes
+   * either normally or exceptionally, is executed with this stage's
+   * result and exception as arguments to the supplied function.
+   *
+   * <p>When this stage is complete, the given function is invoked
+   * with the result (or {@code null} if none) and the exception (or
+   * {@code null} if none) of this stage as arguments, and the
+   * function's result is used to complete the returned stage.
+   *
+   * This differs from
+   * {@link java.util.concurrent.CompletionStage#handle(java.util.function.BiFunction)}
+   * in that the function should return a {@link java.util.concurrent.CompletionStage} rather than
+   * the value directly.
+   *
+   * @param fn the function to use to compute the value of the
+   * returned CompletionStage
+   * @param <U> the function's return type
+   * @return the new CompletionStage
+   */
+  public static <T, U> CompletionStage<U> handleCompose(
+          CompletionStage<T> future,
+          final BiFunction<? super T, Throwable, ? extends CompletionStage<U>> fn) {
+    return dereference(future.handle(fn));
+  }
+
+  /**
+   * Returns a new CompletionStage that, when this stage completes
+   * exceptionally, is executed with this stage's exception as the
+   * argument to the supplied function.  Otherwise, if this stage
+   * completes normally, then the returned stage also completes
+   * normally with the same value.
+   *
+   * This differs from
+   * {@link java.util.concurrent.CompletionStage#exceptionally(java.util.function.Function)}
+   * in that the function should return a {@link java.util.concurrent.CompletionStage} rather than
+   * the value directly.
+   *
+   * @param fn the function to use to compute the value of the
+   * returned CompletionStage if this CompletionStage completed
+   * exceptionally
+   * @return the new CompletionStage
+   */
+  public static <T> CompletionStage<T> exceptionallyCompose(
+          CompletionStage<T> future,
+          final Function<Throwable, ? extends CompletionStage<T>> fn) {
+    return dereference(wrap(future).exceptionally(fn));
+  }
+
+  /**
+   * This takes a stage of a stage of a value and returns a plain future of a value.
+   *
+   * @param future of a future of a value
+   * @return the completion stage of a value
+   */
+  public static <T> CompletionStage<T> dereference(
+          CompletionStage<? extends CompletionStage<T>> future) {
+    //noinspection unchecked
+    return future.thenCompose(Identity.INSTANCE);
+  }
+
+  /**
+   * check that a stage is completed.
+   * @param stage the stage.
+   * @throws IllegalStateException if the stage is not completed.
+   */
+  public static void checkCompleted(CompletionStage<?> stage) {
+    if (!stage.toCompletableFuture().isDone()) {
+      throw new IllegalStateException("future was not completed");
+    }
+  }
+
+  /**
+   * Get the value of a completed stage.
+   *
+   * @param stage a completed stage.
+   * @return the value of the stage if it has one.
+   * @throws IllegalStateException if the stage is not completed.
+   * @throws com.google.common.util.concurrent.UncheckedExecutionException
+   * if the future has failed with a non-runtime exception, otherwise
+   * the actual exception
+   */
+  public static <T> T getCompleted(CompletionStage<T> stage) {
+    CompletableFuture<T> future = stage.toCompletableFuture();
+    checkCompleted(future);
+    try {
+      return future.join();
+    } catch (CompletionException e) {
+      throw Throwables.propagate(e.getCause());
+    }
+  }
+
+  private enum Identity implements Function {
+    INSTANCE;
+
+    @Override
+    public Object apply(Object o) {
+      return o;
+    }
+  }
+
+  private enum WrapFunction implements Function {
+    INSTANCE;
+
+    @Override
+    public Object apply(Object o) {
+      return CompletableFuture.completedFuture(o);
+    }
+  }
+
+  private static <T> CompletionStage<CompletionStage<T>> wrap(CompletionStage<T> future) {
+    //noinspection unchecked
+    return future.thenApply((Function<T, CompletionStage<T>>) WrapFunction.INSTANCE);
+  }
 }

src/test/java/com/spotify/futures/AsyncRetrierTest.java → src/test/java/com/spotify/futures/jdk7/AsyncRetrierTest.java

@@ -14,11 +14,12 @@
  * the License.
  */
 
-package com.spotify.futures;
+package com.spotify.futures.jdk7;
 
 import com.google.common.base.Predicate;
 import com.google.common.base.Supplier;
 import com.google.common.util.concurrent.ListenableFuture;
+import com.spotify.futures.AsyncRetrier;
 import org.jmock.lib.concurrent.DeterministicScheduler;
 import org.junit.Before;
 import org.junit.Test;

src/test/java/com/spotify/futures/FuturesExtraTest.java → src/test/java/com/spotify/futures/jdk7/FuturesExtraTest.java

@@ -13,7 +13,7 @@
  * License for the specific language governing permissions and limitations under
  * the License.
  */
-package com.spotify.futures;
+package com.spotify.futures.jdk7;
 
 import com.google.common.base.Function;
 import com.google.common.collect.Lists;
@@ -22,7 +22,9 @@
 import com.google.common.util.concurrent.ListenableFuture;
 import com.google.common.util.concurrent.SettableFuture;
 import com.google.common.util.concurrent.UncheckedExecutionException;
+import com.spotify.futures.FuturesExtra;
 import com.spotify.futures.FuturesExtra.Consumer;
+import com.spotify.futures.Validator;
 import org.junit.Test;
 
 import java.util.Arrays;

src/test/java/com/spotify/futures/JoinedResultsTest.java → src/test/java/com/spotify/futures/jdk7/JoinedResultsTest.java

@@ -13,19 +13,19 @@
  * License for the specific language governing permissions and limitations under
  * the License.
  */
-package com.spotify.futures;
+package com.spotify.futures.jdk7;
 
-import static org.junit.Assert.assertEquals;
+import com.google.common.util.concurrent.Futures;
+import com.google.common.util.concurrent.ListenableFuture;
+import com.spotify.futures.FuturesExtra;
+import com.spotify.futures.JoinedResults;
+import org.junit.Test;
 
 import java.util.Arrays;
 import java.util.Collections;
 import java.util.List;
 
-import com.google.common.collect.Lists;
-import org.junit.Test;
-
-import com.google.common.util.concurrent.Futures;
-import com.google.common.util.concurrent.ListenableFuture;
+import static org.junit.Assert.assertEquals;
 
 public class JoinedResultsTest {
   @Test

src/test/java/com/spotify/futures/TimeoutFutureTest.java → src/test/java/com/spotify/futures/jdk7/TimeoutFutureTest.java

@@ -13,10 +13,11 @@
  * License for the specific language governing permissions and limitations under
  * the License.
  */
-package com.spotify.futures;
+package com.spotify.futures.jdk7;
 
 import com.google.common.util.concurrent.ListenableFuture;
 import com.google.common.util.concurrent.SettableFuture;
+import com.spotify.futures.FuturesExtra;
 import org.jmock.lib.concurrent.DeterministicScheduler;
 import org.junit.Before;
 import org.junit.Test;