This attacks the problem os async ValueCache lookups in the batch load function and not the load method

Author: bbakerman

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

@@ -67,7 +67,7 @@ public class DataLoader<K, V> {
     private final DataLoaderHelper<K, V> helper;
     private final StatisticsCollector stats;
     private final CacheMap<Object, V> futureCache;
-    private final ValueCache<Object, V> valueCache;
+    private final ValueCache<K, V> valueCache;
 
     /**
      * Creates new DataLoader with the specified batch loader function and default options
@@ -430,8 +430,8 @@ private CacheMap<Object, V> determineFutureCache(DataLoaderOptions loaderOptions
     }
 
     @SuppressWarnings("unchecked")
-    private ValueCache<Object, V> determineValueCache(DataLoaderOptions loaderOptions) {
-        return (ValueCache<Object, V>) loaderOptions.valueCache().orElseGet(ValueCache::defaultValueCache);
+    private ValueCache<K, V> determineValueCache(DataLoaderOptions loaderOptions) {
+        return (ValueCache<K, V>) loaderOptions.valueCache().orElseGet(ValueCache::defaultValueCache);
     }
 
     /**

src/main/java/org/dataloader/DataLoaderHelper.java

@@ -4,6 +4,7 @@
 
 import java.util.Optional;
 import java.util.concurrent.Callable;
+import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.CompletionStage;
 import java.util.function.Consumer;
 import java.util.function.Function;
@@ -26,13 +27,22 @@
  */
 @PublicApi
 public class Try<V> {
-    private static Throwable NIL = new Throwable() {
+    private final static Object NIL = new Object() {
+    };
+
+    private final static Throwable NIL_THROWABLE = new RuntimeException() {
+        @Override
+        public String getMessage() {
+            return "failure";
+        }
+
         @Override
         public synchronized Throwable fillInStackTrace() {
             return this;
         }
     };
 
+
     private final Throwable throwable;
     private final V value;
 
@@ -48,6 +58,12 @@ private Try(V value) {
         this.throwable = null;
     }
 
+
+    @Override
+    public String toString() {
+        return isSuccess() ? "success" : "failure";
+    }
+
     /**
      * Creates a Try that has succeeded with the provided value
      *
@@ -72,6 +88,18 @@ public static <V> Try<V> failed(Throwable throwable) {
         return new Try<>(throwable);
     }
 
+    /**
+     * This returns a Try that has always failed with an consistent exception.  Use this when
+     * yiu dont care about the exception but only that the Try failed.
+     *
+     * @param <V> the type of value
+     *
+     * @return a Try that has failed
+     */
+    public static <V> Try<V> alwaysFailed() {
+        return Try.failed(NIL_THROWABLE);
+    }
+
     /**
      * Calls the callable and if it returns a value, the Try is successful with that value or if throws
      * and exception the Try captures that
@@ -96,7 +124,7 @@ public static <V> Try<V> tryCall(Callable<V> callable) {
      * @param completionStage the completion stage that will complete
      * @param <V>             the value type
      *
-     * @return a Try which is the result of the call
+     * @return a CompletionStage Try which is the result of the call
      */
     public static <V> CompletionStage<Try<V>> tryStage(CompletionStage<V> completionStage) {
         return completionStage.handle((value, throwable) -> {
@@ -107,6 +135,19 @@ public static <V> CompletionStage<Try<V>> tryStage(CompletionStage<V> completion
         });
     }
 
+    /**
+     * Creates a CompletableFuture that, when it completes, will capture into a Try whether the given completionStage
+     * was successful or not
+     *
+     * @param completionStage the completion stage that will complete
+     * @param <V>             the value type
+     *
+     * @return a CompletableFuture Try which is the result of the call
+     */
+    public static <V> CompletableFuture<Try<V>> tryFuture(CompletionStage<V> completionStage) {
+        return tryStage(completionStage).toCompletableFuture();
+    }
+
     /**
      * @return the successful value of this try
      *

src/main/java/org/dataloader/Try.java

@@ -1,8 +1,11 @@
 package org.dataloader;
 
 import org.dataloader.annotations.PublicSpi;
+import org.dataloader.impl.CompletableFutureKit;
 import org.dataloader.impl.NoOpValueCache;
 
+import java.util.ArrayList;
+import java.util.List;
 import java.util.concurrent.CompletableFuture;
 
 /**
@@ -55,8 +58,6 @@ static <K, V> ValueCache<K, V> defaultValueCache() {
      * and not null because null is a valid cacheable value.  An exceptionally completed future will cause {@link DataLoader} to load the key via batch loading
      * instead.
      * <p>
-     * NOTE: Your implementation MUST not throw exceptions, rather it should return a CompletableFuture that has completed exceptionally.  Failure
-     * to do this may cause the {@link DataLoader} code to not run properly.
      *
      * @param key the key to retrieve
      *
@@ -66,10 +67,41 @@ static <K, V> ValueCache<K, V> defaultValueCache() {
     CompletableFuture<V> get(K key);
 
     /**
-     * Stores the value with the specified key, or updates it if the key already exists.
+     * Gets the specified key from the value cache.  If the key is not present, then the returned {@link Try} will be a failed one
+     * other wise it has the cached value.  This is preferred over the {@link #get(Object)} method.
      * <p>
-     * NOTE: Your implementation MUST not throw exceptions, rather it should return a CompletableFuture that has completed exceptionally.  Failure
-     * to do this may cause the {@link DataLoader} code to not run properly.
+     *
+     * @param key the key to retrieve
+     *
+     * @return a future containing the {@link Try} cached value (which maybe null) or a failed {@link Try} if the key does
+     * not exist in the cache.
+     */
+    default CompletableFuture<Try<V>> getValue(K key) {
+        return Try.tryFuture(get(key));
+    }
+
+    /**
+     * Gets the specified keys from the value cache, in a batch call.  If your underlying cache cant do batch caching retrieval
+     * then do not implement this method and it will delegate back to {@link #getValue(Object)} for you
+     * <p>
+     * You MUST return a List that is the same size as the keys passed in.  The code will assert if you do not.
+     *
+     * @param keys the list of keys to get cached values for.
+     *
+     * @return a future containing a list of {@link Try} cached values (which maybe {@link Try#succeeded(Object)} or a failed {@link Try}
+     * per key if they do not exist in the cache.
+     */
+    default CompletableFuture<List<Try<V>>> getValues(List<K> keys) {
+        List<CompletableFuture<Try<V>>> cacheLookups = new ArrayList<>();
+        for (K key : keys) {
+            CompletableFuture<Try<V>> cacheTry = getValue(key);
+            cacheLookups.add(cacheTry);
+        }
+        return CompletableFutureKit.allOf(cacheLookups);
+    }
+
+    /**
+     * Stores the value with the specified key, or updates it if the key already exists.
      *
      * @param key   the key to store
      * @param value the value to store
@@ -78,6 +110,27 @@ static <K, V> ValueCache<K, V> defaultValueCache() {
      */
     CompletableFuture<V> set(K key, V value);
 
+    /**
+     * Stores the value with the specified keys, or updates it if the keys if they already exist.  If your underlying cache cant do batch caching setting
+     * then do not implement this method and it will delegate back to {@link #set(Object, Object)} for you
+     *
+     * @param keys   the keys to store
+     * @param values the values to store
+     *
+     * @return a future containing the stored values for fluent composition
+     */
+    default CompletableFuture<List<V>> setValues(List<K> keys, List<V> values) {
+        List<CompletableFuture<V>> cacheSets = new ArrayList<>();
+        for (int i = 0; i < keys.size(); i++) {
+            K k = keys.get(i);
+            V v = values.get(i);
+            CompletableFuture<V> setCall = set(k, v);
+            CompletableFuture<V> set = Try.tryFuture(setCall).thenApply(ignored -> v);
+            cacheSets.add(set);
+        }
+        return CompletableFutureKit.allOf(cacheSets);
+    }
+
     /**
      * Deletes the entry with the specified key from the value cache, if it exists.
      * <p>

src/main/java/org/dataloader/ValueCache.java

@@ -6,36 +6,20 @@
  * @author <a href="https://github.com/bbakerman/">Brad Baker</a>
  */
 public class ValueCacheOptions {
-    private final boolean dispatchOnCacheMiss;
     private final boolean completeValueAfterCacheSet;
 
     private ValueCacheOptions() {
-        this.dispatchOnCacheMiss = true;
         this.completeValueAfterCacheSet = false;
     }
 
-    private ValueCacheOptions(boolean dispatchOnCacheMiss, boolean completeValueAfterCacheSet) {
-        this.dispatchOnCacheMiss = dispatchOnCacheMiss;
+    private ValueCacheOptions(boolean completeValueAfterCacheSet) {
         this.completeValueAfterCacheSet = completeValueAfterCacheSet;
     }
 
     public static ValueCacheOptions newOptions() {
         return new ValueCacheOptions();
     }
 
-    /**
-     * This controls whether the {@link DataLoader} will called {@link DataLoader#dispatch()} if a
-     * {@link ValueCache#get(Object)} call misses.  In an async world this could take non zero time
-     * to complete and hence previous dispatch calls may have already completed.
-     *
-     * This is true by default.
-     *
-     * @return true if a {@link DataLoader#dispatch()} call will be made on an async {@link ValueCache} miss
-     */
-    public boolean isDispatchOnCacheMiss() {
-        return dispatchOnCacheMiss;
-    }
-
     /**
      * This controls whether the {@link DataLoader} will wait for the {@link ValueCache#set(Object, Object)} call
      * to complete before it completes the returned value.  By default this is false and hence
@@ -51,12 +35,8 @@ public boolean isCompleteValueAfterCacheSet() {
         return completeValueAfterCacheSet;
     }
 
-    public ValueCacheOptions setDispatchOnCacheMiss(boolean flag) {
-        return new ValueCacheOptions(flag, this.completeValueAfterCacheSet);
-    }
-
     public ValueCacheOptions setCompleteValueAfterCacheSet(boolean flag) {
-        return new ValueCacheOptions(this.dispatchOnCacheMiss, flag);
+        return new ValueCacheOptions(flag);
     }
 
 }

src/main/java/org/dataloader/ValueCacheOptions.java

@@ -2,28 +2,26 @@
 
 import org.dataloader.annotations.Internal;
 
-import java.util.Objects;
+import java.util.function.Supplier;
 
 @Internal
 public class Assertions {
 
-    public static void assertState(boolean state, String message) {
+    public static void assertState(boolean state, Supplier<String> message) {
         if (!state) {
-            throw new AssertionException(message);
+            throw new DataLoaderAssertionException(message.get());
         }
     }
 
     public static <T> T nonNull(T t) {
-        return Objects.requireNonNull(t, "nonNull object required");
+        return nonNull(t, () -> "nonNull object required");
     }
 
-    public static <T> T nonNull(T t, String message) {
-        return Objects.requireNonNull(t, message);
-    }
-
-    private static class AssertionException extends IllegalStateException {
-        public AssertionException(String message) {
-            super(message);
+    public static <T> T nonNull(T t, Supplier<String> message) {
+        if (t == null) {
+            throw new NullPointerException(message.get());
         }
+        return t;
     }
+
 }

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

@@ -0,0 +1,7 @@
+package org.dataloader.impl;
+
+public class DataLoaderAssertionException extends IllegalStateException {
+    public DataLoaderAssertionException(String message) {
+        super(message);
+    }
+}

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

@@ -104,7 +104,7 @@ public Throwable cause(int index) {
 
     @Override
     public T get(int index) {
-        assertState(isDone(), "The PromisedValues MUST be complete before calling the get() method");
+        assertState(isDone(), () -> "The PromisedValues MUST be complete before calling the get() method");
         try {
             CompletionStage<T> future = futures.get(index);
             return future.toCompletableFuture().get();
@@ -115,7 +115,7 @@ public T get(int index) {
 
     @Override
     public List<T> toList() {
-        assertState(isDone(), "The PromisedValues MUST be complete before calling the toList() method");
+        assertState(isDone(), () -> "The PromisedValues MUST be complete before calling the toList() method");
         int size = size();
         List<T> list = new ArrayList<>(size);
         for (int index = 0; index < size; index++) {