Have MappedBatchPublisher take in a Set<K> keys

This is more symmetric with `MappedbatchLoader` and preserves
efficiency; we do not need to emit a `Map.Entry` for duplicate keys
(given the strong intention that this will be used to create a `Map`).

Author: AlexandreCarlton

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

@@ -536,7 +536,7 @@ private CompletableFuture<List<V>> invokeBatchPublisher(List<K> keys, List<Objec
     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
@@ -551,10 +551,10 @@ private CompletableFuture<List<V>> invokeMappedBatchPublisher(List<K> keys, List
             //noinspection unchecked
             MappedBatchPublisher<K, V> loadFunction = (MappedBatchPublisher<K, V>) batchLoadFunction;
             if (batchLoaderScheduler != null) {
-                BatchLoaderScheduler.ScheduledBatchPublisherCall loadCall = () -> loadFunction.load(keys, subscriber);
+                BatchLoaderScheduler.ScheduledBatchPublisherCall loadCall = () -> loadFunction.load(setOfKeys, subscriber);
                 batchLoaderScheduler.scheduleBatchPublisher(loadCall, keys, null);
             } else {
-                loadFunction.load(keys, subscriber);
+                loadFunction.load(setOfKeys, subscriber);
             }
         }
         return loadResult;

src/main/java/org/dataloader/MappedBatchPublisher.java

@@ -2,8 +2,8 @@
 
 import org.reactivestreams.Subscriber;
 
-import java.util.List;
 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.
@@ -26,5 +26,5 @@ public interface MappedBatchPublisher<K, V> {
      * @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<Map.Entry<K, V>> subscriber);
+    void load(Set<K> keys, Subscriber<Map.Entry<K, V>> subscriber);
 }

src/test/java/org/dataloader/DataLoaderTest.java

@@ -740,7 +740,7 @@ 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 MappedDataLoaderFactory) {
+        if (factory instanceof MappedDataLoaderFactory || factory instanceof MappedPublisherDataLoaderFactory) {
             assertThat(loadCalls, equalTo(singletonList(asList("A", "B"))));
         } else {
             assertThat(loadCalls, equalTo(singletonList(asList("A", "B", "A"))));

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 -> {

src/test/java/org/dataloader/fixtures/parameterized/MappedPublisherDataLoaderFactory.java

@@ -10,8 +10,12 @@
 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;
 
@@ -69,7 +73,7 @@ public <K> DataLoader<K, K> idLoaderBlowsUpsAfterN(int N, DataLoaderOptions opti
         return newMappedPublisherDataLoader((keys, subscriber) -> {
             loadCalls.add(new ArrayList<>(keys));
 
-            List<K> nKeys = keys.subList(0, N);
+            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);
@@ -81,7 +85,7 @@ public DataLoader<String, String> onlyReturnsNValues(int N, DataLoaderOptions op
         return newMappedPublisherDataLoader((keys, subscriber) -> {
             loadCalls.add(new ArrayList<>(keys));
 
-            List<String> nKeys = keys.subList(0, N);
+            List<String> nKeys = keys.stream().limit(N).collect(toList());
             Flux.fromIterable(nKeys).map(k -> Map.entry(k, k))
                     .subscribe(subscriber);
         }, options);