Extended PromisesValues with more tests and handling more edge cases

Expanded the read me text a lot

Author: bbakerman

README.md

@@ -26,15 +26,12 @@ and Nicholas Schrock (@schrockn) from [Facebook](https://www.facebook.com/), the
 ## Table of contents
 
 - [Features](#features)
+- [Examples](#examples)
 - [Differences to reference implementation](#differences-to-reference-implementation)
   - [Manual dispatching](#manual-dispatching)
 - [Let's get started!](#lets-get-started)
   - [Installing](#installing)
   - [Building](#building)
-- [Project plans](#project-plans)
-  - [Current releases](#current-releases)
-  - [Known issues](#known-issues)
-  - [Future ideas](#future-ideas)
 - [Other information sources](#other-information-sources)
 - [Contributing](#contributing)
 - [Acknowledgements](#acknowledgements)
@@ -184,16 +181,6 @@ To build from source use the Gradle wrapper:
 ./gradlew clean build
 ```
 
-## Project plans
-
-### Current releases
-
-- `1.0.0` Initial release
-
-### Known issues
-
-- Tests on job queue ordering need refactoring to Futures, one test currently omitted
-
 
 ## Other information sources
 
@@ -219,21 +206,22 @@ This library was originally written for use within a [VertX world](http://vertx.
 itself.  All the heavy lifting has been done by this project : [vertx-dataloader](https://github.com/engagingspaces/vertx-dataloader)
 including the extensive testing.
 
+This particular port was done to reduce the dependency on Vertx and to write a pure Java 8 implementation with no dependencies and also
+to use the more normative Java CompletableFuture.  [vertx-core](http://vertx.io/docs/vertx-core/java/) is not a lightweight library by any means
+so having a pure Java 8 implementation is very desirable.
+
 
 This library is entirely inspired by the great works of [Lee Byron](https://github.com/leebyron) and
 [Nicholas Schrock](https://github.com/schrockn) from [Facebook](https://www.facebook.com/) whom we would like to thank, and
 especially @leebyron for taking the time and effort to provide 100% coverage on the codebase. A set of tests which
 were also ported.
 
-This particular port was done to reduce the dependency on Vertx and to write a pure Java 8 implementation with no dependencies and also
-to use the more normative Java CompletableFuture.  [vertx-core](http://vertx.io/docs/vertx-core/java/) is not a lightweight library by any means
-so having a pure Java 8 implementation is very desirable.
 
 
 ## Licensing
 
 This project is licensed under the
 [Apache Commons v2.0](https://www.apache.org/licenses/LICENSE-2.0) license.
 
-Copyright © 2016 Arnold Schrijver and others
+Copyright © 2016 Arnold Schrijver, 2017 Brad Baker and others
 [contributors](https://github.com/bbakerman/java-dataloader/graphs/contributors)

src/main/java/org/dataloader/DataLoader.java

@@ -192,6 +192,37 @@ public PromisedValues<V> dispatch() {
         return batchOfPromisedValues;
     }
 
+    /**
+     * Normally {@link #dispatch()} is an asynchronous operation but this version will 'join' on the
+     * results if dispatch and wait for them to complete.  If the {@link CompletableFuture} callbacks make more
+     * calls to this data loader then the {@link #dispatchDepth()} will be &gt; 0 and this method will loop
+     * around and wait for any other extra batch loads to occur.
+     *
+     * @return the list of all results when the {@link #dispatchDepth()} reached 0
+     */
+    public List<V> dispatchAndJoin() {
+        List<V> results = new ArrayList<>();
+
+        List<V> joinedResults = dispatch().toCompletableFuture().join();
+        results.addAll(joinedResults);
+        while (this.dispatchDepth() > 0) {
+            joinedResults = dispatch().toCompletableFuture().join();
+            results.addAll(joinedResults);
+        }
+        return results;
+    }
+
+
+    /**
+     * @return the depth of the batched key loads that need to be dispatched
+     */
+    public int dispatchDepth() {
+        synchronized (loaderQueue) {
+            return loaderQueue.size();
+        }
+    }
+
+
     /**
      * Clears the future with the specified key from the cache, if caching is enabled, so it will be re-fetched
      * on the next load request.

src/main/java/org/dataloader/PromisedValues.java

@@ -22,6 +22,7 @@
  * also compose a {@link CompletableFuture} of that list of values via {@link #toCompletableFuture()}
  */
 public interface PromisedValues<T> {
+
     /**
      * Returns a new {@link PromisedValues} that is completed when all of
      * the given {@link CompletableFuture}s complete.  If any of the given
@@ -31,7 +32,7 @@ public interface PromisedValues<T> {
      * @param cfs the {@link CompletableFuture}s to combine
      * @param <T> the type of values
      *
-     * @return a new CombinedFutures
+     * @return a new PromisedValues
      */
     static <T> PromisedValues<T> allOf(List<CompletableFuture<T>> cfs) {
         return PromisedValuesImpl.combineAllOf(cfs);
@@ -47,7 +48,7 @@ static <T> PromisedValues<T> allOf(List<CompletableFuture<T>> cfs) {
      * @param f2  the 2nd completable future
      * @param <T> the type of values
      *
-     * @return a new CombinedFutures
+     * @return a new PromisedValues
      */
     static <T> PromisedValues<T> allOf(CompletableFuture<T> f1, CompletableFuture<T> f2) {
         return PromisedValuesImpl.combineAllOf(asList(f1, f2));
@@ -64,12 +65,13 @@ static <T> PromisedValues<T> allOf(CompletableFuture<T> f1, CompletableFuture<T>
      * @param f3  the 3rd completable future
      * @param <T> the type of values
      *
-     * @return a new CombinedFutures
+     * @return a new PromisedValues
      */
     static <T> PromisedValues allOf(CompletableFuture<T> f1, CompletableFuture<T> f2, CompletableFuture<T> f3) {
         return PromisedValuesImpl.combineAllOf(asList(f1, f2, f3));
     }
 
+
     /**
      * Returns a new {@link PromisedValues} that is completed when all of
      * the given {@link CompletableFuture}s complete.  If any of the given
@@ -82,12 +84,80 @@ static <T> PromisedValues allOf(CompletableFuture<T> f1, CompletableFuture<T> f2
      * @param f4  the 4th completable future
      * @param <T> the type of values
      *
-     * @return a new CombinedFutures
+     * @return a new PromisedValues
      */
     static <T> PromisedValues allOf(CompletableFuture<T> f1, CompletableFuture<T> f2, CompletableFuture<T> f3, CompletableFuture<T> f4) {
         return PromisedValuesImpl.combineAllOf(asList(f1, f2, f3, f4));
     }
 
+
+    /**
+     * Returns a new {@link PromisedValues} that is completed when all of
+     * the given {@link PromisedValues}s complete.  If any of the given
+     * {@link PromisedValues}s complete exceptionally, then the returned
+     * {@link PromisedValues} also does so.
+     *
+     * @param cfs the list to combine
+     * @param <T> the type of values
+     *
+     * @return a new PromisedValues
+     */
+    static <T> PromisedValues<T> allPromisedValues(List<PromisedValues<T>> cfs) {
+        return PromisedValuesImpl.combinePromisedValues(cfs);
+    }
+
+    /**
+     * Returns a new {@link PromisedValues} that is completed when all of
+     * the given {@link PromisedValues}s complete.  If any of the given
+     * {@link PromisedValues}s complete exceptionally, then the returned
+     * {@link PromisedValues} also does so.
+     *
+     * @param pv1 the 1st promised value
+     * @param pv2 the 2nd promised value
+     * @param <T> the type of values
+     *
+     * @return a new PromisedValues
+     */
+    static <T> PromisedValues<T> allPromisedValues(PromisedValues<T> pv1, PromisedValues<T> pv2) {
+        return PromisedValuesImpl.combinePromisedValues(asList(pv1, pv2));
+    }
+
+    /**
+     * Returns a new {@link PromisedValues} that is completed when all of
+     * the given {@link PromisedValues}s complete.  If any of the given
+     * {@link PromisedValues}s complete exceptionally, then the returned
+     * {@link PromisedValues} also does so.
+     *
+     * @param pv1 the 1st promised value
+     * @param pv2 the 2nd promised value
+     * @param pv3 the 3rd promised value
+     * @param <T> the type of values
+     *
+     * @return a new PromisedValues
+     */
+    static <T> PromisedValues<T> allPromisedValues(PromisedValues<T> pv1, PromisedValues<T> pv2, PromisedValues<T> pv3) {
+        return PromisedValuesImpl.combinePromisedValues(asList(pv1, pv2, pv3));
+    }
+
+    /**
+     * Returns a new {@link PromisedValues} that is completed when all of
+     * the given {@link PromisedValues}s complete.  If any of the given
+     * {@link PromisedValues}s complete exceptionally, then the returned
+     * {@link PromisedValues} also does so.
+     *
+     * @param pv1 the 1st promised value
+     * @param pv2 the 2nd promised value
+     * @param pv3 the 3rd promised value
+     * @param pv4 the 4th promised value
+     * @param <T> the type of values
+     *
+     * @return a new PromisedValues
+     */
+    static <T> PromisedValues<T> allPromisedValues(PromisedValues<T> pv1, PromisedValues<T> pv2, PromisedValues<T> pv3, PromisedValues<T> pv4) {
+        return PromisedValuesImpl.combinePromisedValues(asList(pv1, pv2, pv3, pv4));
+    }
+
+
     /**
      * When the all the futures complete, this call back will be invoked with this {@link PromisedValues} as a parameter
      *
@@ -169,11 +239,14 @@ static <T> PromisedValues allOf(CompletableFuture<T> f1, CompletableFuture<T> f2
      * (unchecked) {@link CompletionException} with the underlying
      * exception as its cause.
      *
+     * @return the list of completed values similar to {@link #toList()}
+     *
      * @throws CancellationException if the computation was cancelled
      * @throws CompletionException   if this future completed
      *                               exceptionally or a completion computation threw an exception
+     *
      */
-    void join();
+    List<T> join();
 
     /**
      * @return this as a {@link CompletableFuture} that returns the list of underlying values

src/main/java/org/dataloader/impl/PromisedValuesImpl.java

@@ -5,9 +5,11 @@
 import java.util.ArrayList;
 import java.util.List;
 import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CompletionException;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.atomic.AtomicReference;
 import java.util.function.Consumer;
+import java.util.stream.Collectors;
 
 import static org.dataloader.impl.Assertions.assertState;
 import static org.dataloader.impl.Assertions.nonNull;
@@ -22,7 +24,10 @@ private PromisedValuesImpl(List<CompletableFuture<T>> cfs) {
         this.futures = nonNull(cfs);
         this.cause = new AtomicReference<>();
         CompletableFuture[] futuresArray = cfs.toArray(new CompletableFuture[cfs.size()]);
-        this.controller = CompletableFuture.allOf(futuresArray);
+        this.controller = CompletableFuture.allOf(futuresArray).handle((result, throwable) -> {
+            setCause(throwable);
+            return null;
+        });
     }
 
     private PromisedValuesImpl(PromisedValuesImpl<T> other, CompletableFuture<Void> controller) {
@@ -35,30 +40,44 @@ public static <T> PromisedValues<T> combineAllOf(List<CompletableFuture<T>> cfs)
         return new PromisedValuesImpl<>(nonNull(cfs));
     }
 
+    public static <T> PromisedValues<T> combinePromisedValues(List<PromisedValues<T>> promisedValues) {
+        List<CompletableFuture<T>> cfs = promisedValues.stream()
+                .map(pv -> (PromisedValuesImpl<T>) pv)
+                .flatMap(pv -> pv.futures.stream())
+                .collect(Collectors.toList());
+        return new PromisedValuesImpl<>(cfs);
+    }
+
+    private void setCause(Throwable throwable) {
+        if (throwable != null) {
+            if (throwable instanceof CompletionException && throwable.getCause() != null) {
+                cause.set(throwable.getCause());
+            } else {
+                cause.set(throwable);
+            }
+        }
+    }
 
     @Override
     public PromisedValues<T> thenAccept(Consumer<PromisedValues<T>> handler) {
+        nonNull(handler);
         CompletableFuture<Void> newController = controller.handle((result, throwable) -> {
-            if (throwable != null) {
-                cause.set(throwable);
-            }
-            if (handler != null) {
-                handler.accept(this);
-            }
-            return null;
+            setCause(throwable);
+            handler.accept(this);
+            return result;
         });
         return new PromisedValuesImpl<>(this, newController);
     }
 
 
     @Override
     public boolean succeeded() {
-        return CompletableFutureKit.succeeded(controller);
+        return isDone() && cause.get() == null;
     }
 
     @Override
     public boolean failed() {
-        return CompletableFutureKit.failed(controller);
+        return isDone() && cause.get() != null;
     }
 
     @Override
@@ -110,12 +129,14 @@ public int size() {
     }
 
     @Override
-    public void join() {
+    public List<T> join() {
         controller.join();
+        return toList();
     }
 
     @Override
     public CompletableFuture<List<T>> toCompletableFuture() {
         return controller.thenApply(v -> toList());
     }
+
 }