A PR for reactive streams support

build.gradle

@@ -30,7 +30,6 @@ def getDevelopmentVersion() {
     version
 }
 
-
 def releaseVersion = System.env.RELEASE_VERSION
 version = releaseVersion ? releaseVersion : getDevelopmentVersion()
 group = 'com.graphql-java'
@@ -76,6 +75,7 @@ dependencies {
     testImplementation 'org.awaitility:awaitility:2.0.0'
     testImplementation 'io.projectreactor:reactor-core:3.6.6'
     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/BatchPublisher.java

@@ -0,0 +1,22 @@
+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 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>
+ * <b>NOTE:</b> It is <b>required </b> that {@link Subscriber#onNext(V)} is invoked on each value in the same order as
+ * the provided keys.
+ *
+ * @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 BatchPublisher<K, V> {
+    void load(List<K> keys, Subscriber<V> subscriber);
+}

src/main/java/org/dataloader/BatchPublisherWithContext.java

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

src/main/java/org/dataloader/DataLoaderFactory.java

@@ -278,6 +278,274 @@ public static <K, V> DataLoader<K, V> newMappedDataLoaderWithTry(MappedBatchLoad
         return mkDataLoader(batchLoadFunction, options);
     }
 
+    /**
+     * Creates new DataLoader with the specified batch loader function and default options
+     * (batching, caching and unlimited batch size).
+     *
+     * @param batchLoadFunction the batch load function to use
+     * @param <K>               the key type
+     * @param <V>               the value type
+     *
+     * @return a new DataLoader
+     */
+    public static <K, V> DataLoader<K, V> newPublisherDataLoader(BatchPublisher<K, V> batchLoadFunction) {
+        return newPublisherDataLoader(batchLoadFunction, null);
+    }
+
+    /**
+     * Creates new DataLoader with the specified batch loader function with the provided options
+     *
+     * @param batchLoadFunction the batch load function to use
+     * @param options           the options to use
+     * @param <K>               the key type
+     * @param <V>               the value type
+     *
+     * @return a new DataLoader
+     */
+    public static <K, V> DataLoader<K, V> newPublisherDataLoader(BatchPublisher<K, V> batchLoadFunction, DataLoaderOptions options) {
+        return mkDataLoader(batchLoadFunction, options);
+    }
+
+    /**
+     * Creates new DataLoader with the specified batch loader function and default options
+     * (batching, caching and unlimited batch size) where the batch loader function returns a list of
+     * {@link org.dataloader.Try} objects.
+     * <p>
+     * If it's important you to know the exact status of each item in a batch call and whether it threw exceptions then
+     * you can use this form to create the data loader.
+     * <p>
+     * Using Try objects allows you to capture a value returned or an exception that might
+     * have occurred trying to get a value. .
+     *
+     * @param batchLoadFunction the batch load function to use that uses {@link org.dataloader.Try} objects
+     * @param <K>               the key type
+     * @param <V>               the value type
+     *
+     * @return a new DataLoader
+     */
+    public static <K, V> DataLoader<K, V> newPublisherDataLoaderWithTry(BatchPublisher<K, Try<V>> batchLoadFunction) {
+        return newPublisherDataLoaderWithTry(batchLoadFunction, null);
+    }
+
+    /**
+     * Creates new DataLoader with the specified batch loader function and with the provided options
+     * where the batch loader function returns a list of
+     * {@link org.dataloader.Try} objects.
+     *
+     * @param batchLoadFunction the batch load function to use that uses {@link org.dataloader.Try} objects
+     * @param options           the options to use
+     * @param <K>               the key type
+     * @param <V>               the value type
+     *
+     * @return a new DataLoader
+     *
+     * @see #newDataLoaderWithTry(BatchLoader)
+     */
+    public static <K, V> DataLoader<K, V> newPublisherDataLoaderWithTry(BatchPublisher<K, Try<V>> batchLoadFunction, DataLoaderOptions options) {
+        return mkDataLoader(batchLoadFunction, options);
+    }
+
+    /**
+     * Creates new DataLoader with the specified batch loader function and default options
+     * (batching, caching and unlimited batch size).
+     *
+     * @param batchLoadFunction the batch load function to use
+     * @param <K>               the key type
+     * @param <V>               the value type
+     *
+     * @return a new DataLoader
+     */
+    public static <K, V> DataLoader<K, V> newPublisherDataLoader(BatchPublisherWithContext<K, V> batchLoadFunction) {
+        return newPublisherDataLoader(batchLoadFunction, null);
+    }
+
+    /**
+     * Creates new DataLoader with the specified batch loader function with the provided options
+     *
+     * @param batchLoadFunction the batch load function to use
+     * @param options           the options to use
+     * @param <K>               the key type
+     * @param <V>               the value type
+     *
+     * @return a new DataLoader
+     */
+    public static <K, V> DataLoader<K, V> newPublisherDataLoader(BatchPublisherWithContext<K, V> batchLoadFunction, DataLoaderOptions options) {
+        return mkDataLoader(batchLoadFunction, options);
+    }
+
+    /**
+     * Creates new DataLoader with the specified batch loader function and default options
+     * (batching, caching and unlimited batch size) where the batch loader function returns a list of
+     * {@link org.dataloader.Try} objects.
+     * <p>
+     * If it's important you to know the exact status of each item in a batch call and whether it threw exceptions then
+     * you can use this form to create the data loader.
+     * <p>
+     * Using Try objects allows you to capture a value returned or an exception that might
+     * have occurred trying to get a value. .
+     *
+     * @param batchLoadFunction the batch load function to use that uses {@link org.dataloader.Try} objects
+     * @param <K>               the key type
+     * @param <V>               the value type
+     *
+     * @return a new DataLoader
+     */
+    public static <K, V> DataLoader<K, V> newPublisherDataLoaderWithTry(BatchPublisherWithContext<K, Try<V>> batchLoadFunction) {
+        return newPublisherDataLoaderWithTry(batchLoadFunction, null);
+    }
+
+    /**
+     * Creates new DataLoader with the specified batch loader function and with the provided options
+     * where the batch loader function returns a list of
+     * {@link org.dataloader.Try} objects.
+     *
+     * @param batchLoadFunction the batch load function to use that uses {@link org.dataloader.Try} objects
+     * @param options           the options to use
+     * @param <K>               the key type
+     * @param <V>               the value type
+     *
+     * @return a new DataLoader
+     *
+     * @see #newPublisherDataLoaderWithTry(BatchPublisher)
+     */
+    public static <K, V> DataLoader<K, V> newPublisherDataLoaderWithTry(BatchPublisherWithContext<K, Try<V>> batchLoadFunction, DataLoaderOptions options) {
+        return mkDataLoader(batchLoadFunction, options);
+    }
+
+    /**
+     * Creates new DataLoader with the specified batch loader function and default options
+     * (batching, caching and unlimited batch size).
+     *
+     * @param batchLoadFunction the batch load function to use
+     * @param <K>               the key type
+     * @param <V>               the value type
+     *
+     * @return a new DataLoader
+     */
+    public static <K, V> DataLoader<K, V> newMappedPublisherDataLoader(MappedBatchPublisher<K, V> batchLoadFunction) {
+        return newMappedPublisherDataLoader(batchLoadFunction, null);
+    }
+
+    /**
+     * Creates new DataLoader with the specified batch loader function with the provided options
+     *
+     * @param batchLoadFunction the batch load function to use
+     * @param options           the options to use
+     * @param <K>               the key type
+     * @param <V>               the value type
+     *
+     * @return a new DataLoader
+     */
+    public static <K, V> DataLoader<K, V> newMappedPublisherDataLoader(MappedBatchPublisher<K, V> batchLoadFunction, DataLoaderOptions options) {
+        return mkDataLoader(batchLoadFunction, options);
+    }
+
+    /**
+     * Creates new DataLoader with the specified batch loader function and default options
+     * (batching, caching and unlimited batch size) where the batch loader function returns a list of
+     * {@link org.dataloader.Try} objects.
+     * <p>
+     * If it's important you to know the exact status of each item in a batch call and whether it threw exceptions then
+     * you can use this form to create the data loader.
+     * <p>
+     * Using Try objects allows you to capture a value returned or an exception that might
+     * have occurred trying to get a value. .
+     *
+     * @param batchLoadFunction the batch load function to use that uses {@link org.dataloader.Try} objects
+     * @param <K>               the key type
+     * @param <V>               the value type
+     *
+     * @return a new DataLoader
+     */
+    public static <K, V> DataLoader<K, V> newMappedPublisherDataLoaderWithTry(MappedBatchPublisher<K, Try<V>> batchLoadFunction) {
+        return newMappedPublisherDataLoaderWithTry(batchLoadFunction, null);
+    }
+
+    /**
+     * Creates new DataLoader with the specified batch loader function and with the provided options
+     * where the batch loader function returns a list of
+     * {@link org.dataloader.Try} objects.
+     *
+     * @param batchLoadFunction the batch load function to use that uses {@link org.dataloader.Try} objects
+     * @param options           the options to use
+     * @param <K>               the key type
+     * @param <V>               the value type
+     *
+     * @return a new DataLoader
+     *
+     * @see #newDataLoaderWithTry(BatchLoader)
+     */
+    public static <K, V> DataLoader<K, V> newMappedPublisherDataLoaderWithTry(MappedBatchPublisher<K, Try<V>> batchLoadFunction, DataLoaderOptions options) {
+        return mkDataLoader(batchLoadFunction, options);
+    }
+
+    /**
+     * Creates new DataLoader with the specified batch loader function and default options
+     * (batching, caching and unlimited batch size).
+     *
+     * @param batchLoadFunction the batch load function to use
+     * @param <K>               the key type
+     * @param <V>               the value type
+     *
+     * @return a new DataLoader
+     */
+    public static <K, V> DataLoader<K, V> newMappedPublisherDataLoader(MappedBatchPublisherWithContext<K, V> batchLoadFunction) {
+        return newMappedPublisherDataLoader(batchLoadFunction, null);
+    }
+
+    /**
+     * Creates new DataLoader with the specified batch loader function with the provided options
+     *
+     * @param batchLoadFunction the batch load function to use
+     * @param options           the options to use
+     * @param <K>               the key type
+     * @param <V>               the value type
+     *
+     * @return a new DataLoader
+     */
+    public static <K, V> DataLoader<K, V> newMappedPublisherDataLoader(MappedBatchPublisherWithContext<K, V> batchLoadFunction, DataLoaderOptions options) {
+        return mkDataLoader(batchLoadFunction, options);
+    }
+
+    /**
+     * Creates new DataLoader with the specified batch loader function and default options
+     * (batching, caching and unlimited batch size) where the batch loader function returns a list of
+     * {@link org.dataloader.Try} objects.
+     * <p>
+     * If it's important you to know the exact status of each item in a batch call and whether it threw exceptions then
+     * you can use this form to create the data loader.
+     * <p>
+     * Using Try objects allows you to capture a value returned or an exception that might
+     * have occurred trying to get a value. .
+     *
+     * @param batchLoadFunction the batch load function to use that uses {@link org.dataloader.Try} objects
+     * @param <K>               the key type
+     * @param <V>               the value type
+     *
+     * @return a new DataLoader
+     */
+    public static <K, V> DataLoader<K, V> newMappedPublisherDataLoaderWithTry(MappedBatchPublisherWithContext<K, Try<V>> batchLoadFunction) {
+        return newMappedPublisherDataLoaderWithTry(batchLoadFunction, null);
+    }
+
+    /**
+     * Creates new DataLoader with the specified batch loader function and with the provided options
+     * where the batch loader function returns a list of
+     * {@link org.dataloader.Try} objects.
+     *
+     * @param batchLoadFunction the batch load function to use that uses {@link org.dataloader.Try} objects
+     * @param options           the options to use
+     * @param <K>               the key type
+     * @param <V>               the value type
+     *
+     * @return a new DataLoader
+     *
+     * @see #newMappedPublisherDataLoaderWithTry(MappedBatchPublisher)
+     */
+    public static <K, V> DataLoader<K, V> newMappedPublisherDataLoaderWithTry(MappedBatchPublisherWithContext<K, Try<V>> batchLoadFunction, DataLoaderOptions options) {
+        return mkDataLoader(batchLoadFunction, options);
+    }
+
     static <K, V> DataLoader<K, V> mkDataLoader(Object batchLoadFunction, DataLoaderOptions options) {
         return new DataLoader<>(batchLoadFunction, options);
     }