Eliminate *BatchObserver in favour of Publisher

`reactive-streams` has become the de-facto standard for reactive
frameworks; we thus use this as a base to allow seamless interop (rather
than prompt an extra adapter layer).

Author: AlexandreCarlton

build.gradle

@@ -30,7 +30,6 @@ def getDevelopmentVersion() {
     version
 }
 
-def slf4jVersion = '1.7.30'
 def releaseVersion = System.env.RELEASE_VERSION
 version = releaseVersion ? releaseVersion : getDevelopmentVersion()
 group = 'com.graphql-java'
@@ -64,12 +63,18 @@ jar {
     }
 }
 
+def slf4jVersion = '1.7.30'
+def reactiveStreamsVersion = '1.0.3'
+
 dependencies {
     api 'org.slf4j:slf4j-api:' + slf4jVersion
+    api 'org.reactivestreams:reactive-streams:' + reactiveStreamsVersion
+
     testImplementation 'org.slf4j:slf4j-simple:' + slf4jVersion
     testImplementation 'junit:junit:4.12'
     testImplementation 'org.awaitility:awaitility:2.0.0'
     testImplementation 'com.github.ben-manes.caffeine:caffeine:2.9.0'
+    testImplementation 'io.projectreactor:reactor-core:3.6.6'
 }
 
 task sourcesJar(type: Jar) {

src/main/java/org/dataloader/BatchObserver.java

@@ -10,6 +10,8 @@
 import org.dataloader.stats.context.IncrementCacheHitCountStatisticsContext;
 import org.dataloader.stats.context.IncrementLoadCountStatisticsContext;
 import org.dataloader.stats.context.IncrementLoadErrorCountStatisticsContext;
+import org.reactivestreams.Subscriber;
+import org.reactivestreams.Subscription;
 
 import java.time.Clock;
 import java.time.Instant;
@@ -246,7 +248,7 @@ private CompletableFuture<List<V>> dispatchQueueBatch(List<K> keys, List<Object>
         return batchLoad
                 .thenApply(values -> {
                     assertResultSize(keys, values);
-                    if (isObserverLoader() || isMapObserverLoader()) {
+                    if (isPublisherLoader() || isMappedPublisherLoader()) {
                         // We have already completed the queued futures by the time the overall batchLoad future has completed.
                         return values;
                     }
@@ -428,10 +430,10 @@ CompletableFuture<List<V>> invokeLoader(List<K> keys, List<Object> keyContexts,
                     .context(context).keyContexts(keys, keyContexts).build();
             if (isMapLoader()) {
                 batchLoad = invokeMapBatchLoader(keys, environment);
-            } else if (isObserverLoader()) {
-                batchLoad = invokeObserverBatchLoader(keys, keyContexts, queuedFutures, environment);
-            } else if (isMapObserverLoader()) {
-                batchLoad = invokeMappedObserverBatchLoader(keys, keyContexts, queuedFutures, environment);
+            } else if (isPublisherLoader()) {
+                batchLoad = invokePublisherBatchLoader(keys, keyContexts, queuedFutures, environment);
+            } else if (isMappedPublisherLoader()) {
+                batchLoad = invokeMappedPublisherBatchLoader(keys, keyContexts, queuedFutures, environment);
             } else {
                 batchLoad = invokeListBatchLoader(keys, environment);
             }
@@ -503,46 +505,46 @@ private CompletableFuture<List<V>> invokeMapBatchLoader(List<K> keys, BatchLoade
         });
     }
 
-    private CompletableFuture<List<V>> invokeObserverBatchLoader(List<K> keys, List<Object> keyContexts, List<CompletableFuture<V>> queuedFutures, BatchLoaderEnvironment environment) {
+    private CompletableFuture<List<V>> invokePublisherBatchLoader(List<K> keys, List<Object> keyContexts, List<CompletableFuture<V>> queuedFutures, BatchLoaderEnvironment environment) {
         CompletableFuture<List<V>> loadResult = new CompletableFuture<>();
-        BatchObserver<V> observer = new BatchObserverImpl(loadResult, keys, keyContexts, queuedFutures);
+        Subscriber<V> subscriber = new DataLoaderSubscriber(loadResult, keys, keyContexts, queuedFutures);
 
         BatchLoaderScheduler batchLoaderScheduler = loaderOptions.getBatchLoaderScheduler();
-        if (batchLoadFunction instanceof ObserverBatchLoaderWithContext) {
-            ObserverBatchLoaderWithContext<K, V> loadFunction = (ObserverBatchLoaderWithContext<K, V>) batchLoadFunction;
+        if (batchLoadFunction instanceof PublisherBatchLoaderWithContext) {
+            PublisherBatchLoaderWithContext<K, V> loadFunction = (PublisherBatchLoaderWithContext<K, V>) batchLoadFunction;
             if (batchLoaderScheduler != null) {
-                BatchLoaderScheduler.ScheduledObserverBatchLoaderCall loadCall = () -> loadFunction.load(keys, observer, environment);
+                BatchLoaderScheduler.ScheduledObserverBatchLoaderCall loadCall = () -> loadFunction.load(keys, subscriber, environment);
                 batchLoaderScheduler.scheduleObserverBatchLoader(loadCall, keys, environment);
             } else {
-                loadFunction.load(keys, observer, environment);
+                loadFunction.load(keys, subscriber, environment);
             }
         } else {
-            ObserverBatchLoader<K, V> loadFunction = (ObserverBatchLoader<K, V>) batchLoadFunction;
+            PublisherBatchLoader<K, V> loadFunction = (PublisherBatchLoader<K, V>) batchLoadFunction;
             if (batchLoaderScheduler != null) {
-                BatchLoaderScheduler.ScheduledObserverBatchLoaderCall loadCall = () -> loadFunction.load(keys, observer);
+                BatchLoaderScheduler.ScheduledObserverBatchLoaderCall loadCall = () -> loadFunction.load(keys, subscriber);
                 batchLoaderScheduler.scheduleObserverBatchLoader(loadCall, keys, null);
             } else {
-                loadFunction.load(keys, observer);
+                loadFunction.load(keys, subscriber);
             }
         }
         return loadResult;
     }
 
-    private CompletableFuture<List<V>> invokeMappedObserverBatchLoader(List<K> keys, List<Object> keyContexts, List<CompletableFuture<V>> queuedFutures, BatchLoaderEnvironment environment) {
+    private CompletableFuture<List<V>> invokeMappedPublisherBatchLoader(List<K> keys, List<Object> keyContexts, List<CompletableFuture<V>> queuedFutures, BatchLoaderEnvironment environment) {
         CompletableFuture<List<V>> loadResult = new CompletableFuture<>();
-        MappedBatchObserver<K, V> observer = new MappedBatchObserverImpl(loadResult, keys, keyContexts, queuedFutures);
+        Subscriber<Map.Entry<K, V>> observer = new DataLoaderMapEntrySubscriber(loadResult, keys, keyContexts, queuedFutures);
 
         BatchLoaderScheduler batchLoaderScheduler = loaderOptions.getBatchLoaderScheduler();
-        if (batchLoadFunction instanceof MappedObserverBatchLoaderWithContext) {
-            MappedObserverBatchLoaderWithContext<K, V> loadFunction = (MappedObserverBatchLoaderWithContext<K, V>) batchLoadFunction;
+        if (batchLoadFunction instanceof MappedPublisherBatchLoaderWithContext) {
+            MappedPublisherBatchLoaderWithContext<K, V> loadFunction = (MappedPublisherBatchLoaderWithContext<K, V>) batchLoadFunction;
             if (batchLoaderScheduler != null) {
                 BatchLoaderScheduler.ScheduledObserverBatchLoaderCall loadCall = () -> loadFunction.load(keys, observer, environment);
                 batchLoaderScheduler.scheduleObserverBatchLoader(loadCall, keys, environment);
             } else {
                 loadFunction.load(keys, observer, environment);
             }
         } else {
-            MappedObserverBatchLoader<K, V> loadFunction = (MappedObserverBatchLoader<K, V>) batchLoadFunction;
+            MappedPublisherBatchLoader<K, V> loadFunction = (MappedPublisherBatchLoader<K, V>) batchLoadFunction;
             if (batchLoaderScheduler != null) {
                 BatchLoaderScheduler.ScheduledObserverBatchLoaderCall loadCall = () -> loadFunction.load(keys, observer);
                 batchLoaderScheduler.scheduleObserverBatchLoader(loadCall, keys, null);
@@ -557,12 +559,12 @@ private boolean isMapLoader() {
         return batchLoadFunction instanceof MappedBatchLoader || batchLoadFunction instanceof MappedBatchLoaderWithContext;
     }
 
-    private boolean isObserverLoader() {
-        return batchLoadFunction instanceof ObserverBatchLoader;
+    private boolean isPublisherLoader() {
+        return batchLoadFunction instanceof PublisherBatchLoader;
     }
 
-    private boolean isMapObserverLoader() {
-        return batchLoadFunction instanceof MappedObserverBatchLoader;
+    private boolean isMappedPublisherLoader() {
+        return batchLoadFunction instanceof MappedPublisherBatchLoader;
     }
 
     int dispatchDepth() {
@@ -616,7 +618,8 @@ private static <T> DispatchResult<T> emptyDispatchResult() {
         return (DispatchResult<T>) EMPTY_DISPATCH_RESULT;
     }
 
-    private class BatchObserverImpl implements BatchObserver<V> {
+    private class DataLoaderSubscriber implements Subscriber<V> {
+
         private final CompletableFuture<List<V>> valuesFuture;
         private final List<K> keys;
         private final List<Object> callContexts;
@@ -628,7 +631,7 @@ private class BatchObserverImpl implements BatchObserver<V> {
         private boolean onErrorCalled = false;
         private boolean onCompletedCalled = false;
 
-        private BatchObserverImpl(
+        private DataLoaderSubscriber(
             CompletableFuture<List<V>> valuesFuture,
             List<K> keys,
             List<Object> callContexts,
@@ -640,6 +643,11 @@ private BatchObserverImpl(
             this.queuedFutures = queuedFutures;
         }
 
+        @Override
+        public void onSubscribe(Subscription subscription) {
+            subscription.request(keys.size());
+        }
+
         @Override
         public void onNext(V value) {
             assert !onErrorCalled && !onCompletedCalled;
@@ -671,7 +679,7 @@ public void onNext(V value) {
         }
 
         @Override
-        public void onCompleted() {
+        public void onComplete() {
             assert !onErrorCalled;
             onCompletedCalled = true;
 
@@ -701,7 +709,7 @@ public void onError(Throwable ex) {
         }
     }
 
-    private class MappedBatchObserverImpl implements MappedBatchObserver<K, V> {
+    private class DataLoaderMapEntrySubscriber implements Subscriber<Map.Entry<K, V>> {
         private final CompletableFuture<List<V>> valuesFuture;
         private final List<K> keys;
         private final List<Object> callContexts;
@@ -714,7 +722,7 @@ private class MappedBatchObserverImpl implements MappedBatchObserver<K, V> {
         private boolean onErrorCalled = false;
         private boolean onCompletedCalled = false;
 
-        private MappedBatchObserverImpl(
+        private DataLoaderMapEntrySubscriber(
             CompletableFuture<List<V>> valuesFuture,
             List<K> keys,
             List<Object> callContexts,
@@ -737,8 +745,15 @@ private MappedBatchObserverImpl(
         }
 
         @Override
-        public void onNext(K key, V value) {
+        public void onSubscribe(Subscription subscription) {
+            subscription.request(keys.size());
+        }
+
+        @Override
+        public void onNext(Map.Entry<K, V> entry) {
             assert !onErrorCalled && !onCompletedCalled;
+            K key = entry.getKey();
+            V value = entry.getValue();
 
             Object callContext = callContextByKey.get(key);
             CompletableFuture<V> future = queuedFutureByKey.get(key);
@@ -765,7 +780,7 @@ public void onNext(K key, V value) {
         }
 
         @Override
-        public void onCompleted() {
+        public void onComplete() {
             assert !onErrorCalled;
             onCompletedCalled = true;
 

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

@@ -1,17 +1,20 @@
 package org.dataloader;
 
+import org.reactivestreams.Subscriber;
+
 import java.util.List;
+import java.util.Map;
 
 /**
  * A function that is invoked for batch loading a stream of data values indicated by the provided list of keys.
  * <p>
- * The function will call the provided {@link MappedBatchObserver} to process the key/value pairs it has retrieved to allow
+ * The function will call the provided {@link Subscriber} to process the key/value pairs it has retrieved to allow
  * the future returned by {@link DataLoader#load(Object)} to complete as soon as the individual value is available
  * (rather than when all values have been retrieved).
  *
  * @param <K> type parameter indicating the type of keys to use for data load requests.
  * @param <V> type parameter indicating the type of values returned
  */
-public interface MappedObserverBatchLoader<K, V> {
-    void load(List<K> keys, MappedBatchObserver<K, V> observer);
+public interface MappedPublisherBatchLoader<K, V> {
+    void load(List<K> keys, Subscriber<Map.Entry<K, V>> subscriber);
 }

src/main/java/org/dataloader/MappedBatchObserver.java

@@ -0,0 +1,13 @@
+package org.dataloader;
+
+import org.reactivestreams.Subscriber;
+
+import java.util.List;
+import java.util.Map;
+
+/**
+ * A {@link MappedPublisherBatchLoader} with a {@link BatchLoaderEnvironment} provided as an extra parameter to {@link #load}.
+ */
+public interface MappedPublisherBatchLoaderWithContext<K, V> {
+    void load(List<K> keys, Subscriber<Map.Entry<K, V>> subscriber, BatchLoaderEnvironment environment);
+}

src/main/java/org/dataloader/MappedObserverBatchLoaderWithContext.java

@@ -1,11 +1,13 @@
 package org.dataloader;
 
+import org.reactivestreams.Subscriber;
+
 import java.util.List;
 
 /**
  * A function that is invoked for batch loading a stream of data values indicated by the provided list of keys.
  * <p>
- * The function will call the provided {@link BatchObserver} to process the values it has retrieved to allow
+ * The function will call the provided {@link Subscriber} to process the values it has retrieved to allow
  * the future returned by {@link DataLoader#load(Object)} to complete as soon as the individual value is available
  * (rather than when all values have been retrieved).
  * <p>
@@ -14,6 +16,6 @@
  * @param <K> type parameter indicating the type of keys to use for data load requests.
  * @param <V> type parameter indicating the type of values returned
  */
-public interface ObserverBatchLoader<K, V> {
-    void load(List<K> keys, BatchObserver<V> observer);
+public interface PublisherBatchLoader<K, V> {
+    void load(List<K> keys, Subscriber<V> subscriber);
 }

src/main/java/org/dataloader/MappedObserverBatchLoader.java renamed to src/main/java/org/dataloader/MappedPublisherBatchLoader.java

@@ -0,0 +1,12 @@
+package org.dataloader;
+
+import org.reactivestreams.Subscriber;
+
+import java.util.List;
+
+/**
+ * An {@link PublisherBatchLoader} with a {@link BatchLoaderEnvironment} provided as an extra parameter to {@link #load}.
+ */
+public interface PublisherBatchLoaderWithContext<K, V> {
+    void load(List<K> keys, Subscriber<V> subscriber, BatchLoaderEnvironment environment);
+}