diff --git a/README.md b/README.md
index 24a65f6..9ffe265 100644
--- a/README.md
+++ b/README.md
@@ -286,6 +286,66 @@ For example, let's assume you want to load users from a database, you could prob
         // ...
 ```
 
+### Returning a stream of results from your batch publisher
+
+It may be that your batch loader function is a [Reactive Streams](https://www.reactive-streams.org/) [Publisher](https://www.reactive-streams.org/reactive-streams-1.0.3-javadoc/org/reactivestreams/Publisher.html), where values are emitted as an asynchronous stream.
+
+For example, let's say you wanted to load many users from a service without forcing the service to load all
+users into its memory (which may exert considerable pressure on it).
+
+A `org.dataloader.BatchPublisher` may be used to load this data:
+
+```java
+        BatchPublisher<Long, User> batchPublisher = new BatchPublisher<Long, User>() {
+            @Override
+            public void load(List<Long> userIds, Subscriber<User> userSubscriber) {
+                userManager.publishUsersById(userIds, userSubscriber);
+            }
+        };
+        DataLoader<Long, User> userLoader = DataLoaderFactory.newPublisherDataLoader(batchPublisher);
+
+        // ...
+```
+
+Rather than waiting for all values to be returned, this `DataLoader` will complete
+the `CompletableFuture<User>` returned by `Dataloader#load(Long)` as each value is
+processed.
+
+If an exception is thrown, the remaining futures yet to be completed are completed
+exceptionally.
+
+You *MUST* ensure that the values are streamed in the same order as the keys provided,
+with the same cardinality (i.e. the number of values must match the number of keys).
+Failing to do so will result in incorrect data being returned from `DataLoader#load`.
+
+
+### Returning a mapped stream of results from your batch publisher
+
+Your publisher may not necessarily return values in the same order in which it processes keys.
+
+For example, let's say your batch publisher function loads user data which is spread across shards,
+with some shards responding more quickly than others.
+
+In instances like these, `org.dataloader.MappedBatchPublisher` can be used.
+
+```java
+        MappedBatchPublisher<Long, User> mappedBatchPublisher = new MappedBatchPublisher<Long, User>() {
+            @Override
+            public void load(Set<Long> userIds, Subscriber<Map.Entry<Long, User>> userEntrySubscriber) {
+                userManager.publishUsersById(userIds, userEntrySubscriber);
+            }
+        };
+        DataLoader<Long, User> userLoader = DataLoaderFactory.newMappedPublisherDataLoader(mappedBatchPublisher);
+
+        // ...
+```
+
+Like the `BatchPublisher`, if an exception is thrown, the remaining futures yet to be completed are completed
+exceptionally.
+
+Unlike the `BatchPublisher`, however, it is not necessary to return values in the same order as the provided keys,
+or even the same number of values.
+
 ### Error object is not a thing in a type safe Java world
 
 In the reference JS implementation if the batch loader returns an `Error` object back from the `load()` promise is rejected
@@ -541,6 +601,12 @@ The following is a `BatchLoaderScheduler` that waits 10 milliseconds before invo
                     return scheduledCall.invoke();
                 }).thenCompose(Function.identity());
             }
