Add dispatch strategy, implement bfs strategy

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

@@ -188,14 +188,18 @@ CompletableFuture<V> load(K key, Object loadContext) {
         }
 
         ctx.onDispatched();
+        loaderOptions.getDispatchStrategy().loadCalled();
         loadCallFuture.whenComplete(ctx::onCompleted);
+        loadCallFuture.whenComplete((result, error) -> loaderOptions.getDispatchStrategy().loadCompleted());
         return loadCallFuture;
     }
 
     private CompletableFuture<V> incrementCacheHitAndReturnCF(DataLoaderInstrumentationContext<Object> ctx, K key, Object loadContext, CompletableFuture<V> cachedFuture) {
         stats.incrementCacheHitCount(new IncrementCacheHitCountStatisticsContext<>(key, loadContext));
         ctx.onDispatched();
+        loaderOptions.getDispatchStrategy().loadCalled();
         cachedFuture.whenComplete(ctx::onCompleted);
+        cachedFuture.whenComplete((result, error) -> loaderOptions.getDispatchStrategy().loadCompleted());
         return cachedFuture;
     }
 

src/main/java/org/dataloader/DataLoaderOptions.java

@@ -55,6 +55,7 @@ public class DataLoaderOptions {
     private final ValueCacheOptions valueCacheOptions;
     private final BatchLoaderScheduler batchLoaderScheduler;
     private final DataLoaderInstrumentation instrumentation;
+    private final DispatchStrategy dispatchStrategy;
 
     /**
      * Creates a new data loader options with default settings.
@@ -72,6 +73,7 @@ public DataLoaderOptions() {
         valueCacheOptions = DEFAULT_VALUE_CACHE_OPTIONS;
         batchLoaderScheduler = null;
         instrumentation = DataLoaderInstrumentationHelper.NOOP_INSTRUMENTATION;
+        dispatchStrategy = DispatchStrategy.NO_OP;
     }
 
     private DataLoaderOptions(Builder builder) {
@@ -87,6 +89,7 @@ private DataLoaderOptions(Builder builder) {
         this.valueCacheOptions = builder.valueCacheOptions;
         this.batchLoaderScheduler = builder.batchLoaderScheduler;
         this.instrumentation = builder.instrumentation;
+        this.dispatchStrategy = builder.dispatchStrategy;
     }
 
     /**
@@ -116,6 +119,7 @@ public static DataLoaderOptions.Builder newOptions(DataLoaderOptions otherOption
      * Will transform the current options in to a builder ands allow you to build a new set of options
      *
      * @param builderConsumer the consumer of a builder that has this objects starting values
+     *
      * @return a new {@link DataLoaderOptions} object
      */
     public DataLoaderOptions transform(Consumer<Builder> builderConsumer) {
@@ -126,19 +130,22 @@ public DataLoaderOptions transform(Consumer<Builder> builderConsumer) {
 
     @Override
     public boolean equals(Object o) {
-        if (o == null || getClass() != o.getClass()) return false;
+        if (o == null || getClass() != o.getClass()) {
+            return false;
+        }
         DataLoaderOptions that = (DataLoaderOptions) o;
         return batchingEnabled == that.batchingEnabled
-                && cachingEnabled == that.cachingEnabled
-                && cachingExceptionsEnabled == that.cachingExceptionsEnabled
-                && maxBatchSize == that.maxBatchSize
-                && Objects.equals(cacheKeyFunction, that.cacheKeyFunction) &&
-                Objects.equals(cacheMap, that.cacheMap) &&
-                Objects.equals(valueCache, that.valueCache) &&
-                Objects.equals(statisticsCollector, that.statisticsCollector) &&
-                Objects.equals(environmentProvider, that.environmentProvider) &&
-                Objects.equals(valueCacheOptions, that.valueCacheOptions) &&
-                Objects.equals(batchLoaderScheduler, that.batchLoaderScheduler);
+               && cachingEnabled == that.cachingEnabled
+               && cachingExceptionsEnabled == that.cachingExceptionsEnabled
+               && maxBatchSize == that.maxBatchSize
+               && Objects.equals(cacheKeyFunction, that.cacheKeyFunction) &&
+               Objects.equals(cacheMap, that.cacheMap) &&
+               Objects.equals(valueCache, that.valueCache) &&
+               Objects.equals(statisticsCollector, that.statisticsCollector) &&
+               Objects.equals(environmentProvider, that.environmentProvider) &&
+               Objects.equals(valueCacheOptions, that.valueCacheOptions) &&
+               Objects.equals(batchLoaderScheduler, that.batchLoaderScheduler) &&
+               Objects.equals(dispatchStrategy, that.dispatchStrategy);
     }
 
 
@@ -254,7 +261,15 @@ public DataLoaderInstrumentation getInstrumentation() {
         return instrumentation;
     }
 
+    /**
+     * @return the {@link DispatchStrategy} to use for dispatching batch loads
+     */
+    public DispatchStrategy getDispatchStrategy() {
+        return dispatchStrategy;
+    }
+
     public static class Builder {
+        private DispatchStrategy dispatchStrategy = DispatchStrategy.NO_OP;
         private boolean batchingEnabled;
         private boolean cachingEnabled;
         private boolean cachingExceptionsEnabled;
@@ -285,12 +300,14 @@ public Builder() {
             this.valueCacheOptions = other.valueCacheOptions;
             this.batchLoaderScheduler = other.batchLoaderScheduler;
             this.instrumentation = other.instrumentation;
+            this.dispatchStrategy = other.dispatchStrategy;
         }
 
         /**
          * Sets the option that determines whether batch loading is enabled.
          *
          * @param batchingEnabled {@code true} to enable batch loading, {@code false} otherwise
+         *
          * @return this builder for fluent coding
          */
         public Builder setBatchingEnabled(boolean batchingEnabled) {
@@ -302,6 +319,7 @@ public Builder setBatchingEnabled(boolean batchingEnabled) {
          * Sets the option that determines whether caching is enabled.
          *
          * @param cachingEnabled {@code true} to enable caching, {@code false} otherwise
+         *
          * @return this builder for fluent coding
          */
         public Builder setCachingEnabled(boolean cachingEnabled) {
@@ -313,6 +331,7 @@ public Builder setCachingEnabled(boolean cachingEnabled) {
          * Sets the option that determines whether exceptional values are cache enabled.
          *
          * @param cachingExceptionsEnabled {@code true} to enable caching exceptional values, {@code false} otherwise
+         *
          * @return this builder for fluent coding
          */
         public Builder setCachingExceptionsEnabled(boolean cachingExceptionsEnabled) {
@@ -324,6 +343,7 @@ public Builder setCachingExceptionsEnabled(boolean cachingExceptionsEnabled) {
          * Sets the function to use for creating the cache key, if caching is enabled.
          *
          * @param cacheKeyFunction the cache key function to use
+         *
          * @return this builder for fluent coding
          */
         public Builder setCacheKeyFunction(CacheKey<?> cacheKeyFunction) {
@@ -335,6 +355,7 @@ public Builder setCacheKeyFunction(CacheKey<?> cacheKeyFunction) {
          * Sets the cache map implementation to use for caching, if caching is enabled.
          *
          * @param cacheMap the cache map instance
+         *
          * @return this builder for fluent coding
          */
         public Builder setCacheMap(CacheMap<?, ?> cacheMap) {
@@ -346,6 +367,7 @@ public Builder setCacheMap(CacheMap<?, ?> cacheMap) {
          * Sets the value cache implementation to use for caching values, if caching is enabled.
          *
          * @param valueCache the value cache instance
+         *
          * @return this builder for fluent coding
          */
         public Builder setValueCache(ValueCache<?, ?> valueCache) {
@@ -358,6 +380,7 @@ public Builder setValueCache(ValueCache<?, ?> valueCache) {
          * before they are split into multiple class
          *
          * @param maxBatchSize the maximum batch size
+         *
          * @return this builder for fluent coding
          */
         public Builder setMaxBatchSize(int maxBatchSize) {
@@ -371,6 +394,7 @@ public Builder setMaxBatchSize(int maxBatchSize) {
          * a common value
          *
          * @param statisticsCollector the statistics collector to use
+         *
          * @return this builder for fluent coding
          */
         public Builder setStatisticsCollector(Supplier<StatisticsCollector> statisticsCollector) {
@@ -382,6 +406,7 @@ public Builder setStatisticsCollector(Supplier<StatisticsCollector> statisticsCo
          * Sets the batch loader environment provider that will be used to give context to batch load functions
          *
          * @param environmentProvider the batch loader context provider
+         *
          * @return this builder for fluent coding
          */
         public Builder setBatchLoaderContextProvider(BatchLoaderContextProvider environmentProvider) {
@@ -393,6 +418,7 @@ public Builder setBatchLoaderContextProvider(BatchLoaderContextProvider environm
          * Sets the {@link ValueCacheOptions} that control how the {@link ValueCache} will be used
          *
          * @param valueCacheOptions the value cache options
+         *
          * @return this builder for fluent coding
          */
         public Builder setValueCacheOptions(ValueCacheOptions valueCacheOptions) {
@@ -405,6 +431,7 @@ public Builder setValueCacheOptions(ValueCacheOptions valueCacheOptions) {
          * to some future time.
          *
          * @param batchLoaderScheduler the scheduler
+         *
          * @return this builder for fluent coding
          */
         public Builder setBatchLoaderScheduler(BatchLoaderScheduler batchLoaderScheduler) {
@@ -416,13 +443,26 @@ public Builder setBatchLoaderScheduler(BatchLoaderScheduler batchLoaderScheduler
          * Sets in a new {@link DataLoaderInstrumentation}
          *
          * @param instrumentation the new {@link DataLoaderInstrumentation}
+         *
          * @return this builder for fluent coding
          */
         public Builder setInstrumentation(DataLoaderInstrumentation instrumentation) {
             this.instrumentation = nonNull(instrumentation);
             return this;
         }
 
+        /**
+         * Sets in a new {@link DispatchStrategy}
+         *
+         * @param dispatchStrategy the new {@link DispatchStrategy}
+         *
+         * @return the builder for fluent coding
+         */
+        public Builder setDispatchStrategy(DispatchStrategy dispatchStrategy) {
+            this.dispatchStrategy = dispatchStrategy;
+            return this;
+        }
+
         public DataLoaderOptions build() {
             return new DataLoaderOptions(this);
         }