Merge remote-tracking branch 'origin/master' into scheduled-dataloader-registry

# Conflicts:
#	README.md

Author: bbakerman

README.md

@@ -58,6 +58,7 @@ dependencies {
     testCompile 'org.slf4j:slf4j-simple:' + slf4jVersion
     testCompile "junit:junit:4.12"
     testCompile 'org.awaitility:awaitility:2.0.0'
+    testImplementation 'com.github.ben-manes.caffeine:caffeine:2.9.0'
 }
 
 task sourcesJar(type: Jar) {

build.gradle

@@ -22,43 +22,44 @@
 import java.util.concurrent.CompletableFuture;
 
 /**
- * Cache map interface for data loaders that use caching.
+ * CacheMap is used by data loaders that use caching promises to values aka {@link CompletableFuture}<V>.  A better name for this
+ * class might have been FutureCache but that is history now.
  * <p>
- * The default implementation used by the data loader is based on a {@link java.util.LinkedHashMap}. Note that the
- * implementation could also have used a regular {@link java.util.Map} instead of this {@link CacheMap}, but
- * this aligns better to the reference data loader implementation provided by Facebook
+ * The default implementation used by the data loader is based on a {@link java.util.LinkedHashMap}.
  * <p>
- * Also it doesn't require you to implement the full set of map overloads, just the required methods.
+ * This is really a cache of completed {@link CompletableFuture}&lt;V&gt; values in memory.  It is used, when caching is enabled, to
+ * give back the same future to any code that may call it.  If you need a cache of the underlying values that is possible external to the JVM
+ * then you will want to use {{@link ValueCache}} which is designed for external cache access.
  *
- * @param <U> type parameter indicating the type of the cache keys
+ * @param <K> type parameter indicating the type of the cache keys
  * @param <V> type parameter indicating the type of the data that is cached
  *
  * @author <a href="https://github.com/aschrijver/">Arnold Schrijver</a>
  * @author <a href="https://github.com/bbakerman/">Brad Baker</a>
  */
 @PublicSpi
-public interface CacheMap<U, V> {
+public interface CacheMap<K, V> {
 
     /**
      * Creates a new cache map, using the default implementation that is based on a {@link java.util.LinkedHashMap}.
      *
-     * @param <U> type parameter indicating the type of the cache keys
+     * @param <K> type parameter indicating the type of the cache keys
      * @param <V> type parameter indicating the type of the data that is cached
      *
      * @return the cache map
      */
-    static <U, V> CacheMap<U, CompletableFuture<V>> simpleMap() {
+    static <K, V> CacheMap<K, V> simpleMap() {
         return new DefaultCacheMap<>();
     }
 
     /**
-     * Checks whether the specified key is contained in the cach map.
+     * Checks whether the specified key is contained in the cache map.
      *
      * @param key the key to check
      *
      * @return {@code true} if the cache contains the key, {@code false} otherwise
      */
-    boolean containsKey(U key);
+    boolean containsKey(K key);
 
     /**
      * Gets the specified key from the cache map.
@@ -70,7 +71,7 @@ static <U, V> CacheMap<U, CompletableFuture<V>> simpleMap() {
      *
      * @return the cached value, or {@code null} if not found (depends on cache implementation)
      */
-    V get(U key);
+    CompletableFuture<V> get(K key);
 
     /**
      * Creates a new cache map entry with the specified key and value, or updates the value if the key already exists.
@@ -80,7 +81,7 @@ static <U, V> CacheMap<U, CompletableFuture<V>> simpleMap() {
      *
      * @return the cache map for fluent coding
      */
-    CacheMap<U, V> set(U key, V value);
+    CacheMap<K, V> set(K key, CompletableFuture<V> value);
 
     /**
      * Deletes the entry with the specified key from the cache map, if it exists.
@@ -89,12 +90,12 @@ static <U, V> CacheMap<U, CompletableFuture<V>> simpleMap() {
      *
      * @return the cache map for fluent coding
      */
-    CacheMap<U, V> delete(U key);
+    CacheMap<K, V> delete(K key);
 
     /**
      * Clears all entries of the cache map
      *
      * @return the cache map for fluent coding
      */
-    CacheMap<U, V> clear();
+    CacheMap<K, V> clear();
 }

src/main/java/org/dataloader/CacheMap.java

@@ -30,6 +30,7 @@
 import java.util.List;
 import java.util.Optional;
 import java.util.concurrent.CompletableFuture;
+import java.util.function.BiConsumer;
 
 import static org.dataloader.impl.Assertions.nonNull;
 
@@ -64,8 +65,9 @@
 public class DataLoader<K, V> {
 
     private final DataLoaderHelper<K, V> helper;
-    private final CacheMap<Object, CompletableFuture<V>> futureCache;
     private final StatisticsCollector stats;
+    private final CacheMap<Object, V> futureCache;
+    private final ValueCache<Object, V> valueCache;
 
     /**
      * Creates new DataLoader with the specified batch loader function and default options
@@ -413,19 +415,24 @@ public DataLoader(BatchLoader<K, V> batchLoadFunction, DataLoaderOptions options
     @VisibleForTesting
     DataLoader(Object batchLoadFunction, DataLoaderOptions options, Clock clock) {
         DataLoaderOptions loaderOptions = options == null ? new DataLoaderOptions() : options;
-        this.futureCache = determineCacheMap(loaderOptions);
+        this.futureCache = determineFutureCache(loaderOptions);
+        this.valueCache = determineValueCache(loaderOptions);
         // order of keys matter in data loader
         this.stats = nonNull(loaderOptions.getStatisticsCollector());
 
-        this.helper = new DataLoaderHelper<>(this, batchLoadFunction, loaderOptions, this.futureCache, this.stats, clock);
+        this.helper = new DataLoaderHelper<>(this, batchLoadFunction, loaderOptions, this.futureCache, this.valueCache, this.stats, clock);
     }
 
 
     @SuppressWarnings("unchecked")
-    private CacheMap<Object, CompletableFuture<V>> determineCacheMap(DataLoaderOptions loaderOptions) {
-        return loaderOptions.cacheMap().isPresent() ? (CacheMap<Object, CompletableFuture<V>>) loaderOptions.cacheMap().get() : CacheMap.simpleMap();
+    private CacheMap<Object, V> determineFutureCache(DataLoaderOptions loaderOptions) {
+        return (CacheMap<Object, V>) loaderOptions.cacheMap().orElseGet(CacheMap::simpleMap);
     }
 
+    @SuppressWarnings("unchecked")
+    private ValueCache<Object, V> determineValueCache(DataLoaderOptions loaderOptions) {
+        return (ValueCache<Object, V>) loaderOptions.valueCache().orElseGet(ValueCache::defaultValueCache);
+    }
 
     /**
      * This returns the last instant the data loader was dispatched.  When the data loader is created this value is set to now.
@@ -628,9 +635,24 @@ public int dispatchDepth() {
      * @return the data loader for fluent coding
      */
     public DataLoader<K, V> clear(K key) {
+        return clear(key, (v, e) -> {
+        });
+    }
+
+    /**
+     * Clears the future with the specified key from the cache remote value store, if caching is enabled
+     * and a remote store is set, so it will be re-fetched and stored on the next load request.
+     *
+     * @param key     the key to remove
+     * @param handler a handler that will be called after the async remote clear completes
+     *
+     * @return the data loader for fluent coding
+     */
+    public DataLoader<K, V> clear(K key, BiConsumer<Void, Throwable> handler) {
         Object cacheKey = getCacheKey(key);
         synchronized (this) {
             futureCache.delete(cacheKey);
+            valueCache.delete(key).whenComplete(handler);
         }
         return this;
     }
@@ -641,14 +663,29 @@ public DataLoader<K, V> clear(K key) {
      * @return the data loader for fluent coding
      */
     public DataLoader<K, V> clearAll() {
+        return clearAll((v, e) -> {
+        });
+    }
+
+    /**
+     * Clears the entire cache map of the loader, and of the cached value store.
+     *
+     * @param handler a handler that will be called after the async remote clear all completes
+     *
+     * @return the data loader for fluent coding
+     */
+    public DataLoader<K, V> clearAll(BiConsumer<Void, Throwable> handler) {
         synchronized (this) {
             futureCache.clear();
+            valueCache.clear().whenComplete(handler);
         }
         return this;
     }
 
     /**
-     * Primes the cache with the given key and value.
+     * Primes the cache with the given key and value.  Note this will only prime the future cache
+     * and not the value store.  Use {@link ValueCache#set(Object, Object)} if you want
+     * o prime it with values before use
      *
      * @param key   the key
      * @param value the value

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

@@ -16,6 +16,7 @@
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.CompletionStage;
 import java.util.concurrent.atomic.AtomicReference;
+import java.util.function.BiConsumer;
 import java.util.stream.Collectors;
 
 import static java.util.Collections.emptyList;
@@ -61,17 +62,25 @@ Object getCallContext() {
     private final DataLoader<K, V> dataLoader;
     private final Object batchLoadFunction;
     private final DataLoaderOptions loaderOptions;
-    private final CacheMap<Object, CompletableFuture<V>> futureCache;
+    private final CacheMap<Object, V> futureCache;
+    private final ValueCache<Object, V> valueCache;
     private final List<LoaderQueueEntry<K, CompletableFuture<V>>> loaderQueue;
     private final StatisticsCollector stats;
     private final Clock clock;
     private final AtomicReference<Instant> lastDispatchTime;
 
-    DataLoaderHelper(DataLoader<K, V> dataLoader, Object batchLoadFunction, DataLoaderOptions loaderOptions, CacheMap<Object, CompletableFuture<V>> futureCache, StatisticsCollector stats, Clock clock) {
+    DataLoaderHelper(DataLoader<K, V> dataLoader,
+                     Object batchLoadFunction,
+                     DataLoaderOptions loaderOptions,
+                     CacheMap<Object, V> futureCache,
+                     ValueCache<Object, V> valueCache,
+                     StatisticsCollector stats,
+                     Clock clock) {
         this.dataLoader = dataLoader;
         this.batchLoadFunction = batchLoadFunction;
         this.loaderOptions = loaderOptions;
         this.futureCache = futureCache;
+        this.valueCache = valueCache;
         this.loaderQueue = new ArrayList<>();
         this.stats = stats;
         this.clock = clock;
@@ -120,35 +129,13 @@ CompletableFuture<V> load(K key, Object loadContext) {
             boolean batchingEnabled = loaderOptions.batchingEnabled();
             boolean cachingEnabled = loaderOptions.cachingEnabled();
 
-            Object cacheKey = null;
-            if (cachingEnabled) {
-                if (loadContext == null) {
-                    cacheKey = getCacheKey(key);
-                } else {
-                    cacheKey = getCacheKeyWithContext(key, loadContext);
-                }
-            }
             stats.incrementLoadCount();
 
             if (cachingEnabled) {
-                if (futureCache.containsKey(cacheKey)) {
-                    stats.incrementCacheHitCount();
-                    return futureCache.get(cacheKey);
-                }
-            }
-
-            CompletableFuture<V> future = new CompletableFuture<>();
-            if (batchingEnabled) {
-                loaderQueue.add(new LoaderQueueEntry<>(key, future, loadContext));
+                return loadFromCache(key, loadContext, batchingEnabled);
             } else {
-                stats.incrementBatchLoadCountBy(1);
-                // immediate execution of batch function
-                future = invokeLoaderImmediately(key, loadContext);
-            }
-            if (cachingEnabled) {
-                futureCache.set(cacheKey, future);
+                return queueOrInvokeLoader(key, loadContext, batchingEnabled);
             }
-            return future;
         }
     }
 
@@ -296,6 +283,66 @@ private void possiblyClearCacheEntriesOnExceptions(List<K> keys) {
         }
     }
 
+    private CompletableFuture<V> loadFromCache(K key, Object loadContext, boolean batchingEnabled) {
+        final Object cacheKey = loadContext == null ? getCacheKey(key) : getCacheKeyWithContext(key, loadContext);
+
+        if (futureCache.containsKey(cacheKey)) {
+            // We already have a promise for this key, no need to check value cache or queue up load
+            stats.incrementCacheHitCount();
+            return futureCache.get(cacheKey);
+        }
+
+        /*
+        We haven't been asked for this key yet. We want to do one of two things:
+
+        1. Check if our cache store has it. If so:
+            a. Get the value from the cache store
+            b. Add a recovery case so we queue the load if fetching from cache store fails
+            c. Put that future in our futureCache to hit the early return next time
+            d. Return the resilient future
+        2. If not in value cache:
+            a. queue or invoke the load
+            b. Add a success handler to store the result in the cache store
+            c. Return the result
+         */
+        final CompletableFuture<V> future = new CompletableFuture<>();
+
+        valueCache.get(cacheKey).whenComplete((cachedValue, getCallEx) -> {
+            if (getCallEx == null) {
+                future.complete(cachedValue);
+            } else {
+                queueOrInvokeLoader(key, loadContext, batchingEnabled)
+                        .whenComplete(setValueIntoCacheAndCompleteFuture(cacheKey, future));
+            }
+        });
+
+        futureCache.set(cacheKey, future);
+
+        return future;
+    }
+
+    private BiConsumer<V, Throwable> setValueIntoCacheAndCompleteFuture(Object cacheKey, CompletableFuture<V> future) {
+        return (result, loadCallEx) -> {
+            if (loadCallEx == null) {
+                valueCache.set(cacheKey, result)
+                        .whenComplete((v, setCallExIgnored) -> future.complete(result));
+            } else {
+                future.completeExceptionally(loadCallEx);
+            }
+        };
+    }
+
+    private CompletableFuture<V> queueOrInvokeLoader(K key, Object loadContext, boolean batchingEnabled) {
+        if (batchingEnabled) {
+            CompletableFuture<V> future = new CompletableFuture<>();
+            loaderQueue.add(new LoaderQueueEntry<>(key, future, loadContext));
+            return future;
+        } else {
+            stats.incrementBatchLoadCountBy(1);
+            // immediate execution of batch function
+            return invokeLoaderImmediately(key, loadContext);
+        }
+    }
 
     CompletableFuture<V> invokeLoaderImmediately(K key, Object keyContext) {
         List<K> keys = singletonList(key);