+
+             @Override
+             public <K> void scheduleBatchPublisher(ScheduledBatchPublisherCall scheduledCall, List<K> keys, BatchLoaderEnvironment environment) {
+                 snooze(10);
+                 scheduledCall.invoke();
+             }
         };
 ```
 
diff --git a/build.gradle b/build.gradle
index 36a620a..ccbcec6 100644
--- a/build.gradle
+++ b/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,16 +63,23 @@ 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 'org.awaitility:awaitility:2.0.0'
     testImplementation "org.hamcrest:hamcrest:2.2"
+    testImplementation 'io.projectreactor:reactor-core:3.6.6'
     testImplementation 'com.github.ben-manes.caffeine:caffeine:2.9.0'
     testImplementation platform('org.junit:junit-bom:5.10.2')
     testImplementation 'org.junit.jupiter:junit-jupiter-api'
     testImplementation 'org.junit.jupiter:junit-jupiter-params'
     testRuntimeOnly 'org.junit.jupiter:junit-jupiter-engine'
+    testImplementation 'io.projectreactor:reactor-core:3.6.6'
 }
 
 task sourcesJar(type: Jar) {
diff --git a/src/main/java/org/dataloader/BatchPublisher.java b/src/main/java/org/dataloader/BatchPublisher.java
new file mode 100644
index 0000000..c499226
--- /dev/null
+++ b/src/main/java/org/dataloader/BatchPublisher.java
@@ -0,0 +1,36 @@
+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 <b>must</b> 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(Object)} is invoked on each value in the same order as
+ * the provided keys and that you provide a value for every key provided.
+ *
+ * @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
+ * @see BatchLoader for the non-reactive version
+ */
+public interface BatchPublisher<K, V> {
+    /**
+     * Called to batch the provided keys into a stream of values.  You <b>must</b> provide
+     * the same number of values as there as keys, and they <b>must</b> be in the order of the keys.
+     * <p>
+     * The idiomatic approach would be to create a reactive {@link org.reactivestreams.Publisher} that provides
+     * the values given the keys and then subscribe to it with the provided {@link Subscriber}.
+     * <p>
+     * <b>NOTE:</b> It is <b>required</b> that {@link Subscriber#onNext(Object)} is invoked on each value in the same order as
+     * the provided keys and that you provide a value for every key provided.
+     *
+     * @param keys       the collection of keys to load
+     * @param subscriber as values arrive you must call the subscriber for each value
+     */
+    void load(List<K> keys, Subscriber<V> subscriber);
+}
diff --git a/src/main/java/org/dataloader/BatchPublisherWithContext.java b/src/main/java/org/dataloader/BatchPublisherWithContext.java
new file mode 100644
index 0000000..4eadfe9
--- /dev/null
+++ b/src/main/java/org/dataloader/BatchPublisherWithContext.java
@@ -0,0 +1,34 @@
+package org.dataloader;
+
+import org.reactivestreams.Subscriber;
+
+import java.util.List;
+
+/**
+ * This form of {@link BatchPublisher} is given a {@link org.dataloader.BatchLoaderEnvironment} object
+ * that encapsulates the calling context.  A typical use case is passing in security credentials or database details
+ * for example.
+ * <p>
+ * See {@link BatchPublisher} for more details on the design invariants that you must implement in order to
+ * use this interface.
+ */
+public interface BatchPublisherWithContext<K, V> {
+    /**
+     * Called to batch the provided keys into a stream of values.  You <b>must</b> provide
+     * the same number of values as there as keys, and they <b>must</b> be in the order of the keys.
+     * <p>
+     * The idiomatic approach would be to create a reactive {@link org.reactivestreams.Publisher} that provides
+     * the values given the keys and then subscribe to it with the provided {@link Subscriber}.
+     * <p>
+     * <b>NOTE:</b> It is <b>required</b> that {@link Subscriber#onNext(Object)} is invoked on each value in the same order as
+     * the provided keys and that you provide a value for every key provided.
+     * <p>
+     * This is given an environment object to that maybe be useful during the call.  A typical use case
+     * is passing in security credentials or database details for example.
+     *
+     * @param keys        the collection of keys to load
+     * @param subscriber  as values arrive you must call the subscriber for each value
+     * @param environment an environment object that can help with the call
+     */
+    void load(List<K> keys, Subscriber<V> subscriber, BatchLoaderEnvironment environment);
+}
diff --git a/src/main/java/org/dataloader/DataLoaderFactory.java b/src/main/java/org/dataloader/DataLoaderFactory.java
index 013f473..db14f2e 100644
--- a/src/main/java/org/dataloader/DataLoaderFactory.java
+++ b/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);
     }
diff --git a/src/main/java/org/dataloader/DataLoaderHelper.java b/src/main/java/org/dataloader/DataLoaderHelper.java
index d934de2..9cd38d6 100644
--- a/src/main/java/org/dataloader/DataLoaderHelper.java
+++ b/src/main/java/org/dataloader/DataLoaderHelper.java
@@ -3,6 +3,7 @@
 import org.dataloader.annotations.GuardedBy;
 import org.dataloader.annotations.Internal;
 import org.dataloader.impl.CompletableFutureKit;
+import org.dataloader.reactive.ReactiveSupport;
 import org.dataloader.scheduler.BatchLoaderScheduler;
 import org.dataloader.stats.StatisticsCollector;
 import org.dataloader.stats.context.IncrementBatchLoadCountByStatisticsContext;
@@ -10,6 +11,7 @@
 import org.dataloader.stats.context.IncrementCacheHitCountStatisticsContext;
 import org.dataloader.stats.context.IncrementLoadCountStatisticsContext;
 import org.dataloader.stats.context.IncrementLoadErrorCountStatisticsContext;
+import org.reactivestreams.Subscriber;
 
 import java.time.Clock;
 import java.time.Instant;
@@ -152,11 +154,13 @@ CompletableFuture<V> load(K key, Object loadContext) {
         }
     }
 
+    @SuppressWarnings("unchecked")
     Object getCacheKey(K key) {
         return loaderOptions.cacheKeyFunction().isPresent() ?
                 loaderOptions.cacheKeyFunction().get().getKey(key) : key;
     }
 
+    @SuppressWarnings("unchecked")
     Object getCacheKeyWithContext(K key, Object context) {
         return loaderOptions.cacheKeyFunction().isPresent() ?
                 loaderOptions.cacheKeyFunction().get().getKeyWithContext(key, context) : key;
@@ -241,10 +245,14 @@ private CompletableFuture<List<V>> sliceIntoBatchesOfBatches(List<K> keys, List<
     @SuppressWarnings("unchecked")
     private CompletableFuture<List<V>> dispatchQueueBatch(List<K> keys, List<Object> callContexts, List<CompletableFuture<V>> queuedFutures) {
         stats.incrementBatchLoadCountBy(keys.size(), new IncrementBatchLoadCountByStatisticsContext<>(keys, callContexts));
-        CompletableFuture<List<V>> batchLoad = invokeLoader(keys, callContexts, loaderOptions.cachingEnabled());
+        CompletableFuture<List<V>> batchLoad = invokeLoader(keys, callContexts, queuedFutures, loaderOptions.cachingEnabled());
         return batchLoad
                 .thenApply(values -> {
                     assertResultSize(keys, values);
+                    if (isPublisher() || isMappedPublisher()) {
+                        // We have already completed the queued futures by the time the overall batchLoad future has completed.
+                        return values;
+                    }
 
                     List<K> clearCacheKeys = new ArrayList<>();
                     for (int idx = 0; idx < queuedFutures.size(); idx++) {
@@ -342,14 +350,15 @@ private CompletableFuture<V> queueOrInvokeLoader(K key, Object loadContext, bool
     CompletableFuture<V> invokeLoaderImmediately(K key, Object keyContext, boolean cachingEnabled) {
         List<K> keys = singletonList(key);
         List<Object> keyContexts = singletonList(keyContext);
-        return invokeLoader(keys, keyContexts, cachingEnabled)
+        List<CompletableFuture<V>> queuedFutures = singletonList(new CompletableFuture<>());
+        return invokeLoader(keys, keyContexts, queuedFutures, cachingEnabled)
                 .thenApply(list -> list.get(0))
                 .toCompletableFuture();
     }
 
-    CompletableFuture<List<V>> invokeLoader(List<K> keys, List<Object> keyContexts, boolean cachingEnabled) {
+    CompletableFuture<List<V>> invokeLoader(List<K> keys, List<Object> keyContexts, List<CompletableFuture<V>> queuedFutures, boolean cachingEnabled) {
         if (!cachingEnabled) {
-            return invokeLoader(keys, keyContexts);
+            return invokeLoader(keys, keyContexts, queuedFutures);
         }
         CompletableFuture<List<Try<V>>> cacheCallCF = getFromValueCache(keys);
         return cacheCallCF.thenCompose(cachedValues -> {
@@ -360,6 +369,7 @@ CompletableFuture<List<V>> invokeLoader(List<K> keys, List<Object> keyContexts,
             List<Integer> missedKeyIndexes = new ArrayList<>();
             List<K> missedKeys = new ArrayList<>();
             List<Object> missedKeyContexts = new ArrayList<>();
+            List<CompletableFuture<V>> missedQueuedFutures = new ArrayList<>();
 
             // if they return a ValueCachingNotSupported exception then we insert this special marker value, and it
             // means it's a total miss, we need to get all these keys via the batch loader
@@ -369,6 +379,7 @@ CompletableFuture<List<V>> invokeLoader(List<K> keys, List<Object> keyContexts,
                     missedKeyIndexes.add(i);
                     missedKeys.add(keys.get(i));
                     missedKeyContexts.add(keyContexts.get(i));
+                    missedQueuedFutures.add(queuedFutures.get(i));
                 }
             } else {
                 assertState(keys.size() == cachedValues.size(), () -> "The size of the cached values MUST be the same size as the key list");
@@ -393,7 +404,7 @@ CompletableFuture<List<V>> invokeLoader(List<K> keys, List<Object> keyContexts,
                 // we missed some keys from cache, so send them to the batch loader
                 // and then fill in their values
                 //
-                CompletableFuture<List<V>> batchLoad = invokeLoader(missedKeys, missedKeyContexts);
+                CompletableFuture<List<V>> batchLoad = invokeLoader(missedKeys, missedKeyContexts, missedQueuedFutures);
                 return batchLoad.thenCompose(missedValues -> {
                     assertResultSize(missedKeys, missedValues);
 
@@ -412,8 +423,7 @@ CompletableFuture<List<V>> invokeLoader(List<K> keys, List<Object> keyContexts,
         });
     }
 
-
-    CompletableFuture<List<V>> invokeLoader(List<K> keys, List<Object> keyContexts) {
+    CompletableFuture<List<V>> invokeLoader(List<K> keys, List<Object> keyContexts, List<CompletableFuture<V>> queuedFutures) {
         CompletableFuture<List<V>> batchLoad;
         try {
             Object context = loaderOptions.getBatchLoaderContextProvider().getContext();
@@ -421,6 +431,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 (isPublisher()) {
+                batchLoad = invokeBatchPublisher(keys, keyContexts, queuedFutures, environment);
+            } else if (isMappedPublisher()) {
+                batchLoad = invokeMappedBatchPublisher(keys, keyContexts, queuedFutures, environment);
             } else {
                 batchLoad = invokeListBatchLoader(keys, environment);
             }
@@ -492,10 +506,72 @@ 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 = ReactiveSupport.batchSubscriber(loadResult, keys, keyContexts, queuedFutures, helperIntegration());
+
+        BatchLoaderScheduler batchLoaderScheduler = loaderOptions.getBatchLoaderScheduler();
+        if (batchLoadFunction instanceof BatchPublisherWithContext) {
+            //noinspection unchecked
+            BatchPublisherWithContext<K, V> loadFunction = (BatchPublisherWithContext<K, V>) batchLoadFunction;
+            if (batchLoaderScheduler != null) {
+                BatchLoaderScheduler.ScheduledBatchPublisherCall loadCall = () -> loadFunction.load(keys, subscriber, environment);
+                batchLoaderScheduler.scheduleBatchPublisher(loadCall, keys, environment);
+            } else {
+                loadFunction.load(keys, subscriber, environment);
+            }
+        } else {
+            //noinspection unchecked
+            BatchPublisher<K, V> loadFunction = (BatchPublisher<K, V>) batchLoadFunction;
+            if (batchLoaderScheduler != null) {
+                BatchLoaderScheduler.ScheduledBatchPublisherCall loadCall = () -> loadFunction.load(keys, subscriber);
+                batchLoaderScheduler.scheduleBatchPublisher(loadCall, keys, null);
+            } else {
+                loadFunction.load(keys, subscriber);
+            }
+        }
+        return loadResult;
+    }
+
+    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 = ReactiveSupport.mappedBatchSubscriber(loadResult, keys, keyContexts, queuedFutures, helperIntegration());
+        Set<K> setOfKeys = new LinkedHashSet<>(keys);
+        BatchLoaderScheduler batchLoaderScheduler = loaderOptions.getBatchLoaderScheduler();
+        if (batchLoadFunction instanceof MappedBatchPublisherWithContext) {
+            //noinspection unchecked
+            MappedBatchPublisherWithContext<K, V> loadFunction = (MappedBatchPublisherWithContext<K, V>) batchLoadFunction;
+            if (batchLoaderScheduler != null) {
+                BatchLoaderScheduler.ScheduledBatchPublisherCall loadCall = () -> loadFunction.load(keys, subscriber, environment);
+                batchLoaderScheduler.scheduleBatchPublisher(loadCall, keys, environment);
+            } else {
+                loadFunction.load(keys, subscriber, environment);
+            }
+        } else {
+            //noinspection unchecked
+            MappedBatchPublisher<K, V> loadFunction = (MappedBatchPublisher<K, V>) batchLoadFunction;
+            if (batchLoaderScheduler != null) {
+                BatchLoaderScheduler.ScheduledBatchPublisherCall loadCall = () -> loadFunction.load(setOfKeys, subscriber);
+                batchLoaderScheduler.scheduleBatchPublisher(loadCall, keys, null);
+            } else {
+                loadFunction.load(setOfKeys, subscriber);
+            }
+        }
+        return loadResult;
+    }
+
     private boolean isMapLoader() {
         return batchLoadFunction instanceof MappedBatchLoader || batchLoadFunction instanceof MappedBatchLoaderWithContext;
     }
 
+    private boolean isPublisher() {
+        return batchLoadFunction instanceof BatchPublisher;
+    }
+
+    private boolean isMappedPublisher() {
+        return batchLoadFunction instanceof MappedBatchPublisher;
+    }
+
     int dispatchDepth() {
         synchronized (dataLoader) {
             return loaderQueue.size();
@@ -546,4 +622,23 @@ private CompletableFuture<List<V>> setToValueCache(List<V> assembledValues, List
     private static <T> DispatchResult<T> emptyDispatchResult() {
         return (DispatchResult<T>) EMPTY_DISPATCH_RESULT;
     }
+
+    private ReactiveSupport.HelperIntegration<K> helperIntegration() {
+        return new ReactiveSupport.HelperIntegration<>() {
+            @Override
+            public StatisticsCollector getStats() {
+                return stats;
+            }
+
+            @Override
+            public void clearCacheView(K key) {
+                dataLoader.clear(key);
+            }
+
+            @Override
+            public void clearCacheEntriesOnExceptions(List<K> keys) {
+                possiblyClearCacheEntriesOnExceptions(keys);
+            }
+        };
+    }
 }
diff --git a/src/main/java/org/dataloader/MappedBatchPublisher.java b/src/main/java/org/dataloader/MappedBatchPublisher.java
new file mode 100644
index 0000000..754ee52
--- /dev/null
+++ b/src/main/java/org/dataloader/MappedBatchPublisher.java
@@ -0,0 +1,30 @@
+package org.dataloader;
+
+import org.reactivestreams.Subscriber;
+
+import java.util.Map;
+import java.util.Set;
+
+/**
+ * A function that is invoked for batch loading a stream of data values indicated by the provided list of keys.
+ * <p>
+ * The function <b>must</b> 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
+ * @see MappedBatchLoader for the non-reactive version
+ */
+public interface MappedBatchPublisher<K, V> {
+    /**
+     * Called to batch the provided keys into a stream of map entries of keys and values.
+     * <p>
+     * The idiomatic approach would be to create a reactive {@link org.reactivestreams.Publisher} that provides
+     * the values given the keys and then subscribe to it with the provided {@link Subscriber}.
+     *
+     * @param keys       the collection of keys to load
+     * @param subscriber as values arrive you must call the subscriber for each value
+     */
+    void load(Set<K> keys, Subscriber<Map.Entry<K, V>> subscriber);
+}
diff --git a/src/main/java/org/dataloader/MappedBatchPublisherWithContext.java b/src/main/java/org/dataloader/MappedBatchPublisherWithContext.java
new file mode 100644
index 0000000..2e94152
--- /dev/null
+++ b/src/main/java/org/dataloader/MappedBatchPublisherWithContext.java
@@ -0,0 +1,32 @@
+package org.dataloader;
+
+import org.reactivestreams.Subscriber;
+
+import java.util.List;
+import java.util.Map;
+
+/**
+ * This form of {@link MappedBatchPublisher} is given a {@link org.dataloader.BatchLoaderEnvironment} object
+ * that encapsulates the calling context.  A typical use case is passing in security credentials or database details
+ * for example.
+ * <p>
+ * See {@link MappedBatchPublisher} for more details on the design invariants that you must implement in order to
+ * use this interface.
+ */
+public interface MappedBatchPublisherWithContext<K, V> {
+
+    /**
+     * Called to batch the provided keys into a stream of map entries of keys and values.
+     * <p>
+     * The idiomatic approach would be to create a reactive {@link org.reactivestreams.Publisher} that provides
+     * the values given the keys and then subscribe to it with the provided {@link Subscriber}.
+     * <p>
+     * This is given an environment object to that maybe be useful during the call.  A typical use case
+     * is passing in security credentials or database details for example.
+     *
+     * @param keys        the collection of keys to load
+     * @param subscriber  as values arrive you must call the subscriber for each value
+     * @param environment an environment object that can help with the call
+     */
+    void load(List<K> keys, Subscriber<Map.Entry<K, V>> subscriber, BatchLoaderEnvironment environment);
+}
diff --git a/src/main/java/org/dataloader/reactive/AbstractBatchSubscriber.java b/src/main/java/org/dataloader/reactive/AbstractBatchSubscriber.java
new file mode 100644
index 0000000..c2f5438
--- /dev/null
+++ b/src/main/java/org/dataloader/reactive/AbstractBatchSubscriber.java
@@ -0,0 +1,104 @@
+package org.dataloader.reactive;
+
+import org.dataloader.Try;
+import org.dataloader.stats.context.IncrementBatchLoadExceptionCountStatisticsContext;
+import org.dataloader.stats.context.IncrementLoadErrorCountStatisticsContext;
+import org.reactivestreams.Subscriber;
+import org.reactivestreams.Subscription;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CompletionException;
+
+import static org.dataloader.impl.Assertions.assertState;
+
+/**
+ * The base class for our reactive subscriber support
+ *
+ * @param <T> for two
+ */
+abstract class AbstractBatchSubscriber<K, V, T> implements Subscriber<T> {
+
+    final CompletableFuture<List<V>> valuesFuture;
+    final List<K> keys;
+    final List<Object> callContexts;
+    final List<CompletableFuture<V>> queuedFutures;
+    final ReactiveSupport.HelperIntegration<K> helperIntegration;
+
+    List<K> clearCacheKeys = new ArrayList<>();
+    List<V> completedValues = new ArrayList<>();
+    boolean onErrorCalled = false;
+    boolean onCompleteCalled = false;
+
+    AbstractBatchSubscriber(
+            CompletableFuture<List<V>> valuesFuture,
+            List<K> keys,
+            List<Object> callContexts,
+            List<CompletableFuture<V>> queuedFutures,
+            ReactiveSupport.HelperIntegration<K> helperIntegration
+    ) {
+        this.valuesFuture = valuesFuture;
+        this.keys = keys;
+        this.callContexts = callContexts;
+        this.queuedFutures = queuedFutures;
+        this.helperIntegration = helperIntegration;
+    }
+
+    @Override
+    public void onSubscribe(Subscription subscription) {
+        subscription.request(keys.size());
+    }
+
+    @Override
+    public void onNext(T v) {
+        assertState(!onErrorCalled, () -> "onError has already been called; onNext may not be invoked.");
+        assertState(!onCompleteCalled, () -> "onComplete has already been called; onNext may not be invoked.");
+    }
+
+    @Override
+    public void onComplete() {
+        assertState(!onErrorCalled, () -> "onError has already been called; onComplete may not be invoked.");
+        onCompleteCalled = true;
+    }
+
+    @Override
+    public void onError(Throwable throwable) {
+        assertState(!onCompleteCalled, () -> "onComplete has already been called; onError may not be invoked.");
+        onErrorCalled = true;
+
+        helperIntegration.getStats().incrementBatchLoadExceptionCount(new IncrementBatchLoadExceptionCountStatisticsContext<>(keys, callContexts));
+    }
+
+    /*
+     * A value has arrived - how do we complete the future that's associated with it in a common way
+     */
+    void onNextValue(K key, V value, Object callContext, List<CompletableFuture<V>> futures) {
+        if (value instanceof Try) {
+            // we allow the batch loader to return a Try so we can better represent a computation
+            // that might have worked or not.
+            //noinspection unchecked
+            Try<V> tryValue = (Try<V>) value;
+            if (tryValue.isSuccess()) {
+                futures.forEach(f -> f.complete(tryValue.get()));
+            } else {
+                helperIntegration.getStats().incrementLoadErrorCount(new IncrementLoadErrorCountStatisticsContext<>(key, callContext));
+                futures.forEach(f -> f.completeExceptionally(tryValue.getThrowable()));
+                clearCacheKeys.add(key);
+            }
+        } else {
+            futures.forEach(f -> f.complete(value));
+        }
+    }
+
+    Throwable unwrapThrowable(Throwable ex) {
+        if (ex instanceof CompletionException) {
+            ex = ex.getCause();
+        }
+        return ex;
+    }
+
+    void possiblyClearCacheEntriesOnExceptions() {
+        helperIntegration.clearCacheEntriesOnExceptions(clearCacheKeys);
+    }
+}
diff --git a/src/main/java/org/dataloader/reactive/BatchSubscriberImpl.java b/src/main/java/org/dataloader/reactive/BatchSubscriberImpl.java
new file mode 100644
index 0000000..d0b8110
--- /dev/null
+++ b/src/main/java/org/dataloader/reactive/BatchSubscriberImpl.java
@@ -0,0 +1,86 @@
+package org.dataloader.reactive;
+
+import org.dataloader.impl.DataLoaderAssertionException;
+
+import java.util.List;
+import java.util.concurrent.CompletableFuture;
+
+/**
+ * This class can be used to subscribe to a {@link org.reactivestreams.Publisher} and then
+ * have the values it receives complete the data loader keys.  The keys and values must be
+ * in index order.
+ * <p>
+ * This is a reactive version of {@link org.dataloader.BatchLoader}
+ *
+ * @param <K> the type of keys
+ * @param <V> the type of values
+ */
+class BatchSubscriberImpl<K, V> extends AbstractBatchSubscriber<K, V, V> {
+
+    private int idx = 0;
+
+    BatchSubscriberImpl(
+            CompletableFuture<List<V>> valuesFuture,
+            List<K> keys,
+            List<Object> callContexts,
+            List<CompletableFuture<V>> queuedFutures,
+            ReactiveSupport.HelperIntegration<K> helperIntegration
+    ) {
+        super(valuesFuture, keys, callContexts, queuedFutures, helperIntegration);
+    }
+
+    // onNext may be called by multiple threads - for the time being, we pass 'synchronized' to guarantee
+    // correctness (at the cost of speed).
+    @Override
+    public synchronized void onNext(V value) {
+        super.onNext(value);
+
+        if (idx >= keys.size()) {
+            // hang on they have given us more values than we asked for in keys
+            // we cant handle this
+            return;
+        }
+        K key = keys.get(idx);
+        Object callContext = callContexts.get(idx);
+        CompletableFuture<V> future = queuedFutures.get(idx);
+        onNextValue(key, value, callContext, List.of(future));
+
+        completedValues.add(value);
+        idx++;
+    }
+
+
+    @Override
+    public synchronized void onComplete() {
+        super.onComplete();
+        if (keys.size() != completedValues.size()) {
+            // we have more or less values than promised
+            // we will go through all the outstanding promises and mark those that
+            // have not finished as failed
+            for (CompletableFuture<V> queuedFuture : queuedFutures) {
+                if (!queuedFuture.isDone()) {
+                    queuedFuture.completeExceptionally(new DataLoaderAssertionException("The size of the promised values MUST be the same size as the key list"));
+                }
+            }
+        }
+        possiblyClearCacheEntriesOnExceptions();
+        valuesFuture.complete(completedValues);
+    }
+
+    @Override
+    public synchronized void onError(Throwable ex) {
+        super.onError(ex);
+        ex = unwrapThrowable(ex);
+        // Set the remaining keys to the exception.
+        for (int i = idx; i < queuedFutures.size(); i++) {
+            K key = keys.get(i);
+            CompletableFuture<V> future = queuedFutures.get(i);
+            if (!future.isDone()) {
+                future.completeExceptionally(ex);
+                // clear any cached view of this key because it failed
+                helperIntegration.clearCacheView(key);
+            }
+        }
+        valuesFuture.completeExceptionally(ex);
+    }
+}
diff --git a/src/main/java/org/dataloader/reactive/MappedBatchSubscriberImpl.java b/src/main/java/org/dataloader/reactive/MappedBatchSubscriberImpl.java
new file mode 100644
index 0000000..d56efa0
--- /dev/null
+++ b/src/main/java/org/dataloader/reactive/MappedBatchSubscriberImpl.java
@@ -0,0 +1,103 @@
+package org.dataloader.reactive;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+
+/**
+ * This class can be used to subscribe to a {@link org.reactivestreams.Publisher} and then
+ * have the values it receives complete the data loader keys in a map lookup fashion.
+ * <p>
+ * This is a reactive version of {@link org.dataloader.MappedBatchLoader}
+ *
+ * @param <K> the type of keys
+ * @param <V> the type of values
+ */
+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<>();
+
+
+    MappedBatchSubscriberImpl(
+            CompletableFuture<List<V>> valuesFuture,
+            List<K> keys,
+            List<Object> callContexts,
+            List<CompletableFuture<V>> queuedFutures,
+            ReactiveSupport.HelperIntegration<K> helperIntegration
+    ) {
+        super(valuesFuture, keys, callContexts, queuedFutures, helperIntegration);
+        this.callContextByKey = new HashMap<>();
+        this.queuedFuturesByKey = new HashMap<>();
+        for (int idx = 0; idx < queuedFutures.size(); idx++) {
+            K key = keys.get(idx);
+            Object callContext = callContexts.get(idx);
+            CompletableFuture<V> queuedFuture = queuedFutures.get(idx);
+            callContextByKey.put(key, callContext);
+            queuedFuturesByKey.computeIfAbsent(key, k -> new ArrayList<>()).add(queuedFuture);
+        }
+    }
+
+
+    @Override
+    public synchronized void onNext(Map.Entry<K, V> entry) {
+        super.onNext(entry);
+        K key = entry.getKey();
+        V value = entry.getValue();
+
+        Object callContext = callContextByKey.get(key);
+        List<CompletableFuture<V>> futures = queuedFuturesByKey.getOrDefault(key, List.of());
+
+        onNextValue(key, value, callContext, futures);
+
+        // did we have an actual key for this value - ignore it if they send us one outside the key set
+        if (!futures.isEmpty()) {
+            completedValuesByKey.put(key, value);
+        }
+    }
+
+    @Override
+    public synchronized void onComplete() {
+        super.onComplete();
+
+        possiblyClearCacheEntriesOnExceptions();
+        List<V> values = new ArrayList<>(keys.size());
+        for (K key : keys) {
+            V value = completedValuesByKey.get(key);
+            values.add(value);
+
+            List<CompletableFuture<V>> futures = queuedFuturesByKey.getOrDefault(key, List.of());
+            for (CompletableFuture<V> future : futures) {
+                if (!future.isDone()) {
+                    // we have a future that never came back for that key
+                    // but the publisher is done sending in data - it must be null
+                    // e.g. for key X when found no value
+                    future.complete(null);
+                }
+            }
+        }
+        valuesFuture.complete(values);
+    }
+
+    @Override
+    public synchronized void onError(Throwable ex) {
+        super.onError(ex);
+        ex = unwrapThrowable(ex);
+        // Complete the futures for the remaining keys with the exception.
+        for (int idx = 0; idx < queuedFutures.size(); idx++) {
+            K key = keys.get(idx);
+            List<CompletableFuture<V>> futures = queuedFuturesByKey.get(key);
+            if (!completedValuesByKey.containsKey(key)) {
+                for (CompletableFuture<V> future : futures) {
+                    future.completeExceptionally(ex);
+                }
+                // clear any cached view of this key because they all failed
+                helperIntegration.clearCacheView(key);
+            }
+        }
+        valuesFuture.completeExceptionally(ex);
+    }
+}
diff --git a/src/main/java/org/dataloader/reactive/ReactiveSupport.java b/src/main/java/org/dataloader/reactive/ReactiveSupport.java
new file mode 100644
index 0000000..fc03bb0
--- /dev/null
+++ b/src/main/java/org/dataloader/reactive/ReactiveSupport.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);
+    }
+}
diff --git a/src/main/java/org/dataloader/scheduler/BatchLoaderScheduler.java b/src/main/java/org/dataloader/scheduler/BatchLoaderScheduler.java
index 7cddd54..e7e95d9 100644
--- a/src/main/java/org/dataloader/scheduler/BatchLoaderScheduler.java
+++ b/src/main/java/org/dataloader/scheduler/BatchLoaderScheduler.java
@@ -5,6 +5,8 @@
 import org.dataloader.DataLoader;
 import org.dataloader.DataLoaderOptions;
 import org.dataloader.MappedBatchLoader;
+import org.dataloader.MappedBatchPublisher;
+import org.dataloader.BatchPublisher;
 
 import java.util.List;
 import java.util.Map;
@@ -42,6 +44,13 @@ interface ScheduledMappedBatchLoaderCall<K, V> {
         CompletionStage<Map<K, V>> invoke();
     }
 
+    /**
+     * This represents a callback that will invoke a {@link BatchPublisher} or {@link MappedBatchPublisher} function under the covers
+     */
+    interface ScheduledBatchPublisherCall {
+        void invoke();
+    }
+
     /**
      * This is called to schedule a {@link BatchLoader} call.
      *
@@ -71,4 +80,16 @@ interface ScheduledMappedBatchLoaderCall<K, V> {
      * @return a promise to the values that come from the {@link BatchLoader}
      */
     <K, V> CompletionStage<Map<K, V>> scheduleMappedBatchLoader(ScheduledMappedBatchLoaderCall<K, V> scheduledCall, List<K> keys, BatchLoaderEnvironment environment);
+
+    /**
+     * This is called to schedule a {@link BatchPublisher} call.
+     *
+     * @param scheduledCall the callback that needs to be invoked to allow the {@link BatchPublisher} to proceed.
+     * @param keys          this is the list of keys that will be passed to the {@link BatchPublisher}.
+     *                      This is provided only for informative reasons and, you can't change the keys that are used
+     * @param environment   this is the {@link BatchLoaderEnvironment} in place,
+     *                      which can be null if it's a simple {@link BatchPublisher} call
+     * @param <K>           the key type
+     */
+    <K> void scheduleBatchPublisher(ScheduledBatchPublisherCall scheduledCall, List<K> keys, BatchLoaderEnvironment environment);
 }
diff --git a/src/test/java/ReadmeExamples.java b/src/test/java/ReadmeExamples.java
index d25dfa7..a20c0ea 100644
--- a/src/test/java/ReadmeExamples.java
+++ b/src/test/java/ReadmeExamples.java
@@ -1,11 +1,13 @@
 import org.dataloader.BatchLoader;
 import org.dataloader.BatchLoaderEnvironment;
 import org.dataloader.BatchLoaderWithContext;
+import org.dataloader.BatchPublisher;
 import org.dataloader.CacheMap;
 import org.dataloader.DataLoader;
 import org.dataloader.DataLoaderFactory;
 import org.dataloader.DataLoaderOptions;
 import org.dataloader.MappedBatchLoaderWithContext;
+import org.dataloader.MappedBatchPublisher;
 import org.dataloader.Try;
 import org.dataloader.fixtures.SecurityCtx;
 import org.dataloader.fixtures.User;
@@ -15,6 +17,7 @@
 import org.dataloader.scheduler.BatchLoaderScheduler;
 import org.dataloader.stats.Statistics;
 import org.dataloader.stats.ThreadLocalStatisticsCollector;
+import org.reactivestreams.Subscriber;
 
 import java.time.Duration;
 import java.util.ArrayList;
@@ -171,7 +174,7 @@ private void tryExample() {
         }
     }
 
-    private void tryBatcLoader() {
+    private void tryBatchLoader() {
         DataLoader<String, User> dataLoader = DataLoaderFactory.newDataLoaderWithTry(new BatchLoader<String, Try<User>>() {
             @Override
             public CompletionStage<List<Try<User>>> load(List<String> keys) {
@@ -187,6 +190,26 @@ public CompletionStage<List<Try<User>>> load(List<String> keys) {
         });
     }
 
+    private void batchPublisher() {
+        BatchPublisher<Long, User> batchPublisher = new BatchPublisher<Long, User>() {
+            @Override
+            public void load(List<Long> userIds, Subscriber<User> userSubscriber) {
+                userManager.publishUsersById(userIds, userSubscriber);
+            }
+        };
+        DataLoader<Long, User> userLoader = DataLoaderFactory.newPublisherDataLoader(batchPublisher);
+    }
+
+    private void mappedBatchPublisher() {
+        MappedBatchPublisher<Long, User> mappedBatchPublisher = new MappedBatchPublisher<Long, User>() {
+            @Override
+            public void load(Set<Long> userIds, Subscriber<Map.Entry<Long, User>> userEntrySubscriber) {
+                userManager.publishUsersById(userIds, userEntrySubscriber);
+            }
+        };
+        DataLoader<Long, User> userLoader = DataLoaderFactory.newMappedPublisherDataLoader(mappedBatchPublisher);
+    }
+
     DataLoader<String, User> userDataLoader;
 
     private void clearCacheOnError() {
@@ -304,6 +327,12 @@ public <K, V> CompletionStage<Map<K, V>> scheduleMappedBatchLoader(ScheduledMapp
                     return scheduledCall.invoke();
                 }).thenCompose(Function.identity());
             }
+
+            @Override
+            public <K> void scheduleBatchPublisher(ScheduledBatchPublisherCall scheduledCall, List<K> keys, BatchLoaderEnvironment environment) {
+                snooze(10);
+                scheduledCall.invoke();
+            }
         };
     }
 
diff --git a/src/test/java/org/dataloader/DataLoaderTest.java b/src/test/java/org/dataloader/DataLoaderTest.java
index db71c1e..1ce34ea 100644
--- a/src/test/java/org/dataloader/DataLoaderTest.java
+++ b/src/test/java/org/dataloader/DataLoaderTest.java
@@ -16,12 +16,19 @@
 
 package org.dataloader;
 
+import org.awaitility.Duration;
 import org.dataloader.fixtures.CustomCacheMap;
 import org.dataloader.fixtures.JsonObject;
-import org.dataloader.fixtures.TestKit;
 import org.dataloader.fixtures.User;
 import org.dataloader.fixtures.UserManager;
+import org.dataloader.fixtures.parameterized.ListDataLoaderFactory;
+import org.dataloader.fixtures.parameterized.MappedDataLoaderFactory;
+import org.dataloader.fixtures.parameterized.MappedPublisherDataLoaderFactory;
+import org.dataloader.fixtures.parameterized.PublisherDataLoaderFactory;
+import org.dataloader.fixtures.parameterized.TestDataLoaderFactory;
+import org.dataloader.fixtures.parameterized.TestReactiveDataLoaderFactory;
 import org.dataloader.impl.CompletableFutureKit;
+import org.dataloader.impl.DataLoaderAssertionException;
 import org.junit.jupiter.api.Named;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.params.ParameterizedTest;
@@ -46,11 +53,16 @@
 import static java.util.Arrays.asList;
 import static java.util.Collections.emptyList;
 import static java.util.Collections.singletonList;
+import static java.util.concurrent.CompletableFuture.*;
 import static org.awaitility.Awaitility.await;
 import static org.dataloader.DataLoaderFactory.newDataLoader;
 import static org.dataloader.DataLoaderFactory.newMappedDataLoader;
+import static org.dataloader.DataLoaderFactory.newMappedPublisherDataLoader;
+import static org.dataloader.DataLoaderFactory.newMappedPublisherDataLoaderWithTry;
+import static org.dataloader.DataLoaderFactory.newPublisherDataLoader;
+import static org.dataloader.DataLoaderFactory.newPublisherDataLoaderWithTry;
 import static org.dataloader.DataLoaderOptions.newOptions;
-import static org.dataloader.fixtures.TestKit.futureError;
+import static org.dataloader.fixtures.TestKit.areAllDone;
 import static org.dataloader.fixtures.TestKit.listFrom;
 import static org.dataloader.impl.CompletableFutureKit.cause;
 import static org.hamcrest.MatcherAssert.assertThat;
@@ -59,6 +71,7 @@
 import static org.hamcrest.Matchers.instanceOf;
 import static org.hamcrest.Matchers.is;
 import static org.junit.jupiter.api.Assertions.assertArrayEquals;
+import static org.junit.jupiter.api.Assertions.fail;
 
 /**
  * Tests for {@link DataLoader}.
@@ -99,7 +112,7 @@ public void basic_map_batch_loading() {
                     mapOfResults.put(k, k);
                 }
             });
-            return CompletableFuture.completedFuture(mapOfResults);
+            return completedFuture(mapOfResults);
         };
         DataLoader<String, String> loader = DataLoaderFactory.newMappedDataLoader(evensOnlyMappedBatchLoader);
 
@@ -419,7 +432,7 @@ public void should_Allow_priming_the_cache_with_a_future(TestDataLoaderFactory f
         List<Collection<String>> loadCalls = new ArrayList<>();
         DataLoader<String, String> identityLoader = factory.idLoader(new DataLoaderOptions(), loadCalls);
 
-        DataLoader<String, String> dlFluency = identityLoader.prime("A", CompletableFuture.completedFuture("A"));
+        DataLoader<String, String> dlFluency = identityLoader.prime("A", completedFuture("A"));
         assertThat(dlFluency, equalTo(identityLoader));
 
         CompletableFuture<String> future1 = identityLoader.load("A");
@@ -727,10 +740,10 @@ public void should_work_with_duplicate_keys_when_caching_disabled(TestDataLoader
         assertThat(future1.get(), equalTo("A"));
         assertThat(future2.get(), equalTo("B"));
         assertThat(future3.get(), equalTo("A"));
-        if (factory instanceof ListDataLoaderFactory) {
-            assertThat(loadCalls, equalTo(singletonList(asList("A", "B", "A"))));
-        } else {
+        if (factory instanceof MappedDataLoaderFactory || factory instanceof MappedPublisherDataLoaderFactory) {
             assertThat(loadCalls, equalTo(singletonList(asList("A", "B"))));
+        } else {
+            assertThat(loadCalls, equalTo(singletonList(asList("A", "B", "A"))));
         }
     }
 
@@ -987,7 +1000,7 @@ public void batches_multiple_requests_with_max_batch_size(TestDataLoaderFactory
 
         identityLoader.dispatch();
 
-        CompletableFuture.allOf(f1, f2, f3).join();
+        allOf(f1, f2, f3).join();
 
         assertThat(f1.join(), equalTo(1));
         assertThat(f2.join(), equalTo(2));
@@ -1030,13 +1043,13 @@ public void should_Batch_loads_occurring_within_futures(TestDataLoaderFactory fa
 
         AtomicBoolean v4Called = new AtomicBoolean();
 
-        CompletableFuture.supplyAsync(nullValue).thenAccept(v1 -> {
+        supplyAsync(nullValue).thenAccept(v1 -> {
             identityLoader.load("a");
-            CompletableFuture.supplyAsync(nullValue).thenAccept(v2 -> {
+            supplyAsync(nullValue).thenAccept(v2 -> {
                 identityLoader.load("b");
-                CompletableFuture.supplyAsync(nullValue).thenAccept(v3 -> {
+                supplyAsync(nullValue).thenAccept(v3 -> {
                     identityLoader.load("c");
-                    CompletableFuture.supplyAsync(nullValue).thenAccept(
+                    supplyAsync(nullValue).thenAccept(
                             v4 -> {
                                 identityLoader.load("d");
                                 v4Called.set(true);
@@ -1053,12 +1066,101 @@ public void should_Batch_loads_occurring_within_futures(TestDataLoaderFactory fa
                 singletonList(asList("a", "b", "c", "d"))));
     }
 
+    @ParameterizedTest
+    @MethodSource("dataLoaderFactories")
+    public void should_blowup_after_N_keys(TestDataLoaderFactory factory) {
+        if (!(factory instanceof TestReactiveDataLoaderFactory)) {
+            return;
+        }
+        //
+        // if we blow up after emitting N keys, the N keys should work but the rest of the keys
+        // should be exceptional
+        DataLoader<Integer, Integer> identityLoader = ((TestReactiveDataLoaderFactory) factory).idLoaderBlowsUpsAfterN(3, new DataLoaderOptions(), new ArrayList<>());
+        CompletableFuture<Integer> cf1 = identityLoader.load(1);
+        CompletableFuture<Integer> cf2 = identityLoader.load(2);
+        CompletableFuture<Integer> cf3 = identityLoader.load(3);
+        CompletableFuture<Integer> cf4 = identityLoader.load(4);
+        CompletableFuture<Integer> cf5 = identityLoader.load(5);
+        identityLoader.dispatch();
+        await().until(cf5::isDone);
+
+        assertThat(cf1.join(), equalTo(1));
+        assertThat(cf2.join(), equalTo(2));
+        assertThat(cf3.join(), equalTo(3));
+        assertThat(cf4.isCompletedExceptionally(), is(true));
+        assertThat(cf5.isCompletedExceptionally(), is(true));
+
+    }
+
+    @ParameterizedTest
+    @MethodSource("dataLoaderFactories")
+    public void when_values_size_are_less_then_key_size(TestDataLoaderFactory factory) {
+        //
+        // what happens if we want 4 values but are only given 2 back say
+        //
+        DataLoader<String, String> identityLoader = factory.onlyReturnsNValues(2, new DataLoaderOptions(), new ArrayList<>());
+        CompletableFuture<String> cf1 = identityLoader.load("A");
+        CompletableFuture<String> cf2 = identityLoader.load("B");
+        CompletableFuture<String> cf3 = identityLoader.load("C");
+        CompletableFuture<String> cf4 = identityLoader.load("D");
+        identityLoader.dispatch();
+
+        await().atMost(Duration.FIVE_SECONDS).until(() -> areAllDone(cf1, cf2, cf3, cf4));
+
+        if (factory instanceof ListDataLoaderFactory) {
+            assertThat(cause(cf1), instanceOf(DataLoaderAssertionException.class));
+            assertThat(cause(cf2), instanceOf(DataLoaderAssertionException.class));
+            assertThat(cause(cf3), instanceOf(DataLoaderAssertionException.class));
+            assertThat(cause(cf4), instanceOf(DataLoaderAssertionException.class));
+        } else if (factory instanceof PublisherDataLoaderFactory) {
+            // some have completed progressively but the other never did
+            assertThat(cf1.join(), equalTo("A"));
+            assertThat(cf2.join(), equalTo("B"));
+            assertThat(cause(cf3), instanceOf(DataLoaderAssertionException.class));
+            assertThat(cause(cf4), instanceOf(DataLoaderAssertionException.class));
+        } else {
+            // with the maps it's ok to have fewer results
+            assertThat(cf1.join(), equalTo("A"));
+            assertThat(cf2.join(), equalTo("B"));
+            assertThat(cf3.join(), equalTo(null));
+            assertThat(cf4.join(), equalTo(null));
+        }
+    }
+
+    @ParameterizedTest
+    @MethodSource("dataLoaderFactories")
+    public void when_values_size_are_more_then_key_size(TestDataLoaderFactory factory) {
+        //
+        // what happens if we want 4 values but only given 6 back say
+        //
+        DataLoader<String, String> identityLoader = factory.idLoaderReturnsTooMany(2, new DataLoaderOptions(), new ArrayList<>());
+        CompletableFuture<String> cf1 = identityLoader.load("A");
+        CompletableFuture<String> cf2 = identityLoader.load("B");
+        CompletableFuture<String> cf3 = identityLoader.load("C");
+        CompletableFuture<String> cf4 = identityLoader.load("D");
+        identityLoader.dispatch();
+        await().atMost(Duration.FIVE_SECONDS).until(() -> areAllDone(cf1, cf2, cf3, cf4));
+
+
+        if (factory instanceof ListDataLoaderFactory) {
+            assertThat(cause(cf1), instanceOf(DataLoaderAssertionException.class));
+            assertThat(cause(cf2), instanceOf(DataLoaderAssertionException.class));
+            assertThat(cause(cf3), instanceOf(DataLoaderAssertionException.class));
+            assertThat(cause(cf4), instanceOf(DataLoaderAssertionException.class));
+        } else {
+            assertThat(cf1.join(), equalTo("A"));
+            assertThat(cf2.join(), equalTo("B"));
+            assertThat(cf3.join(), equalTo("C"));
+            assertThat(cf4.join(), equalTo("D"));
+        }
+    }
+
     @Test
     public void can_call_a_loader_from_a_loader() throws Exception {
         List<Collection<String>> deepLoadCalls = new ArrayList<>();
         DataLoader<String, String> deepLoader = newDataLoader(keys -> {
             deepLoadCalls.add(keys);
-            return CompletableFuture.completedFuture(keys);
+            return completedFuture(keys);
         });
 
         List<Collection<String>> aLoadCalls = new ArrayList<>();
@@ -1078,7 +1180,7 @@ public void can_call_a_loader_from_a_loader() throws Exception {
         CompletableFuture<String> b1 = bLoader.load("B1");
         CompletableFuture<String> b2 = bLoader.load("B2");
 
-        CompletableFuture.allOf(
+        allOf(
                 aLoader.dispatch(),
                 deepLoader.dispatch(),
                 bLoader.dispatch(),
@@ -1104,11 +1206,10 @@ public void can_call_a_loader_from_a_loader() throws Exception {
     public void should_allow_composition_of_data_loader_calls() {
         UserManager userManager = new UserManager();
 
-        BatchLoader<Long, User> userBatchLoader = userIds -> CompletableFuture
-                .supplyAsync(() -> userIds
-                        .stream()
-                        .map(userManager::loadUserById)
-                        .collect(Collectors.toList()));
+        BatchLoader<Long, User> userBatchLoader = userIds -> supplyAsync(() -> userIds
+                .stream()
+                .map(userManager::loadUserById)
+                .collect(Collectors.toList()));
         DataLoader<Long, User> userLoader = newDataLoader(userBatchLoader);
 
         AtomicBoolean gandalfCalled = new AtomicBoolean(false);
@@ -1144,127 +1245,21 @@ private static CacheKey<JsonObject> getJsonObjectCacheMapFn() {
                 .collect(Collectors.joining());
     }
 
-    private static Stream<Arguments> dataLoaderFactories() {
-        return Stream.of(
-                Arguments.of(Named.of("List DataLoader", new ListDataLoaderFactory())),
-                Arguments.of(Named.of("Mapped DataLoader", new MappedDataLoaderFactory()))
-        );
-    }
-
-    public interface TestDataLoaderFactory {
-        <K, V> DataLoader<K, V> idLoader(DataLoaderOptions options, List<Collection<K>> loadCalls);
-        <K, V> DataLoader<K, V> idLoaderBlowsUps(DataLoaderOptions options, List<Collection<K>> loadCalls);
-        <K> DataLoader<K, Object> idLoaderAllExceptions(DataLoaderOptions options, List<Collection<K>> loadCalls);
-        DataLoader<Integer, Object> idLoaderOddEvenExceptions(DataLoaderOptions options, List<Collection<Integer>> loadCalls);
-    }
-
-    private static class ListDataLoaderFactory implements TestDataLoaderFactory {
-        @Override
-        public <K, V> DataLoader<K, V> idLoader(DataLoaderOptions options, List<Collection<K>> loadCalls) {
-            return newDataLoader(keys -> {
-                loadCalls.add(new ArrayList<>(keys));
-                @SuppressWarnings("unchecked")
-                List<V> values = keys.stream()
-                        .map(k -> (V) k)
-                        .collect(Collectors.toList());
-                return CompletableFuture.completedFuture(values);
-            }, options);
-        }
-
-        @Override
-        public <K, V> DataLoader<K, V> idLoaderBlowsUps(
-                DataLoaderOptions options, List<Collection<K>> loadCalls) {
-            return newDataLoader(keys -> {
-                loadCalls.add(new ArrayList<>(keys));
-                return TestKit.futureError();
-            }, options);
-        }
-
-        @Override
-        public <K> DataLoader<K, Object> idLoaderAllExceptions(DataLoaderOptions options, List<Collection<K>> loadCalls) {
-            return newDataLoader(keys -> {
-                loadCalls.add(new ArrayList<>(keys));
-
-                List<Object> errors = keys.stream().map(k -> new IllegalStateException("Error")).collect(Collectors.toList());
-                return CompletableFuture.completedFuture(errors);
-            }, options);
-        }
-
-        @Override
-        public DataLoader<Integer, Object> idLoaderOddEvenExceptions(DataLoaderOptions options, List<Collection<Integer>> loadCalls) {
-            return newDataLoader(keys -> {
-                loadCalls.add(new ArrayList<>(keys));
-
-                List<Object> errors = new ArrayList<>();
-                for (Integer key : keys) {
-                    if (key % 2 == 0) {
-                        errors.add(key);
-                    } else {
-                        errors.add(new IllegalStateException("Error"));
-                    }
-                }
-                return CompletableFuture.completedFuture(errors);
-            }, options);
-        }
-    }
-
-    private static class MappedDataLoaderFactory implements TestDataLoaderFactory {
-
-        @Override
-        public <K, V> DataLoader<K, V> idLoader(
-                DataLoaderOptions options, List<Collection<K>> loadCalls) {
-            return newMappedDataLoader((keys) -> {
-                loadCalls.add(new ArrayList<>(keys));
-                Map<K, V> map = new HashMap<>();
-                //noinspection unchecked
-                keys.forEach(k -> map.put(k, (V) k));
-                return CompletableFuture.completedFuture(map);
-            }, options);
-        }
-
-        @Override
-        public <K, V> DataLoader<K, V> idLoaderBlowsUps(DataLoaderOptions options, List<Collection<K>> loadCalls) {
-            return newMappedDataLoader((keys) -> {
-                loadCalls.add(new ArrayList<>(keys));
-                return futureError();
-            }, options);
-        }
-
-        @Override
-        public <K> DataLoader<K, Object> idLoaderAllExceptions(
-                DataLoaderOptions options, List<Collection<K>> loadCalls) {
-            return newMappedDataLoader(keys -> {
-                loadCalls.add(new ArrayList<>(keys));
-                Map<K, Object> errorByKey = new HashMap<>();
-                keys.forEach(k -> errorByKey.put(k, new IllegalStateException("Error")));
-                return CompletableFuture.completedFuture(errorByKey);
-            }, options);
-        }
-
-        @Override
-        public DataLoader<Integer, Object> idLoaderOddEvenExceptions(
-                DataLoaderOptions options, List<Collection<Integer>> loadCalls) {
-            return newMappedDataLoader(keys -> {
-                loadCalls.add(new ArrayList<>(keys));
-
-                Map<Integer, Object> errorByKey = new HashMap<>();
-                for (Integer key : keys) {
-                    if (key % 2 == 0) {
-                        errorByKey.put(key, key);
-                    } else {
-                        errorByKey.put(key, new IllegalStateException("Error"));
-                    }
-                }
-                return CompletableFuture.completedFuture(errorByKey);
-            }, options);
-        }
-    }
-
     private static class ThrowingCacheMap extends CustomCacheMap {
+
         @Override
         public CompletableFuture<Object> get(String key) {
             throw new RuntimeException("Cache implementation failed.");
         }
     }
+
+    private static Stream<Arguments> dataLoaderFactories() {
+        return Stream.of(
+                Arguments.of(Named.of("List DataLoader", new ListDataLoaderFactory())),
+                Arguments.of(Named.of("Mapped DataLoader", new MappedDataLoaderFactory())),
+                Arguments.of(Named.of("Publisher DataLoader", new PublisherDataLoaderFactory())),
+                Arguments.of(Named.of("Mapped Publisher DataLoader", new MappedPublisherDataLoaderFactory()))
+        );
+    }
 }
 
diff --git a/src/test/java/org/dataloader/fixtures/TestKit.java b/src/test/java/org/dataloader/fixtures/TestKit.java
index adffb06..c22988d 100644
--- a/src/test/java/org/dataloader/fixtures/TestKit.java
+++ b/src/test/java/org/dataloader/fixtures/TestKit.java
@@ -131,4 +131,13 @@ public static <T> Set<T> asSet(T... elements) {
     public static <T> Set<T> asSet(Collection<T> elements) {
         return new LinkedHashSet<>(elements);
     }
+
+    public static boolean areAllDone(CompletableFuture<?>... cfs) {
+        for (CompletableFuture<?> cf : cfs) {
+            if (! cf.isDone()) {
+                return false;
+            }
+        }
+        return true;
+    }
 }
diff --git a/src/test/java/org/dataloader/fixtures/UserManager.java b/src/test/java/org/dataloader/fixtures/UserManager.java
index 24fee0d..4fed3f7 100644
--- a/src/test/java/org/dataloader/fixtures/UserManager.java
+++ b/src/test/java/org/dataloader/fixtures/UserManager.java
@@ -1,5 +1,8 @@
 package org.dataloader.fixtures;
 
+import org.reactivestreams.Subscriber;
+import reactor.core.publisher.Flux;
+
 import java.util.HashMap;
 import java.util.LinkedHashMap;
 import java.util.List;
@@ -52,6 +55,14 @@ public List<User> loadUsersById(List<Long> userIds) {
         return userIds.stream().map(this::loadUserById).collect(Collectors.toList());
     }
 
+    public void publishUsersById(List<Long> userIds, Subscriber<? super User> userSubscriber) {
+        Flux.fromIterable(loadUsersById(userIds)).subscribe(userSubscriber);
+    }
+
+    public void publishUsersById(Set<Long> userIds, Subscriber<? super Map.Entry<Long, User>> userEntrySubscriber) {
+        Flux.fromIterable(loadMapOfUsersByIds(null, userIds).entrySet()).subscribe(userEntrySubscriber);
+    }
+
     public Map<Long, User> loadMapOfUsersByIds(SecurityCtx callCtx, Set<Long> userIds) {
         Map<Long, User> map = new HashMap<>();
         userIds.forEach(userId -> {
diff --git a/src/test/java/org/dataloader/fixtures/parameterized/ListDataLoaderFactory.java b/src/test/java/org/dataloader/fixtures/parameterized/ListDataLoaderFactory.java
new file mode 100644
index 0000000..ee1f1d7
--- /dev/null
+++ b/src/test/java/org/dataloader/fixtures/parameterized/ListDataLoaderFactory.java
@@ -0,0 +1,79 @@
+package org.dataloader.fixtures.parameterized;
+
+import org.dataloader.DataLoader;
+import org.dataloader.DataLoaderOptions;
+import org.dataloader.fixtures.TestKit;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.List;
+import java.util.stream.Collectors;
+
+import static java.util.concurrent.CompletableFuture.completedFuture;
+import static org.dataloader.DataLoaderFactory.newDataLoader;
+
+public class ListDataLoaderFactory implements TestDataLoaderFactory {
+    @Override
+    public <K> DataLoader<K, K> idLoader(DataLoaderOptions options, List<Collection<K>> loadCalls) {
+        return newDataLoader(keys -> {
+            loadCalls.add(new ArrayList<>(keys));
+            return completedFuture(keys);
+        }, options);
+    }
+
+    @Override
+    public <K> DataLoader<K, K> idLoaderBlowsUps(
+            DataLoaderOptions options, List<Collection<K>> loadCalls) {
+        return newDataLoader(keys -> {
+            loadCalls.add(new ArrayList<>(keys));
+            return TestKit.futureError();
+        }, options);
+    }
+
+    @Override
+    public <K> DataLoader<K, Object> idLoaderAllExceptions(DataLoaderOptions options, List<Collection<K>> loadCalls) {
+        return newDataLoader(keys -> {
+            loadCalls.add(new ArrayList<>(keys));
+
+            List<Object> errors = keys.stream().map(k -> new IllegalStateException("Error")).collect(Collectors.toList());
+            return completedFuture(errors);
+        }, options);
+    }
+
+    @Override
+    public DataLoader<Integer, Object> idLoaderOddEvenExceptions(DataLoaderOptions options, List<Collection<Integer>> loadCalls) {
+        return newDataLoader(keys -> {
+            loadCalls.add(new ArrayList<>(keys));
+
+            List<Object> errors = new ArrayList<>();
+            for (Integer key : keys) {
+                if (key % 2 == 0) {
+                    errors.add(key);
+                } else {
+                    errors.add(new IllegalStateException("Error"));
+                }
+            }
+            return completedFuture(errors);
+        }, options);
+    }
+
+    @Override
+    public DataLoader<String, String> onlyReturnsNValues(int N, DataLoaderOptions options, ArrayList<Object> loadCalls) {
+        return newDataLoader(keys -> {
+            loadCalls.add(new ArrayList<>(keys));
+            return completedFuture(keys.subList(0, N));
+        }, options);
+    }
+
+    @Override
+    public DataLoader<String, String> idLoaderReturnsTooMany(int howManyMore, DataLoaderOptions options, ArrayList<Object> loadCalls) {
+        return newDataLoader(keys -> {
+            loadCalls.add(new ArrayList<>(keys));
+            List<String> l = new ArrayList<>(keys);
+            for (int i = 0; i < howManyMore; i++) {
+                l.add("extra-" + i);
+            }
+            return completedFuture(l);
+        }, options);
+    }
+}
diff --git a/src/test/java/org/dataloader/fixtures/parameterized/MappedDataLoaderFactory.java b/src/test/java/org/dataloader/fixtures/parameterized/MappedDataLoaderFactory.java
new file mode 100644
index 0000000..8f41441
--- /dev/null
+++ b/src/test/java/org/dataloader/fixtures/parameterized/MappedDataLoaderFactory.java
@@ -0,0 +1,95 @@
+package org.dataloader.fixtures.parameterized;
+
+import org.dataloader.DataLoader;
+import org.dataloader.DataLoaderOptions;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+import static java.util.concurrent.CompletableFuture.completedFuture;
+import static org.dataloader.DataLoaderFactory.newMappedDataLoader;
+import static org.dataloader.fixtures.TestKit.futureError;
+
+public class MappedDataLoaderFactory implements TestDataLoaderFactory {
+
+    @Override
+    public <K> DataLoader<K, K> idLoader(
+            DataLoaderOptions options, List<Collection<K>> loadCalls) {
+        return newMappedDataLoader((keys) -> {
+            loadCalls.add(new ArrayList<>(keys));
+            Map<K, K> map = new HashMap<>();
+            keys.forEach(k -> map.put(k, k));
+            return completedFuture(map);
+        }, options);
+    }
+
+    @Override
+    public <K> DataLoader<K, K> idLoaderBlowsUps(DataLoaderOptions options, List<Collection<K>> loadCalls) {
+        return newMappedDataLoader((keys) -> {
+            loadCalls.add(new ArrayList<>(keys));
+            return futureError();
+        }, options);
+    }
+
+    @Override
+    public <K> DataLoader<K, Object> idLoaderAllExceptions(
+            DataLoaderOptions options, List<Collection<K>> loadCalls) {
+        return newMappedDataLoader(keys -> {
+            loadCalls.add(new ArrayList<>(keys));
+            Map<K, Object> errorByKey = new HashMap<>();
+            keys.forEach(k -> errorByKey.put(k, new IllegalStateException("Error")));
+            return completedFuture(errorByKey);
+        }, options);
+    }
+
+    @Override
+    public DataLoader<Integer, Object> idLoaderOddEvenExceptions(
+            DataLoaderOptions options, List<Collection<Integer>> loadCalls) {
+        return newMappedDataLoader(keys -> {
+            loadCalls.add(new ArrayList<>(keys));
+
+            Map<Integer, Object> errorByKey = new HashMap<>();
+            for (Integer key : keys) {
+                if (key % 2 == 0) {
+                    errorByKey.put(key, key);
+                } else {
+                    errorByKey.put(key, new IllegalStateException("Error"));
+                }
+            }
+            return completedFuture(errorByKey);
+        }, options);
+    }
+
+    @Override
+    public DataLoader<String, String> onlyReturnsNValues(int N, DataLoaderOptions options, ArrayList<Object> loadCalls) {
+        return newMappedDataLoader(keys -> {
+            loadCalls.add(new ArrayList<>(keys));
+
+            Map<String, String> collect = List.copyOf(keys).subList(0, N).stream().collect(Collectors.toMap(
+                    k -> k, v -> v
+            ));
+            return completedFuture(collect);
+        }, options);
+    }
+
+    @Override
+    public DataLoader<String, String> idLoaderReturnsTooMany(int howManyMore, DataLoaderOptions options, ArrayList<Object> loadCalls) {
+        return newMappedDataLoader(keys -> {
+            loadCalls.add(new ArrayList<>(keys));
+
+            List<String> l = new ArrayList<>(keys);
+            for (int i = 0; i < howManyMore; i++) {
+                l.add("extra-" + i);
+            }
+
+            Map<String, String> collect = l.stream().collect(Collectors.toMap(
+                    k -> k, v -> v
+            ));
+            return completedFuture(collect);
+        }, options);
+    }
+}
diff --git a/src/test/java/org/dataloader/fixtures/parameterized/MappedPublisherDataLoaderFactory.java b/src/test/java/org/dataloader/fixtures/parameterized/MappedPublisherDataLoaderFactory.java
new file mode 100644
index 0000000..9c92330
--- /dev/null
+++ b/src/test/java/org/dataloader/fixtures/parameterized/MappedPublisherDataLoaderFactory.java
@@ -0,0 +1,108 @@
+package org.dataloader.fixtures.parameterized;
+
+import org.dataloader.DataLoader;
+import org.dataloader.DataLoaderOptions;
+import org.dataloader.Try;
+import reactor.core.publisher.Flux;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+import static java.util.stream.Collectors.toList;
+import static java.util.stream.Collectors.toSet;
+import static org.dataloader.DataLoaderFactory.newMappedPublisherDataLoader;
+import static org.dataloader.DataLoaderFactory.newMappedPublisherDataLoaderWithTry;
+
+public class MappedPublisherDataLoaderFactory implements TestDataLoaderFactory, TestReactiveDataLoaderFactory {
+
+    @Override
+    public <K> DataLoader<K, K> idLoader(
+            DataLoaderOptions options, List<Collection<K>> loadCalls) {
+        return newMappedPublisherDataLoader((keys, subscriber) -> {
+            loadCalls.add(new ArrayList<>(keys));
+            Map<K, K> map = new HashMap<>();
+            keys.forEach(k -> map.put(k, k));
+            Flux.fromIterable(map.entrySet()).subscribe(subscriber);
+        }, options);
+    }
+
+    @Override
+    public <K> DataLoader<K, K> idLoaderBlowsUps(DataLoaderOptions options, List<Collection<K>> loadCalls) {
+        return newMappedPublisherDataLoader((keys, subscriber) -> {
+            loadCalls.add(new ArrayList<>(keys));
+            Flux.<Map.Entry<K, K>>error(new IllegalStateException("Error")).subscribe(subscriber);
+        }, options);
+    }
+
+    @Override
+    public <K> DataLoader<K, Object> idLoaderAllExceptions(
+            DataLoaderOptions options, List<Collection<K>> loadCalls) {
+        return newMappedPublisherDataLoaderWithTry((keys, subscriber) -> {
+            loadCalls.add(new ArrayList<>(keys));
+            Stream<Map.Entry<K, Try<Object>>> failures = keys.stream().map(k -> Map.entry(k, Try.failed(new IllegalStateException("Error"))));
+            Flux.fromStream(failures).subscribe(subscriber);
+        }, options);
+    }
+
+    @Override
+    public DataLoader<Integer, Object> idLoaderOddEvenExceptions(
+            DataLoaderOptions options, List<Collection<Integer>> loadCalls) {
+        return newMappedPublisherDataLoaderWithTry((keys, subscriber) -> {
+            loadCalls.add(new ArrayList<>(keys));
+
+            Map<Integer, Try<Object>> errorByKey = new HashMap<>();
+            for (Integer key : keys) {
+                if (key % 2 == 0) {
+                    errorByKey.put(key, Try.succeeded(key));
+                } else {
+                    errorByKey.put(key, Try.failed(new IllegalStateException("Error")));
+                }
+            }
+            Flux.fromIterable(errorByKey.entrySet()).subscribe(subscriber);
+        }, options);
+    }
+
+    @Override
+    public <K> DataLoader<K, K> idLoaderBlowsUpsAfterN(int N, DataLoaderOptions options, List<Collection<K>> loadCalls) {
+        return newMappedPublisherDataLoader((keys, subscriber) -> {
+            loadCalls.add(new ArrayList<>(keys));
+
+            List<K> nKeys = keys.stream().limit(N).collect(toList());
+            Flux<Map.Entry<K, K>> subFlux = Flux.fromIterable(nKeys).map(k -> Map.entry(k, k));
+            subFlux.concatWith(Flux.error(new IllegalStateException("Error")))
+                    .subscribe(subscriber);
+        }, options);
+    }
+
+    @Override
+    public DataLoader<String, String> onlyReturnsNValues(int N, DataLoaderOptions options, ArrayList<Object> loadCalls) {
+        return newMappedPublisherDataLoader((keys, subscriber) -> {
+            loadCalls.add(new ArrayList<>(keys));
+
+            List<String> nKeys = keys.stream().limit(N).collect(toList());
+            Flux.fromIterable(nKeys).map(k -> Map.entry(k, k))
+                    .subscribe(subscriber);
+        }, options);
+    }
+
+    @Override
+    public DataLoader<String, String> idLoaderReturnsTooMany(int howManyMore, DataLoaderOptions options, ArrayList<Object> loadCalls) {
+        return newMappedPublisherDataLoader((keys, subscriber) -> {
+            loadCalls.add(new ArrayList<>(keys));
+
+            List<String> l = new ArrayList<>(keys);
+            for (int i = 0; i < howManyMore; i++) {
+                l.add("extra-" + i);
+            }
+
+            Flux.fromIterable(l).map(k -> Map.entry(k, k))
+                    .subscribe(subscriber);
+        }, options);
+    }
+}
diff --git a/src/test/java/org/dataloader/fixtures/parameterized/PublisherDataLoaderFactory.java b/src/test/java/org/dataloader/fixtures/parameterized/PublisherDataLoaderFactory.java
new file mode 100644
index 0000000..d75ff38
--- /dev/null
+++ b/src/test/java/org/dataloader/fixtures/parameterized/PublisherDataLoaderFactory.java
@@ -0,0 +1,100 @@
+package org.dataloader.fixtures.parameterized;
+
+import org.dataloader.DataLoader;
+import org.dataloader.DataLoaderOptions;
+import org.dataloader.Try;
+import reactor.core.publisher.Flux;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.List;
+import java.util.stream.Stream;
+
+import static org.dataloader.DataLoaderFactory.newPublisherDataLoader;
+import static org.dataloader.DataLoaderFactory.newPublisherDataLoaderWithTry;
+
+public class PublisherDataLoaderFactory implements TestDataLoaderFactory, TestReactiveDataLoaderFactory {
+
+    @Override
+    public <K> DataLoader<K, K> idLoader(
+            DataLoaderOptions options, List<Collection<K>> loadCalls) {
+        return newPublisherDataLoader((keys, subscriber) -> {
+            loadCalls.add(new ArrayList<>(keys));
+            Flux.fromIterable(keys).subscribe(subscriber);
+        }, options);
+    }
+
+    @Override
+    public <K> DataLoader<K, K> idLoaderBlowsUps(DataLoaderOptions options, List<Collection<K>> loadCalls) {
+        return newPublisherDataLoader((keys, subscriber) -> {
+            loadCalls.add(new ArrayList<>(keys));
+            Flux.<K>error(new IllegalStateException("Error")).subscribe(subscriber);
+        }, options);
+    }
+
+    @Override
+    public <K> DataLoader<K, Object> idLoaderAllExceptions(
+            DataLoaderOptions options, List<Collection<K>> loadCalls) {
+        return newPublisherDataLoaderWithTry((keys, subscriber) -> {
+            loadCalls.add(new ArrayList<>(keys));
+            Stream<Try<Object>> failures = keys.stream().map(k -> Try.failed(new IllegalStateException("Error")));
+            Flux.fromStream(failures).subscribe(subscriber);
+        }, options);
+    }
+
+    @Override
+    public DataLoader<Integer, Object> idLoaderOddEvenExceptions(
+            DataLoaderOptions options, List<Collection<Integer>> loadCalls) {
+        return newPublisherDataLoaderWithTry((keys, subscriber) -> {
+            loadCalls.add(new ArrayList<>(keys));
+
+            List<Try<Object>> errors = new ArrayList<>();
+            for (Integer key : keys) {
+                if (key % 2 == 0) {
+                    errors.add(Try.succeeded(key));
+                } else {
+                    errors.add(Try.failed(new IllegalStateException("Error")));
+                }
+            }
+            Flux.fromIterable(errors).subscribe(subscriber);
+        }, options);
+    }
+
+    @Override
+    public <K> DataLoader<K, K> idLoaderBlowsUpsAfterN(int N, DataLoaderOptions options, List<Collection<K>> loadCalls) {
+        return newPublisherDataLoader((keys, subscriber) -> {
+            loadCalls.add(new ArrayList<>(keys));
+
+            List<K> nKeys = keys.subList(0, N);
+            Flux<K> subFlux = Flux.fromIterable(nKeys);
+            subFlux.concatWith(Flux.error(new IllegalStateException("Error")))
+                    .subscribe(subscriber);
+        }, options);
+    }
+
+    @Override
+    public DataLoader<String, String> onlyReturnsNValues(int N, DataLoaderOptions options, ArrayList<Object> loadCalls) {
+        return newPublisherDataLoader((keys, subscriber) -> {
+            loadCalls.add(new ArrayList<>(keys));
+
+            List<String> nKeys = keys.subList(0, N);
+            Flux.fromIterable(nKeys)
+                    .subscribe(subscriber);
+        }, options);
+    }
+
+    @Override
+    public DataLoader<String, String> idLoaderReturnsTooMany(int howManyMore, DataLoaderOptions options, ArrayList<Object> loadCalls) {
+        return newPublisherDataLoader((keys, subscriber) -> {
+            loadCalls.add(new ArrayList<>(keys));
+
+            List<String> l = new ArrayList<>(keys);
+            for (int i = 0; i < howManyMore; i++) {
+                l.add("extra-" + i);
+            }
+
+            Flux.fromIterable(l)
+                    .subscribe(subscriber);
+        }, options);
+    }
+}
diff --git a/src/test/java/org/dataloader/fixtures/parameterized/TestDataLoaderFactory.java b/src/test/java/org/dataloader/fixtures/parameterized/TestDataLoaderFactory.java
new file mode 100644
index 0000000..8c1bc22
--- /dev/null
+++ b/src/test/java/org/dataloader/fixtures/parameterized/TestDataLoaderFactory.java
@@ -0,0 +1,22 @@
+package org.dataloader.fixtures.parameterized;
+
+import org.dataloader.DataLoader;
+import org.dataloader.DataLoaderOptions;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.List;
+
+public interface TestDataLoaderFactory {
+    <K> DataLoader<K, K> idLoader(DataLoaderOptions options, List<Collection<K>> loadCalls);
+
+    <K> DataLoader<K, K> idLoaderBlowsUps(DataLoaderOptions options, List<Collection<K>> loadCalls);
+
+    <K> DataLoader<K, Object> idLoaderAllExceptions(DataLoaderOptions options, List<Collection<K>> loadCalls);
+
+    DataLoader<Integer, Object> idLoaderOddEvenExceptions(DataLoaderOptions options, List<Collection<Integer>> loadCalls);
+
+    DataLoader<String, String> onlyReturnsNValues(int N, DataLoaderOptions options, ArrayList<Object> loadCalls);
+
+    DataLoader<String, String> idLoaderReturnsTooMany(int howManyMore, DataLoaderOptions options, ArrayList<Object> loadCalls);
+}
diff --git a/src/test/java/org/dataloader/fixtures/parameterized/TestReactiveDataLoaderFactory.java b/src/test/java/org/dataloader/fixtures/parameterized/TestReactiveDataLoaderFactory.java
new file mode 100644
index 0000000..d45932c
--- /dev/null
+++ b/src/test/java/org/dataloader/fixtures/parameterized/TestReactiveDataLoaderFactory.java
@@ -0,0 +1,11 @@
+package org.dataloader.fixtures.parameterized;
+
+import org.dataloader.DataLoader;
+import org.dataloader.DataLoaderOptions;
+
+import java.util.Collection;
+import java.util.List;
+
+public interface TestReactiveDataLoaderFactory {
+    <K> DataLoader<K, K> idLoaderBlowsUpsAfterN(int N, DataLoaderOptions options, List<Collection<K>> loadCalls);
+}
diff --git a/src/test/java/org/dataloader/scheduler/BatchLoaderSchedulerTest.java b/src/test/java/org/dataloader/scheduler/BatchLoaderSchedulerTest.java
index 274ed8c..ff9ec8e 100644
--- a/src/test/java/org/dataloader/scheduler/BatchLoaderSchedulerTest.java
+++ b/src/test/java/org/dataloader/scheduler/BatchLoaderSchedulerTest.java
@@ -36,6 +36,11 @@ public <K, V> CompletionStage<List<V>> scheduleBatchLoader(ScheduledBatchLoaderC
         public <K, V> CompletionStage<Map<K, V>> scheduleMappedBatchLoader(ScheduledMappedBatchLoaderCall<K, V> scheduledCall, List<K> keys, BatchLoaderEnvironment environment) {
             return scheduledCall.invoke();
         }
+
+        @Override
+        public <K> void scheduleBatchPublisher(ScheduledBatchPublisherCall scheduledCall, List<K> keys, BatchLoaderEnvironment environment) {
+            scheduledCall.invoke();
+        }
     };
 
     private BatchLoaderScheduler delayedScheduling(int ms) {
@@ -56,6 +61,12 @@ public <K, V> CompletionStage<Map<K, V>> scheduleMappedBatchLoader(ScheduledMapp
                     return scheduledCall.invoke();
                 }).thenCompose(Function.identity());
             }
+
+            @Override
+            public <K> void scheduleBatchPublisher(ScheduledBatchPublisherCall scheduledCall, List<K> keys, BatchLoaderEnvironment environment) {
+                snooze(ms);
+                scheduledCall.invoke();
+            }
         };
     }
 
@@ -139,6 +150,15 @@ public <K, V> CompletionStage<Map<K, V>> scheduleMappedBatchLoader(ScheduledMapp
                     return scheduledCall.invoke();
                 }).thenCompose(Function.identity());
             }
+
+            @Override
+            public <K> void scheduleBatchPublisher(ScheduledBatchPublisherCall scheduledCall, List<K> keys, BatchLoaderEnvironment environment) {
+                CompletableFuture.supplyAsync(() -> {
+                    snooze(10);
+                    scheduledCall.invoke();
+                    return null;
+                });
+            }
         };
         DataLoaderOptions options = DataLoaderOptions.newOptions().setBatchLoaderScheduler(funkyScheduler);