made them non public and created a static factory support class

Author: bbakerman

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

@@ -3,9 +3,7 @@
 import org.dataloader.annotations.GuardedBy;
 import org.dataloader.annotations.Internal;
 import org.dataloader.impl.CompletableFutureKit;
-import org.dataloader.reactive.HelperIntegration;
-import org.dataloader.reactive.MappedBatchSubscriber;
-import org.dataloader.reactive.BatchSubscriber;
+import org.dataloader.reactive.ReactiveSupport;
 import org.dataloader.scheduler.BatchLoaderScheduler;
 import org.dataloader.stats.StatisticsCollector;
 import org.dataloader.stats.context.IncrementBatchLoadCountByStatisticsContext;
@@ -510,7 +508,7 @@ private CompletableFuture<List<V>> invokeMapBatchLoader(List<K> keys, BatchLoade
 
     private CompletableFuture<List<V>> invokeBatchPublisher(List<K> keys, List<Object> keyContexts, List<CompletableFuture<V>> queuedFutures, BatchLoaderEnvironment environment) {
         CompletableFuture<List<V>> loadResult = new CompletableFuture<>();
-        Subscriber<V> subscriber = new BatchSubscriber<>(loadResult, keys, keyContexts, queuedFutures, helperIntegration());
+        Subscriber<V> subscriber = ReactiveSupport.batchSubscriber(loadResult, keys, keyContexts, queuedFutures, helperIntegration());
 
         BatchLoaderScheduler batchLoaderScheduler = loaderOptions.getBatchLoaderScheduler();
         if (batchLoadFunction instanceof BatchPublisherWithContext) {
@@ -535,8 +533,8 @@ private CompletableFuture<List<V>> invokeBatchPublisher(List<K> keys, List<Objec
         return loadResult;
     }
 
-    private HelperIntegration<K> helperIntegration() {
-        return new HelperIntegration<>() {
+    private ReactiveSupport.HelperIntegration<K> helperIntegration() {
+        return new ReactiveSupport.HelperIntegration<>() {
             @Override
             public StatisticsCollector getStats() {
                 return stats;
@@ -556,7 +554,7 @@ public void clearCacheEntriesOnExceptions(List<K> keys) {
 
     private CompletableFuture<List<V>> invokeMappedBatchPublisher(List<K> keys, List<Object> keyContexts, List<CompletableFuture<V>> queuedFutures, BatchLoaderEnvironment environment) {
         CompletableFuture<List<V>> loadResult = new CompletableFuture<>();
-        Subscriber<Map.Entry<K, V>> subscriber = new MappedBatchSubscriber<>(loadResult, keys, keyContexts, queuedFutures, helperIntegration());
+        Subscriber<Map.Entry<K, V>> subscriber = ReactiveSupport.mappedBatchSubscriber(loadResult, keys, keyContexts, queuedFutures, helperIntegration());
 
         BatchLoaderScheduler batchLoaderScheduler = loaderOptions.getBatchLoaderScheduler();
         if (batchLoadFunction instanceof MappedBatchPublisherWithContext) {

src/main/java/org/dataloader/reactive/AbstractBatchSubscriber.java

@@ -24,7 +24,7 @@ abstract class AbstractBatchSubscriber<K, V, T> implements Subscriber<T> {
     final List<K> keys;
     final List<Object> callContexts;
     final List<CompletableFuture<V>> queuedFutures;
-    final HelperIntegration<K> helperIntegration;
+    final ReactiveSupport.HelperIntegration<K> helperIntegration;
 
     List<K> clearCacheKeys = new ArrayList<>();
     List<V> completedValues = new ArrayList<>();
@@ -36,7 +36,7 @@ abstract class AbstractBatchSubscriber<K, V, T> implements Subscriber<T> {
             List<K> keys,
             List<Object> callContexts,
             List<CompletableFuture<V>> queuedFutures,
-            HelperIntegration<K> helperIntegration
+            ReactiveSupport.HelperIntegration<K> helperIntegration
     ) {
         this.valuesFuture = valuesFuture;
         this.keys = keys;

src/main/java/org/dataloader/reactive/BatchSubscriber.java renamed to src/main/java/org/dataloader/reactive/BatchSubscriberImpl.java

@@ -15,16 +15,16 @@
  * @param <K> the type of keys
  * @param <V> the type of values
  */
-public class BatchSubscriber<K, V> extends AbstractBatchSubscriber<K, V, V> {
+class BatchSubscriberImpl<K, V> extends AbstractBatchSubscriber<K, V, V> {
 
     private int idx = 0;
 
-    public BatchSubscriber(
+    BatchSubscriberImpl(
             CompletableFuture<List<V>> valuesFuture,
             List<K> keys,
             List<Object> callContexts,
             List<CompletableFuture<V>> queuedFutures,
-            HelperIntegration<K> helperIntegration
+            ReactiveSupport.HelperIntegration<K> helperIntegration
     ) {
         super(valuesFuture, keys, callContexts, queuedFutures, helperIntegration);
     }

src/main/java/org/dataloader/reactive/HelperIntegration.java

@@ -15,20 +15,19 @@
  * @param <K> the type of keys
  * @param <V> the type of values
  */
-public class MappedBatchSubscriber<K, V> extends AbstractBatchSubscriber<K, V, Map.Entry<K, V>> {
+class MappedBatchSubscriberImpl<K, V> extends AbstractBatchSubscriber<K, V, Map.Entry<K, V>> {
 
     private final Map<K, Object> callContextByKey;
     private final Map<K, List<CompletableFuture<V>>> queuedFuturesByKey;
     private final Map<K, V> completedValuesByKey = new HashMap<>();
 
 
-    public MappedBatchSubscriber(
+    MappedBatchSubscriberImpl(
             CompletableFuture<List<V>> valuesFuture,
             List<K> keys,
             List<Object> callContexts,
             List<CompletableFuture<V>> queuedFutures,
-            HelperIntegration<K> helperIntegration
-
+            ReactiveSupport.HelperIntegration<K> helperIntegration
     ) {
         super(valuesFuture, keys, callContexts, queuedFutures, helperIntegration);
         this.callContextByKey = new HashMap<>();

src/main/java/org/dataloader/reactive/MappedBatchSubscriber.java renamed to src/main/java/org/dataloader/reactive/MappedBatchSubscriberImpl.java

@@ -0,0 +1,45 @@
+package org.dataloader.reactive;
+
+import org.dataloader.stats.StatisticsCollector;
+import org.reactivestreams.Subscriber;
+
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+
+public class ReactiveSupport {
+
+    public static <K, V> Subscriber<V> batchSubscriber(
+            CompletableFuture<List<V>> valuesFuture,
+            List<K> keys,
+            List<Object> callContexts,
+            List<CompletableFuture<V>> queuedFutures,
+            ReactiveSupport.HelperIntegration<K> helperIntegration
+    ) {
+        return new BatchSubscriberImpl<>(valuesFuture, keys, callContexts, queuedFutures, helperIntegration);
+    }
+
+    public static <K, V> Subscriber<Map.Entry<K, V>> mappedBatchSubscriber(
+            CompletableFuture<List<V>> valuesFuture,
+            List<K> keys,
+            List<Object> callContexts,
+            List<CompletableFuture<V>> queuedFutures,
+            ReactiveSupport.HelperIntegration<K> helperIntegration
+    ) {
+        return new MappedBatchSubscriberImpl<>(valuesFuture, keys, callContexts, queuedFutures, helperIntegration);
+    }
+
+    /**
+     * Just some callbacks to the data loader code to do common tasks
+     *
+     * @param <K> for keys
+     */
+    public interface HelperIntegration<K> {
+
+        StatisticsCollector getStats();
+
+        void clearCacheView(K key);
+
+        void clearCacheEntriesOnExceptions(List<K> keys);
+    }
+}