diff --git a/build.gradle b/build.gradle index 8ceecad3..67a9be6f 100644 --- a/build.gradle +++ b/build.gradle @@ -37,6 +37,7 @@ repositories { } } +def slf4jVersion = '1.7.25' def releaseVersion = System.properties.RELEASE_VERSION version = releaseVersion ? releaseVersion : new SimpleDateFormat('yyyy-MM-dd\'T\'HH-mm-ss').format(new Date()) group = 'org.dataloader' @@ -55,6 +56,8 @@ task myJavadocs(type: Javadoc) { } dependencies { + compile 'org.slf4j:slf4j-api:' + slf4jVersion + testCompile 'org.slf4j:slf4j-simple:' + slf4jVersion testCompile "junit:junit:$junitVersion" testCompile 'org.awaitility:awaitility:2.0.0' } diff --git a/src/main/java/org/dataloader/DataLoader.java b/src/main/java/org/dataloader/DataLoader.java index 14508e22..6ca4b619 100644 --- a/src/main/java/org/dataloader/DataLoader.java +++ b/src/main/java/org/dataloader/DataLoader.java @@ -57,7 +57,7 @@ public class DataLoader { private final DataLoaderHelper helper; - private final DataLoaderOptions loaderOptions; + protected final DataLoaderOptions loaderOptions; private final CacheMap> futureCache; private final StatisticsCollector stats; diff --git a/src/main/java/org/dataloader/nextgen/AutoDataLoader.java b/src/main/java/org/dataloader/nextgen/AutoDataLoader.java new file mode 100644 index 00000000..0fa5e833 --- /dev/null +++ b/src/main/java/org/dataloader/nextgen/AutoDataLoader.java @@ -0,0 +1,140 @@ +/* + * To change this license header, choose License Headers in Project Properties. + * To change this template file, choose Tools | Templates + * and open the template in the editor. + */ +package org.dataloader.nextgen; + +import java.util.ArrayList; +import static java.util.Collections.emptyList; +import java.util.List; +import java.util.Objects; +import java.util.concurrent.CompletableFuture; +import static java.util.concurrent.CompletableFuture.completedFuture; +import org.dataloader.BatchLoader; +import org.dataloader.DataLoader; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * An extension to the existing DataLoader that is automatically triggered + * by a background thread when the loading thread awaits for the data to arrive. + * + * @author Greg Kesler + */ +public class AutoDataLoader extends DataLoader implements Runnable, AutoCloseable { + private final Dispatcher dispatcher; + private volatile int requested; + private volatile CompletableFuture> result = completedFuture(emptyList()); + + private static final Logger LOGGER = LoggerFactory.getLogger(AutoDataLoader.class); + + public AutoDataLoader(BatchLoader batchLoadFunction) { + this(batchLoadFunction, new Dispatcher()); + } + + public AutoDataLoader(BatchLoader batchLoadFunction, Dispatcher dispatcher) { + this(batchLoadFunction, null, dispatcher); + } + + public AutoDataLoader(BatchLoader batchLoadFunction, AutoDataLoaderOptions options) { + this(batchLoadFunction, options, options.dispatcher()); + } + + private AutoDataLoader(BatchLoader batchLoadFunction, AutoDataLoaderOptions options, Dispatcher dispatcher) { + super(batchLoadFunction, options); + + Objects.requireNonNull(dispatcher); + this.dispatcher = dispatcher.register(this); + } + + @Override + protected void finalize() throws Throwable { + close(); + } + + @Override + public void close() throws Exception { + dispatcher.unregister(this); + } + + private synchronized void newResult () { + result = new Result(); + LOGGER.debug("created new result future"); + } + + @Override + public synchronized CompletableFuture load(K key, Object keyContext) { + LOGGER.debug("load requested, key={}", key); + CompletableFuture load = super.load(key, keyContext); + + if (loaderOptions.batchingEnabled()) { + LOGGER.debug("requesting dispatch for batched loader"); + // set read/modify/update fence here to ensure correct order of operations + if (++requested == 1) { + newResult(); + } + + dispatcher.scheduleBatch(this); + } + + return load; + } + + @Override + public void run() { + dispatchFully() + .thenAccept(value -> { + synchronized (this) { + if (result.complete(value)) { + requested = 0; + LOGGER.debug("run completed!"); + } + } + }); + } + + private CompletableFuture> dispatchFully () { + return (dispatchDepth() > 0) + ? super.dispatch() + .thenCombine(dispatchFully(), (value, temp) -> { + value.addAll(temp); + return value; + }) + : completedFuture(emptyList()); + } + + @Override + public int dispatchDepth() { + int dispatchDepth = super.dispatchDepth(); + LOGGER.debug("dispatchDept={}, requested={}", dispatchDepth, requested); + return dispatchDepth; + } + + + @Override + public synchronized CompletableFuture> dispatch() { + return result; + } + + @Override + public List dispatchAndJoin() { + return dispatch().join(); + } + + private class Result extends CompletableFuture> { + @Override + public boolean complete(List value) { + // set read fence here to ensure correct order of operations + int requestedCount = requested; + result.addAll(value); + int resultSize = result.size(); + LOGGER.debug("completing...requested={}, received={}", requestedCount, resultSize); + + return requestedCount == resultSize && + super.complete(result); + } + + private final List result = new ArrayList<>(); + } +} diff --git a/src/main/java/org/dataloader/nextgen/AutoDataLoaderOptions.java b/src/main/java/org/dataloader/nextgen/AutoDataLoaderOptions.java new file mode 100644 index 00000000..a2647fb4 --- /dev/null +++ b/src/main/java/org/dataloader/nextgen/AutoDataLoaderOptions.java @@ -0,0 +1,170 @@ +/* + * Copyright (c) 2016 The original author or authors + * + * All rights reserved. This program and the accompanying materials + * are made available under the terms of the Eclipse Public License v1.0 + * and Apache License v2.0 which accompanies this distribution. + * + * The Eclipse Public License is available at + * http://www.eclipse.org/legal/epl-v10.html + * + * The Apache License v2.0 is available at + * http://www.opensource.org/licenses/apache2.0.php + * + * You may elect to redistribute this code under either of these licenses. + */ + +package org.dataloader.nextgen; + +import org.dataloader.*; +import org.dataloader.stats.StatisticsCollector; + +import java.util.function.Supplier; + + +/** + * Configuration options for {@link AutoDataLoader} instances. + * + * @author Greg Kesler + */ +public class AutoDataLoaderOptions extends DataLoaderOptions { + public static final Dispatcher SHARED_DISPATCHER = new Dispatcher(); + + private Dispatcher dispatcher; + + /** + * Creates a new data loader options with default settings. + */ + public AutoDataLoaderOptions() { + dispatcher = SHARED_DISPATCHER; + } + + /** + * Clones the provided data loader options. + * + * @param other the other options instance + */ + public AutoDataLoaderOptions(AutoDataLoaderOptions other) { + super(other); + + this.dispatcher = other.dispatcher; + } + + /** + * @return a new default data loader options that you can then customize + */ + public static AutoDataLoaderOptions newOptions() { + return new AutoDataLoaderOptions(); + } + + /** + * Sets the option that determines whether batch loading is enabled. + * + * @param batchingEnabled {@code true} to enable batch loading, {@code false} otherwise + * + * @return the data loader options for fluent coding + */ + public AutoDataLoaderOptions setBatchingEnabled(boolean batchingEnabled) { + return (AutoDataLoaderOptions)super.setBatchingEnabled(batchingEnabled); + } + + /** + * Sets the option that determines whether caching is enabled. + * + * @param cachingEnabled {@code true} to enable caching, {@code false} otherwise + * + * @return the data loader options for fluent coding + */ + public AutoDataLoaderOptions setCachingEnabled(boolean cachingEnabled) { + return (AutoDataLoaderOptions)super.setCachingEnabled(cachingEnabled); + } + + /** + * Sets the option that determines whether exceptional values are cachedis enabled. + * + * @param cachingExceptionsEnabled {@code true} to enable caching exceptional values, {@code false} otherwise + * + * @return the data loader options for fluent coding + */ + public AutoDataLoaderOptions setCachingExceptionsEnabled(boolean cachingExceptionsEnabled) { + return (AutoDataLoaderOptions)super.setCachingExceptionsEnabled(cachingExceptionsEnabled); + } + + /** + * Sets the function to use for creating the cache key, if caching is enabled. + * + * @param cacheKeyFunction the cache key function to use + * + * @return the data loader options for fluent coding + */ + public AutoDataLoaderOptions setCacheKeyFunction(CacheKey cacheKeyFunction) { + return (AutoDataLoaderOptions)super.setCacheKeyFunction(cacheKeyFunction); + } + + /** + * Sets the cache map implementation to use for caching, if caching is enabled. + * + * @param cacheMap the cache map instance + * + * @return the data loader options for fluent coding + */ + public AutoDataLoaderOptions setCacheMap(CacheMap cacheMap) { + return (AutoDataLoaderOptions)super.setCacheMap(cacheMap); + } + + /** + * Sets the maximum number of keys that will be presented to the {@link BatchLoader} function + * before they are split into multiple class + * + * @param maxBatchSize the maximum batch size + * + * @return the data loader options for fluent coding + */ + public AutoDataLoaderOptions setMaxBatchSize(int maxBatchSize) { + return (AutoDataLoaderOptions)super.setMaxBatchSize(maxBatchSize); + } + + /** + * Sets the statistics collector supplier that will be used with these data loader options. Since it uses + * the supplier pattern, you can create a new statistics collector on each call or you can reuse + * a common value + * + * @param statisticsCollector the statistics collector to use + * + * @return the data loader options for fluent coding + */ + public AutoDataLoaderOptions setStatisticsCollector(Supplier statisticsCollector) { + return (AutoDataLoaderOptions)super.setStatisticsCollector(statisticsCollector); + } + + /** + * Sets the batch loader environment provider that will be used to give context to batch load functions + * + * @param contextProvider the batch loader context provider + * + * @return the data loader options for fluent coding + */ + public AutoDataLoaderOptions setBatchLoaderContextProvider(BatchLoaderContextProvider contextProvider) { + return (AutoDataLoaderOptions)super.setBatchLoaderContextProvider(contextProvider); + } + + /** + * Retrieves Dispatcher instance configured on this data loader opitons + * + * @return configured Dispatcher object + */ + public Dispatcher dispatcher () { + return dispatcher; + } + + /** + * Sets a Dispatcher to be used by AutoDataLoaders created with this object + * + * @param dispatcher new Dispatcher for these options + * @return this instance to allow method chaining + */ + public AutoDataLoaderOptions setDispatcher (Dispatcher dispatcher) { + this.dispatcher = dispatcher; + return this; + } +} diff --git a/src/main/java/org/dataloader/nextgen/Dispatcher.java b/src/main/java/org/dataloader/nextgen/Dispatcher.java new file mode 100644 index 00000000..35ee74ba --- /dev/null +++ b/src/main/java/org/dataloader/nextgen/Dispatcher.java @@ -0,0 +1,248 @@ +/* + * To change this license header, choose License Headers in Project Properties. + * To change this template file, choose Tools | Templates + * and open the template in the editor. + */ +package org.dataloader.nextgen; + +import java.util.Map; +import java.util.Objects; +import java.util.Optional; +import java.util.Queue; +import java.util.WeakHashMap; +import java.util.concurrent.ConcurrentLinkedQueue; +import java.util.concurrent.Executor; +import java.util.concurrent.ForkJoinPool; +import java.util.concurrent.atomic.AtomicIntegerFieldUpdater; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * "Magic" class to run AutoDataLoader batched dispatch mechanism when + * the data loader client thread becomes awaiting for results. + * Upon the requestor starts waiting, the temporary dispatcher thread + * kicks off and triggers AutoDataLoader.run method. Inferred load requests + * are also batched into the final result available via overriden AutoDataLoader.dispatch() + * method. + * + * To create AutoDataLoader that fully utilizes ForkJoinPool lightweight threading + * use the following code + * {@code + * Dispatcher commonPoolDispatcher = new Dispatcher(); + * BatchLoader batchLoader = ... + * AutoDataLoader commonPoolLoader = new AutoDataLoader<>(batchLoader, new AutoDataLoaderOptions()); + * } + * + * To create AutoDataLoader that is limited to run on 1 thread use the folllowing code + * {@code + * Dispatcher singleThreadDispatcher = new Dispatcher(Executors.newSingleThreadExecutor()); + * BatchLoader batchLoader = ... + * AutoDataLoader commonPoolLoader = new AutoDataLoader<>(batchLoader, singleThreadDispatcher); + * } + * + * To create AutoDataLoader with DataLoaderOptions and to choose threading model, use the code + * below: + * {@code + * Dispatcher singleThreadDispatcher = new Dispatcher(Executors.newSingleThreadExecutor()); + * AutoDataLoaderOptions options = new AutoDataLoaderOptions().setDispatcher(singleThreadDispatcher); + * BatchLoader batchLoader = ... + * AutoDataLoader commonPoolLoader = new AutoDataLoader<>(batchLoader, options); + * } + * + * @see AutoDataLoader#load(java.lang.Object) + * @see AutoDataLoader#loadMany(java.util.List) + * @see AutoDataLoader#run() + * @see AutoDataLoader#dispatch() + * + * @author Greg Kesler + */ +public class Dispatcher implements Runnable, AutoCloseable { + private final Map dataLoaders = new WeakHashMap<>(); + private final Queue commands = new ConcurrentLinkedQueue<>(); + private final Executor executor; + + private static final int IDLE = 0; + private static final int RUNNING = 1; + private static final int CLOSING = 2; + private static final int CLOSED = 3; + + private volatile int state = IDLE; + private static final AtomicIntegerFieldUpdater STATE = AtomicIntegerFieldUpdater + .newUpdater(Dispatcher.class, "state"); + + private static final Logger LOGGER = LoggerFactory.getLogger(Dispatcher.class); + + /** + * Creates a new instance that uses ForkJoinPool.commonPool() + */ + public Dispatcher () { + this(ForkJoinPool.commonPool()); + } + + /** + * Creates a new instance with specified Executor. + * + * @param executor executor to run Dispatcher tasks + */ + public Dispatcher (Executor executor) { + Objects.requireNonNull(executor); + + this.executor = executor; + } + + @Override + protected void finalize() throws Throwable { + close(); + } + + /** + * Registers an new AutoDataLoader with this Dispatcher. + * Dispatcher will associate current thread with the new data loader + * and it will monitor the state in order to trigger data loader dispatch + * + * @param dataLoader a data loader to monitor + * @return this instance to allow method chaining + */ + public synchronized Dispatcher register (AutoDataLoader dataLoader) { + Objects.requireNonNull(dataLoader); + + dataLoaders.put(dataLoader, Thread.currentThread()); + return this; + } + + /** + * Unregisters a AutoDataLoader from this Dispatcher. + * + * @param dataLoader a data loader to unregister + * @return this instance to allow method chaining + */ + public synchronized Dispatcher unregister (AutoDataLoader dataLoader) { + Objects.requireNonNull(dataLoader); + + dataLoaders.remove(dataLoader); + return this; + } + + private synchronized Optional ownerOf (AutoDataLoader dataLoader) { + return Optional.ofNullable(dataLoaders.get(dataLoader)); + } + + /** + * Requests to schedule batching for the specified data loader + * @param dataLoader data loader to schedule dispatch + */ + public void scheduleBatch (AutoDataLoader dataLoader) { + LOGGER.debug("dispatch requested for {}", dataLoader); + request(new DispatchCommand(dataLoader)); + } + + protected void request (Command command) { + Objects.requireNonNull(command); + + commands.offer(command); + if (STATE.compareAndSet(this, IDLE, RUNNING)) { + LOGGER.debug("scheduling execution for pended commands"); + executor.execute(this); + // FIXME: should I allow multithreaded dispatch for different dataloaders? + } + } + + private static boolean isWaiting (Thread thread) { + switch (thread.getState()) { + case WAITING: + case TIMED_WAITING: + return true; + } + + return false; + } + + @Override + public void close() { + if (STATE.compareAndSet(this, RUNNING, CLOSING)) { + request(new CloseCommand()); + LOGGER.debug("close requested..."); + + int s; + while ((s = state) == CLOSING); + } + + state = CLOSED; + LOGGER.debug("closed!"); + } + + @Override + public void run() { + try { + int s; + Command command; + while ((s = state) == RUNNING && (command = commands.poll()) != null) { + command.run(); + } + } finally { + LOGGER.debug("no more commands"); + STATE.compareAndSet(this, RUNNING, IDLE); + } + } + + protected abstract class Command implements Runnable { + protected abstract void execute (); + protected void handle (Throwable e) { + LOGGER.warn("{}", e); + } + + @Override + public void run() { + try { + LOGGER.debug("executing command {}", this); + execute(); + } catch (Throwable e) { + handle(e); + } finally { + LOGGER.debug("finished command {}", this); + } + } + } + + private class DispatchCommand extends Command { + private final AutoDataLoader dataLoader; + private final Thread owner; + + DispatchCommand (AutoDataLoader dataLoader) { + Objects.requireNonNull(dataLoader); + + this.dataLoader = dataLoader; + this.owner = ownerOf(dataLoader) + .orElseThrow(() -> new IllegalArgumentException("Unknown DataLoader " + dataLoader)); + } + + @Override + protected void execute() { + if (!isWaiting(owner)) { + // wait while thread is running + commands.offer(this); + } else { + // and now do the dispatch + dataLoader.run(); + } + } + + @Override + public String toString() { + return "DispatchCommand{" + "dataLoader=" + dataLoader + ", owner=" + owner + '}'; + } + } + + private class CloseCommand extends Command { + @Override + protected void execute() { + commands.clear(); + STATE.lazySet(Dispatcher.this, CLOSED); + } + + @Override + public String toString() { + return "CloseCommand{" + '}'; + } + } +} diff --git a/src/test/java/org/dataloader/JsonObject.java b/src/test/java/org/dataloader/JsonObject.java index 7509aec7..634886ef 100644 --- a/src/test/java/org/dataloader/JsonObject.java +++ b/src/test/java/org/dataloader/JsonObject.java @@ -4,11 +4,11 @@ import java.util.Map; import java.util.stream.Stream; -class JsonObject { +public class JsonObject { private final Map values; - JsonObject() { + public JsonObject() { values = new LinkedHashMap<>(); } diff --git a/src/test/java/org/dataloader/TestKit.java b/src/test/java/org/dataloader/TestKit.java index 82f73d62..fa65ee08 100644 --- a/src/test/java/org/dataloader/TestKit.java +++ b/src/test/java/org/dataloader/TestKit.java @@ -17,7 +17,7 @@ public static Collection listFrom(int i, int max) { return ints; } - static CompletableFuture futureError() { + public static CompletableFuture futureError() { return failedFuture(new IllegalStateException("Error")); } } diff --git a/src/test/java/org/dataloader/nextgen/AutoDataLoaderTest.java b/src/test/java/org/dataloader/nextgen/AutoDataLoaderTest.java new file mode 100644 index 00000000..d586992e --- /dev/null +++ b/src/test/java/org/dataloader/nextgen/AutoDataLoaderTest.java @@ -0,0 +1,1136 @@ +/* + * Copyright (c) 2016 The original author or authors + * + * All rights reserved. This program and the accompanying materials + * are made available under the terms of the Eclipse Public License v1.0 + * and Apache License v2.0 which accompanies this distribution. + * + * The Eclipse Public License is available at + * http://www.eclipse.org/legal/epl-v10.html + * + * The Apache License v2.0 is available at + * http://www.opensource.org/licenses/apache2.0.php + * + * You may elect to redistribute this code under either of these licenses. + */ + +package org.dataloader.nextgen; + +import org.dataloader.*; +import org.dataloader.fixtures.User; +import org.dataloader.fixtures.UserManager; +import org.dataloader.impl.CompletableFutureKit; +import org.junit.Test; + +import java.util.ArrayList; +import java.util.Collection; +import java.util.List; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CompletionStage; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.function.Supplier; +import java.util.stream.Collectors; + +import static java.util.Arrays.asList; +import static java.util.Collections.emptyList; +import static java.util.Collections.singletonList; +import java.util.HashSet; +import java.util.Set; +import static java.util.concurrent.CompletableFuture.allOf; +import java.util.concurrent.Executor; +import java.util.concurrent.Executors; +import java.util.stream.Stream; +import static org.awaitility.Awaitility.await; +import static org.dataloader.TestKit.listFrom; +import static org.dataloader.impl.CompletableFutureKit.cause; +import static org.hamcrest.Matchers.empty; +import static org.hamcrest.Matchers.equalTo; +import static org.hamcrest.Matchers.instanceOf; +import static org.hamcrest.Matchers.is; +import org.junit.After; +import static org.junit.Assert.assertArrayEquals; +import static org.junit.Assert.assertThat; +import org.junit.Before; +import org.junit.Ignore; +import org.junit.Rule; +import org.junit.rules.TestName; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * Tests for {@link AutoDataLoader}. + *

+ * The tests are a port of the existing tests in + * the facebook/dataloader project. + *

+ * Acknowledgments go to Lee Byron for providing excellent coverage. + * + * @author Greg Kesler + */ +public class AutoDataLoaderTest { + private volatile Dispatcher dispatcher; + + private static final Logger LOGGER = LoggerFactory.getLogger(AutoDataLoaderTest.class); + + @Before + public void setUp () throws Exception { + LOGGER.info("=========================================================="); + LOGGER.info("| {}", testName.getMethodName()); + LOGGER.info("=========================================================="); + + dispatcher = new Dispatcher(); + } + + @After + public void tierDown () { +// final AtomicBoolean closed = new AtomicBoolean(false); +// CompletableFuture +// .runAsync(dispatcher::close) +// .thenAccept(v -> closed.set(true)); +// +// await().untilAtomic(closed, is(true)); + dispatcher.close(); + } + + @Rule + public TestName testName = new TestName(); + + @Test + public void should_Build_a_really_really_simple_data_loader() { + AtomicBoolean success = new AtomicBoolean(); + DataLoader identityLoader = new AutoDataLoader<>(keysAsValues(), dispatcher); + + CompletionStage future1 = identityLoader.load(1); + + future1.thenAccept(value -> { + assertThat(value, equalTo(1)); + success.set(true); + }); +// identityLoader.dispatch(); + await().untilAtomic(success, is(true)); + } + + @Test + public void should_Support_loading_multiple_keys_in_one_call() { + AtomicBoolean success = new AtomicBoolean(); + DataLoader identityLoader = new AutoDataLoader<>(keysAsValues(), dispatcher); + + CompletionStage> futureAll = identityLoader.loadMany(asList(1, 2)); + futureAll.thenAccept(promisedValues -> { + assertThat(promisedValues.size(), is(2)); + success.set(true); + }); +// identityLoader.dispatch(); + await().untilAtomic(success, is(true)); + assertThat(futureAll.toCompletableFuture().join(), equalTo(asList(1, 2))); + } + + @Test + public void should_Resolve_to_empty_list_when_no_keys_supplied() { + AtomicBoolean success = new AtomicBoolean(); + DataLoader identityLoader = new AutoDataLoader<>(keysAsValues(), dispatcher); + CompletableFuture> futureEmpty = identityLoader.loadMany(emptyList()); + futureEmpty.thenAccept(promisedValues -> { + assertThat(promisedValues.size(), is(0)); + success.set(true); + }); +// identityLoader.dispatch(); + await().untilAtomic(success, is(true)); + assertThat(futureEmpty.join(), empty()); + } + + @Test + public void should_Batch_multiple_requests() throws ExecutionException, InterruptedException { + List> loadCalls = new ArrayList<>(); + DataLoader identityLoader = idLoader(new AutoDataLoaderOptions(), loadCalls, dispatcher); + + CompletableFuture future1 = identityLoader.load(1); + CompletableFuture future2 = identityLoader.load(2); +// identityLoader.dispatch(); + + await().until(() -> future1.isDone() && future2.isDone()); + assertThat(future1.get(), equalTo(1)); + assertThat(future2.get(), equalTo(2)); + assertThat(loadCalls, equalTo(singletonList(asList(1, 2)))); + } + + @Test + public void should_Coalesce_identical_requests() throws ExecutionException, InterruptedException { + List> loadCalls = new ArrayList<>(); + DataLoader identityLoader = idLoader(new AutoDataLoaderOptions(), loadCalls, dispatcher); + + CompletableFuture future1a = identityLoader.load(1); + CompletableFuture future1b = identityLoader.load(1); + assertThat(future1a, equalTo(future1b)); +// identityLoader.dispatch(); + + await().until(future1a::isDone); + assertThat(future1a.get(), equalTo(1)); + assertThat(future1b.get(), equalTo(1)); + assertThat(loadCalls, equalTo(singletonList(singletonList(1)))); + } + + @Test + public void should_Cache_repeated_requests() throws ExecutionException, InterruptedException { + List> loadCalls = new ArrayList<>(); + DataLoader identityLoader = idLoader(new AutoDataLoaderOptions(), loadCalls, dispatcher); + + CompletableFuture future1 = identityLoader.load("A"); + CompletableFuture future2 = identityLoader.load("B"); +// identityLoader.dispatch(); + + await().until(() -> future1.isDone() && future2.isDone()); + assertThat(future1.get(), equalTo("A")); + assertThat(future2.get(), equalTo("B")); + assertThat(loadCalls, equalTo(singletonList(asList("A", "B")))); + + CompletableFuture future1a = identityLoader.load("A"); + CompletableFuture future3 = identityLoader.load("C"); +// identityLoader.dispatch(); + + await().until(() -> future1a.isDone() && future3.isDone()); + assertThat(future1a.get(), equalTo("A")); + assertThat(future3.get(), equalTo("C")); + assertThat(loadCalls, equalTo(asList(asList("A", "B"), singletonList("C")))); + + CompletableFuture future1b = identityLoader.load("A"); + CompletableFuture future2a = identityLoader.load("B"); + CompletableFuture future3a = identityLoader.load("C"); +// identityLoader.dispatch(); + + await().until(() -> future1b.isDone() && future2a.isDone() && future3a.isDone()); + assertThat(future1b.get(), equalTo("A")); + assertThat(future2a.get(), equalTo("B")); + assertThat(future3a.get(), equalTo("C")); + assertThat(loadCalls, equalTo(asList(asList("A", "B"), singletonList("C")))); + } + + @Test + public void should_Not_redispatch_previous_load() throws ExecutionException, InterruptedException { + List> loadCalls = new ArrayList<>(); + DataLoader identityLoader = idLoader(new AutoDataLoaderOptions(), loadCalls, dispatcher); + + CompletableFuture future1 = identityLoader.load("A"); +// identityLoader.dispatch(); + + await().until(() -> future1.isDone()); + CompletableFuture future2 = identityLoader.load("B"); +// identityLoader.dispatch(); + + await().until(() -> future1.isDone() && future2.isDone()); + assertThat(future1.get(), equalTo("A")); + assertThat(future2.get(), equalTo("B")); + assertThat(loadCalls, equalTo(asList(singletonList("A"), singletonList("B")))); + } + + @Test + public void should_Cache_on_redispatch() throws ExecutionException, InterruptedException { + List> loadCalls = new ArrayList<>(); + DataLoader identityLoader = idLoader(new AutoDataLoaderOptions(), loadCalls, dispatcher); + + CompletableFuture future1 = identityLoader.load("A"); +// identityLoader.dispatch(); + + await().until(() -> future1.isDone()); + CompletableFuture> future2 = identityLoader.loadMany(asList("A", "B")); +// identityLoader.dispatch(); + + await().until(() -> future1.isDone() && future2.isDone()); + assertThat(future1.get(), equalTo("A")); + assertThat(future2.get(), equalTo(asList("A", "B"))); + assertThat(loadCalls, equalTo(asList(singletonList("A"), singletonList("B")))); + } + + @Test + public void should_Clear_single_value_in_loader() throws ExecutionException, InterruptedException { + List> loadCalls = new ArrayList<>(); + DataLoader identityLoader = idLoader(new AutoDataLoaderOptions(), loadCalls, dispatcher); + + CompletableFuture future1 = identityLoader.load("A"); + CompletableFuture future2 = identityLoader.load("B"); +// identityLoader.dispatch(); + + await().until(() -> future1.isDone() && future2.isDone()); + assertThat(future1.get(), equalTo("A")); + assertThat(future2.get(), equalTo("B")); + assertThat(loadCalls, equalTo(singletonList(asList("A", "B")))); + + identityLoader.clear("A"); + + CompletableFuture future1a = identityLoader.load("A"); + CompletableFuture future2a = identityLoader.load("B"); +// identityLoader.dispatch(); + + await().until(() -> future1a.isDone() && future2a.isDone()); + assertThat(future1a.get(), equalTo("A")); + assertThat(future2a.get(), equalTo("B")); + assertThat(loadCalls, equalTo(asList(asList("A", "B"), singletonList("A")))); + } + + @Test + public void should_Clear_all_values_in_loader() throws ExecutionException, InterruptedException { + List> loadCalls = new ArrayList<>(); + DataLoader identityLoader = idLoader(new AutoDataLoaderOptions(), loadCalls, dispatcher); + + CompletableFuture future1 = identityLoader.load("A"); + CompletableFuture future2 = identityLoader.load("B"); +// identityLoader.dispatch(); + + await().until(() -> future1.isDone() && future2.isDone()); + assertThat(future1.get(), equalTo("A")); + assertThat(future2.get(), equalTo("B")); + assertThat(loadCalls, equalTo(singletonList(asList("A", "B")))); + + identityLoader.clearAll(); + + CompletableFuture future1a = identityLoader.load("A"); + CompletableFuture future2a = identityLoader.load("B"); +// identityLoader.dispatch(); + + await().until(() -> future1a.isDone() && future2a.isDone()); + assertThat(future1a.get(), equalTo("A")); + assertThat(future2a.get(), equalTo("B")); + assertThat(loadCalls, equalTo(asList(asList("A", "B"), asList("A", "B")))); + } + + @Test + public void should_Allow_priming_the_cache() throws ExecutionException, InterruptedException { + List> loadCalls = new ArrayList<>(); + DataLoader identityLoader = idLoader(new AutoDataLoaderOptions(), loadCalls, dispatcher); + + identityLoader.prime("A", "A"); + + CompletableFuture future1 = identityLoader.load("A"); + CompletableFuture future2 = identityLoader.load("B"); +// identityLoader.dispatch(); + + await().until(() -> future1.isDone() && future2.isDone()); + assertThat(future1.get(), equalTo("A")); + assertThat(future2.get(), equalTo("B")); + assertThat(loadCalls, equalTo(singletonList(singletonList("B")))); + } + + @Test + public void should_Not_prime_keys_that_already_exist() throws ExecutionException, InterruptedException { + List> loadCalls = new ArrayList<>(); + AutoDataLoader identityLoader = idLoader(new AutoDataLoaderOptions(), loadCalls, dispatcher); + + identityLoader.prime("A", "X"); + + CompletableFuture future1 = identityLoader.load("A"); + CompletableFuture future2 = identityLoader.load("B"); +// CompletableFuture> composite = identityLoader.dispatch(); + +// await().until(composite::isDone); + await().until(() -> future1.isDone() && future2.isDone()); + assertThat(future1.get(), equalTo("X")); + assertThat(future2.get(), equalTo("B")); + + identityLoader.prime("A", "Y"); + identityLoader.prime("B", "Y"); + + CompletableFuture future1a = identityLoader.load("A"); + CompletableFuture future2a = identityLoader.load("B"); +// CompletableFuture> composite2 = identityLoader.dispatch(); + +// await().until(composite2::isDone); + await().until(() -> future1a.isDone() && future2a.isDone()); + assertThat(future1a.get(), equalTo("X")); + assertThat(future2a.get(), equalTo("B")); + assertThat(loadCalls, equalTo(singletonList(singletonList("B")))); + } + + @Test + public void should_Allow_to_forcefully_prime_the_cache() throws ExecutionException, InterruptedException { + List> loadCalls = new ArrayList<>(); + AutoDataLoader identityLoader = idLoader(new AutoDataLoaderOptions(), loadCalls, dispatcher); + + identityLoader.prime("A", "X"); + + CompletableFuture future1 = identityLoader.load("A"); + CompletableFuture future2 = identityLoader.load("B"); +// CompletableFuture> composite = identityLoader.dispatch(); + +// await().until(composite::isDone); + await().until(() -> future1.isDone() && future2.isDone()); + assertThat(future1.get(), equalTo("X")); + assertThat(future2.get(), equalTo("B")); + + identityLoader.clear("A").prime("A", "Y"); + identityLoader.clear("B").prime("B", "Y"); + + CompletableFuture future1a = identityLoader.load("A"); + CompletableFuture future2a = identityLoader.load("B"); +// CompletableFuture> composite2 = identityLoader.dispatch(); + +// await().until(composite2::isDone); + await().until(() -> future1a.isDone() && future2a.isDone()); + assertThat(future1a.get(), equalTo("Y")); + assertThat(future2a.get(), equalTo("Y")); + assertThat(loadCalls, equalTo(singletonList(singletonList("B")))); + } + + @Test + public void should_not_Cache_failed_fetches_on_complete_failure() { + List> loadCalls = new ArrayList<>(); + DataLoader errorLoader = idLoaderBlowsUps(new AutoDataLoaderOptions(), loadCalls, dispatcher); + + CompletableFuture future1 = errorLoader.load(1); +// errorLoader.dispatch(); + + await().until(future1::isDone); + assertThat(future1.isCompletedExceptionally(), is(true)); + assertThat(cause(future1), instanceOf(IllegalStateException.class)); + + CompletableFuture future2 = errorLoader.load(1); +// errorLoader.dispatch(); + + await().until(future2::isDone); + assertThat(future2.isCompletedExceptionally(), is(true)); + assertThat(cause(future2), instanceOf(IllegalStateException.class)); + assertThat(loadCalls, equalTo(asList(singletonList(1), singletonList(1)))); + } + + @Test + public void should_Resolve_to_error_to_indicate_failure() throws ExecutionException, InterruptedException { + List> loadCalls = new ArrayList<>(); + DataLoader evenLoader = idLoaderOddEvenExceptions(new AutoDataLoaderOptions(), loadCalls, dispatcher); + + CompletableFuture future1 = evenLoader.load(1); +// evenLoader.dispatch(); + + await().until(future1::isDone); + assertThat(future1.isCompletedExceptionally(), is(true)); + assertThat(cause(future1), instanceOf(IllegalStateException.class)); + + CompletableFuture future2 = evenLoader.load(2); +// evenLoader.dispatch(); + + await().until(future2::isDone); + assertThat(future2.get(), equalTo(2)); + assertThat(loadCalls, equalTo(asList(singletonList(1), singletonList(2)))); + } + + // Accept any kind of key. + + @Test + public void should_Represent_failures_and_successes_simultaneously() throws ExecutionException, InterruptedException { + AtomicBoolean success = new AtomicBoolean(); + List> loadCalls = new ArrayList<>(); + AutoDataLoader evenLoader = idLoaderOddEvenExceptions(new AutoDataLoaderOptions(), loadCalls, dispatcher); + + CompletableFuture future1 = evenLoader.load(1); + CompletableFuture future2 = evenLoader.load(2); + CompletableFuture future3 = evenLoader.load(3); + CompletableFuture future4 = evenLoader.load(4); + + await().until(allOf(future1, future2, future3, future4)::isDone); + CompletableFuture> result = evenLoader.dispatch(); + result.thenAccept(promisedValues -> success.set(true)); + + await().untilAtomic(success, is(true)); + + assertThat(future1.isCompletedExceptionally(), is(true)); + assertThat(cause(future1), instanceOf(IllegalStateException.class)); + assertThat(future2.get(), equalTo(2)); + assertThat(future3.isCompletedExceptionally(), is(true)); + assertThat(future4.get(), equalTo(4)); + + assertThat(loadCalls, equalTo(singletonList(asList(1, 2, 3, 4)))); + } + + // Accepts options + + @Test + public void should_Cache_failed_fetches() { + List> loadCalls = new ArrayList<>(); + DataLoader errorLoader = idLoaderAllExceptions(new AutoDataLoaderOptions(), loadCalls, dispatcher); + + CompletableFuture future1 = errorLoader.load(1); +// errorLoader.dispatch(); + + await().until(future1::isDone); + assertThat(future1.isCompletedExceptionally(), is(true)); + assertThat(cause(future1), instanceOf(IllegalStateException.class)); + + CompletableFuture future2 = errorLoader.load(1); +// errorLoader.dispatch(); + + await().until(future2::isDone); + assertThat(future2.isCompletedExceptionally(), is(true)); + assertThat(cause(future2), instanceOf(IllegalStateException.class)); + + assertThat(loadCalls, equalTo(singletonList(singletonList(1)))); + } + + @Test + public void should_NOT_Cache_failed_fetches_if_told_not_too() { + AutoDataLoaderOptions options = AutoDataLoaderOptions.newOptions().setCachingExceptionsEnabled(false); + List> loadCalls = new ArrayList<>(); + DataLoader errorLoader = idLoaderAllExceptions(options, loadCalls, dispatcher); + + CompletableFuture future1 = errorLoader.load(1); +// errorLoader.dispatch(); + + await().until(future1::isDone); + assertThat(future1.isCompletedExceptionally(), is(true)); + assertThat(cause(future1), instanceOf(IllegalStateException.class)); + + CompletableFuture future2 = errorLoader.load(1); +// errorLoader.dispatch(); + + await().until(future2::isDone); + assertThat(future2.isCompletedExceptionally(), is(true)); + assertThat(cause(future2), instanceOf(IllegalStateException.class)); + + assertThat(loadCalls, equalTo(asList(singletonList(1), singletonList(1)))); + } + + + // Accepts object key in custom cacheKey function + + @Test + public void should_Handle_priming_the_cache_with_an_error() { + List> loadCalls = new ArrayList<>(); + DataLoader identityLoader = idLoader(new AutoDataLoaderOptions(), loadCalls, dispatcher); + + identityLoader.prime(1, new IllegalStateException("Error")); + + CompletableFuture future1 = identityLoader.load(1); +// identityLoader.dispatch(); + + await().until(future1::isDone); + assertThat(future1.isCompletedExceptionally(), is(true)); + assertThat(cause(future1), instanceOf(IllegalStateException.class)); + assertThat(loadCalls, equalTo(emptyList())); + } + + @Test + public void should_Clear_values_from_cache_after_errors() { + List> loadCalls = new ArrayList<>(); + DataLoader errorLoader = idLoaderBlowsUps(new AutoDataLoaderOptions(), loadCalls, dispatcher); + + CompletableFuture future1 = errorLoader.load(1); + future1.handle((value, t) -> { + if (t != null) { + // Presumably determine if this error is transient, and only clear the cache in that case. + errorLoader.clear(1); + } + return null; + }); +// errorLoader.dispatch(); + + await().until(future1::isDone); + assertThat(future1.isCompletedExceptionally(), is(true)); + assertThat(cause(future1), instanceOf(IllegalStateException.class)); + + CompletableFuture future2 = errorLoader.load(1); + future2.handle((value, t) -> { + if (t != null) { + // Again, only do this if you can determine the error is transient. + errorLoader.clear(1); + } + return null; + }); +// errorLoader.dispatch(); + + await().until(future2::isDone); + assertThat(future2.isCompletedExceptionally(), is(true)); + assertThat(cause(future2), instanceOf(IllegalStateException.class)); + assertThat(loadCalls, equalTo(asList(singletonList(1), singletonList(1)))); + } + + @Test + public void should_Propagate_error_to_all_loads() { + List> loadCalls = new ArrayList<>(); + DataLoader errorLoader = idLoaderBlowsUps(new AutoDataLoaderOptions(), loadCalls, dispatcher); + + CompletableFuture future1 = errorLoader.load(1); + CompletableFuture future2 = errorLoader.load(2); +// errorLoader.dispatch(); + + await().until(future1::isDone); + assertThat(future1.isCompletedExceptionally(), is(true)); + Throwable cause = cause(future1); + assert cause != null; + assertThat(cause, instanceOf(IllegalStateException.class)); + assertThat(cause.getMessage(), equalTo("Error")); + + await().until(future2::isDone); + cause = cause(future2); + assert cause != null; + assertThat(cause.getMessage(), equalTo(cause.getMessage())); + assertThat(loadCalls, equalTo(singletonList(asList(1, 2)))); + } + + @Test + public void should_Accept_objects_as_keys() { + List> loadCalls = new ArrayList<>(); + AutoDataLoader identityLoader = idLoader(new AutoDataLoaderOptions(), loadCalls, dispatcher); + + Object keyA = new Object(); + Object keyB = new Object(); + + // Fetches as expected + + CompletableFuture a = identityLoader.load(keyA); + CompletableFuture b = identityLoader.load(keyB); + +// await().until(identityLoader.dispatch()::isDone); + await().until(() -> a.isDone() && b.isDone()); + identityLoader.dispatch().thenAccept(promisedValues -> { + assertThat(promisedValues.get(0), equalTo(keyA)); + assertThat(promisedValues.get(1), equalTo(keyB)); + }); + + assertThat(loadCalls.size(), equalTo(1)); + assertThat(loadCalls.get(0).size(), equalTo(2)); + assertThat(loadCalls.get(0).toArray()[0], equalTo(keyA)); + assertThat(loadCalls.get(0).toArray()[1], equalTo(keyB)); + + // Caching + identityLoader.clear(keyA); + //noinspection SuspiciousMethodCalls + loadCalls.remove(keyA); + + CompletableFuture nextA = identityLoader.load(keyA); + CompletableFuture nextB = identityLoader.load(keyB); + +// await().until(identityLoader.dispatch()::isDone); + await().until(allOf(nextA, nextA)::isDone); + identityLoader.dispatch().thenAccept(promisedValues -> { + assertThat(promisedValues.get(0), equalTo(keyA)); + assertThat(identityLoader.getCacheKey(keyB), equalTo(keyB)); + }); + + assertThat(loadCalls.size(), equalTo(2)); + assertThat(loadCalls.get(1).size(), equalTo(1)); + assertThat(loadCalls.get(1).toArray()[0], equalTo(keyA)); + } + + @Test + public void should_Disable_caching() throws ExecutionException, InterruptedException { + List> loadCalls = new ArrayList<>(); + DataLoader identityLoader = + idLoader(newOptions().setCachingEnabled(false), loadCalls, dispatcher); + + CompletableFuture future1 = identityLoader.load("A"); + CompletableFuture future2 = identityLoader.load("B"); +// identityLoader.dispatch(); + + await().until(() -> future1.isDone() && future2.isDone()); + assertThat(future1.get(), equalTo("A")); + assertThat(future2.get(), equalTo("B")); + assertThat(loadCalls, equalTo(singletonList(asList("A", "B")))); + + CompletableFuture future1a = identityLoader.load("A"); + CompletableFuture future3 = identityLoader.load("C"); +// identityLoader.dispatch(); + + await().until(() -> future1a.isDone() && future3.isDone()); + assertThat(future1a.get(), equalTo("A")); + assertThat(future3.get(), equalTo("C")); + assertThat(loadCalls, equalTo(asList(asList("A", "B"), asList("A", "C")))); + + CompletableFuture future1b = identityLoader.load("A"); + CompletableFuture future2a = identityLoader.load("B"); + CompletableFuture future3a = identityLoader.load("C"); +// identityLoader.dispatch(); + + await().until(() -> future1b.isDone() && future2a.isDone() && future3a.isDone()); + assertThat(future1b.get(), equalTo("A")); + assertThat(future2a.get(), equalTo("B")); + assertThat(future3a.get(), equalTo("C")); + assertThat(loadCalls, equalTo(asList(asList("A", "B"), + asList("A", "C"), asList("A", "B", "C")))); + } + + @Test + public void should_work_with_duplicate_keys_when_caching_disabled() throws ExecutionException, InterruptedException { + List> loadCalls = new ArrayList<>(); + DataLoader identityLoader = + idLoader(newOptions().setCachingEnabled(false), loadCalls, dispatcher); + + CompletableFuture future1 = identityLoader.load("A"); + CompletableFuture future2 = identityLoader.load("B"); + CompletableFuture future3 = identityLoader.load("A"); +// identityLoader.dispatch(); + + await().until(() -> future1.isDone() && future2.isDone() && future3.isDone()); + assertThat(future1.get(), equalTo("A")); + assertThat(future2.get(), equalTo("B")); + assertThat(future3.get(), equalTo("A")); + assertThat(loadCalls, equalTo(singletonList(asList("A", "B", "A")))); + } + + @Test + public void should_work_with_duplicate_keys_when_caching_enabled() throws ExecutionException, InterruptedException { + List> loadCalls = new ArrayList<>(); + DataLoader identityLoader = + idLoader(newOptions().setCachingEnabled(true), loadCalls, dispatcher); + + CompletableFuture future1 = identityLoader.load("A"); + CompletableFuture future2 = identityLoader.load("B"); + CompletableFuture future3 = identityLoader.load("A"); +// identityLoader.dispatch(); + + await().until(() -> future1.isDone() && future2.isDone() && future3.isDone()); + assertThat(future1.get(), equalTo("A")); + assertThat(future2.get(), equalTo("B")); + assertThat(future3.get(), equalTo("A")); + assertThat(loadCalls, equalTo(singletonList(asList("A", "B")))); + } + + // It is resilient to job queue ordering + + @Test + public void should_Accept_objects_with_a_complex_key() throws ExecutionException, InterruptedException { + List> loadCalls = new ArrayList<>(); + AutoDataLoaderOptions options = newOptions().setCacheKeyFunction(getJsonObjectCacheMapFn()); + DataLoader identityLoader = idLoader(options, loadCalls, dispatcher); + + JsonObject key1 = new JsonObject().put("id", 123); + JsonObject key2 = new JsonObject().put("id", 123); + + CompletableFuture future1 = identityLoader.load(key1); + CompletableFuture future2 = identityLoader.load(key2); +// identityLoader.dispatch(); + + await().until(() -> future1.isDone() && future2.isDone()); + assertThat(loadCalls, equalTo(singletonList(singletonList(key1)))); + assertThat(future1.get(), equalTo(key1)); + assertThat(future2.get(), equalTo(key1)); + } + + // Helper methods + + @Test + public void should_Clear_objects_with_complex_key() throws ExecutionException, InterruptedException { + List> loadCalls = new ArrayList<>(); + AutoDataLoaderOptions options = newOptions().setCacheKeyFunction(getJsonObjectCacheMapFn()); + DataLoader identityLoader = idLoader(options, loadCalls, dispatcher); + + JsonObject key1 = new JsonObject().put("id", 123); + JsonObject key2 = new JsonObject().put("id", 123); + + CompletableFuture future1 = identityLoader.load(key1); +// identityLoader.dispatch(); + + await().until(future1::isDone); + identityLoader.clear(key2); // clear equivalent object key + + CompletableFuture future2 = identityLoader.load(key1); +// identityLoader.dispatch(); + + await().until(future2::isDone); + assertThat(loadCalls, equalTo(asList(singletonList(key1), singletonList(key1)))); + assertThat(future1.get(), equalTo(key1)); + assertThat(future2.get(), equalTo(key1)); + } + + @Test + public void should_Accept_objects_with_different_order_of_keys() throws ExecutionException, InterruptedException { + List> loadCalls = new ArrayList<>(); + AutoDataLoaderOptions options = newOptions().setCacheKeyFunction(getJsonObjectCacheMapFn()); + DataLoader identityLoader = idLoader(options, loadCalls, dispatcher); + + JsonObject key1 = new JsonObject().put("a", 123).put("b", 321); + JsonObject key2 = new JsonObject().put("b", 321).put("a", 123); + + // Fetches as expected + + CompletableFuture future1 = identityLoader.load(key1); + CompletableFuture future2 = identityLoader.load(key2); +// identityLoader.dispatch(); + + await().until(() -> future1.isDone() && future2.isDone()); + assertThat(loadCalls, equalTo(singletonList(singletonList(key1)))); + assertThat(loadCalls.size(), equalTo(1)); + assertThat(future1.get(), equalTo(key1)); + assertThat(future2.get(), equalTo(key1)); + } + + @Test + public void should_Allow_priming_the_cache_with_an_object_key() throws ExecutionException, InterruptedException { + List> loadCalls = new ArrayList<>(); + AutoDataLoaderOptions options = newOptions().setCacheKeyFunction(getJsonObjectCacheMapFn()); + DataLoader identityLoader = idLoader(options, loadCalls, dispatcher); + + JsonObject key1 = new JsonObject().put("id", 123); + JsonObject key2 = new JsonObject().put("id", 123); + + identityLoader.prime(key1, key1); + + CompletableFuture future1 = identityLoader.load(key1); + CompletableFuture future2 = identityLoader.load(key2); +// identityLoader.dispatch(); + + await().until(() -> future1.isDone() && future2.isDone()); + assertThat(loadCalls, equalTo(emptyList())); + assertThat(future1.get(), equalTo(key1)); + assertThat(future2.get(), equalTo(key1)); + } + + @Test + public void should_Accept_a_custom_cache_map_implementation() throws ExecutionException, InterruptedException { + CustomCacheMap customMap = new CustomCacheMap(); + List> loadCalls = new ArrayList<>(); + AutoDataLoaderOptions options = newOptions().setCacheMap(customMap); + AutoDataLoader identityLoader = idLoader(options, loadCalls, dispatcher); + + // Fetches as expected + + CompletableFuture future1 = identityLoader.load("a"); + CompletableFuture future2 = identityLoader.load("b"); +// CompletableFuture> composite = identityLoader.dispatch(); + +// await().until(composite::isDone); + await().until(allOf(future1, future2)::isDone); + assertThat(future1.get(), equalTo("a")); + assertThat(future2.get(), equalTo("b")); + + assertThat(loadCalls, equalTo(singletonList(asList("a", "b")))); + assertArrayEquals(customMap.stash.keySet().toArray(), asList("a", "b").toArray()); + + CompletableFuture future3 = identityLoader.load("c"); + CompletableFuture future2a = identityLoader.load("b"); +// composite = identityLoader.dispatch(); + + await().until(() -> future3.isDone() && future2a.isDone()); + assertThat(future3.get(), equalTo("c")); + assertThat(future2a.get(), equalTo("b")); + + assertThat(loadCalls, equalTo(asList(asList("a", "b"), singletonList("c")))); + assertArrayEquals(customMap.stash.keySet().toArray(), asList("a", "b", "c").toArray()); + + // Supports clear + + identityLoader.clear("b"); + assertArrayEquals(customMap.stash.keySet().toArray(), asList("a", "c").toArray()); + + CompletableFuture future2b = identityLoader.load("b"); +// composite = identityLoader.dispatch(); + +// await().until(composite::isDone); + await().until(future2b::isDone); + assertThat(future2b.get(), equalTo("b")); + assertThat(loadCalls, equalTo(asList(asList("a", "b"), + singletonList("c"), singletonList("b")))); + assertArrayEquals(customMap.stash.keySet().toArray(), asList("a", "c", "b").toArray()); + + // Supports clear all + + identityLoader.clearAll(); + assertArrayEquals(customMap.stash.keySet().toArray(), emptyList().toArray()); + } + + @Test + public void batching_disabled_should_dispatch_immediately() throws Exception { + List> loadCalls = new ArrayList<>(); + AutoDataLoaderOptions options = newOptions().setBatchingEnabled(false); + DataLoader identityLoader = idLoader(options, loadCalls, dispatcher); + + CompletableFuture fa = identityLoader.load("A"); + CompletableFuture fb = identityLoader.load("B"); + + // caching is on still + CompletableFuture fa1 = identityLoader.load("A"); + CompletableFuture fb1 = identityLoader.load("B"); + + List values = CompletableFutureKit.allOf(asList(fa, fb, fa1, fb1)).join(); + + assertThat(fa.join(), equalTo("A")); + assertThat(fb.join(), equalTo("B")); + assertThat(fa1.join(), equalTo("A")); + assertThat(fb1.join(), equalTo("B")); + + assertThat(values, equalTo(asList("A", "B", "A", "B"))); + + assertThat(loadCalls, equalTo(asList( + singletonList("A"), + singletonList("B")))); + + } + + @Test + public void batching_disabled_and_caching_disabled_should_dispatch_immediately_and_forget() throws Exception { + List> loadCalls = new ArrayList<>(); + AutoDataLoaderOptions options = newOptions().setBatchingEnabled(false).setCachingEnabled(false); + DataLoader identityLoader = idLoader(options, loadCalls, dispatcher); + + CompletableFuture fa = identityLoader.load("A"); + CompletableFuture fb = identityLoader.load("B"); + + // caching is off + CompletableFuture fa1 = identityLoader.load("A"); + CompletableFuture fb1 = identityLoader.load("B"); + + List values = CompletableFutureKit.allOf(asList(fa, fb, fa1, fb1)).join(); + + assertThat(fa.join(), equalTo("A")); + assertThat(fb.join(), equalTo("B")); + assertThat(fa1.join(), equalTo("A")); + assertThat(fb1.join(), equalTo("B")); + + assertThat(values, equalTo(asList("A", "B", "A", "B"))); + + assertThat(loadCalls, equalTo(asList( + singletonList("A"), + singletonList("B"), + singletonList("A"), + singletonList("B") + ))); + + } + + @Test + public void batches_multiple_requests_with_max_batch_size() throws Exception { + List> loadCalls = new ArrayList<>(); + DataLoader identityLoader = idLoader(newOptions().setMaxBatchSize(2), loadCalls, dispatcher); + + CompletableFuture f1 = identityLoader.load(1); + CompletableFuture f2 = identityLoader.load(2); + CompletableFuture f3 = identityLoader.load(3); + +// identityLoader.dispatch(); + + await().until(allOf(f1, f2, f3)::isDone); + + assertThat(f1.join(), equalTo(1)); + assertThat(f2.join(), equalTo(2)); + assertThat(f3.join(), equalTo(3)); + + assertThat(loadCalls, equalTo(asList(asList(1, 2), singletonList(3)))); + + } + + @Test + public void can_split_max_batch_sizes_correctly() throws Exception { + List> loadCalls = new ArrayList<>(); + AutoDataLoader identityLoader = idLoader(newOptions().setMaxBatchSize(5), loadCalls, dispatcher); + + List> futures = new ArrayList<>(); + for (int i = 0; i < 21; i++) { + futures.add(identityLoader.load(i)); + } + List> expectedCalls = new ArrayList<>(); + expectedCalls.add(listFrom(0, 5)); + expectedCalls.add(listFrom(5, 10)); + expectedCalls.add(listFrom(10, 15)); + expectedCalls.add(listFrom(15, 20)); + expectedCalls.add(listFrom(20, 21)); + + await().until(allOf(futures.toArray(new CompletableFuture[futures.size()]))::isDone); + List result = identityLoader.dispatch().join(); + + assertThat(result, equalTo(listFrom(0, 21))); + assertThat(loadCalls, equalTo(expectedCalls)); + } + + @Test + @Ignore + public void can_split_max_batch_sizes_correctly_10_times () throws Exception { + for (int i = 0; i < 10; i++) { + can_split_max_batch_sizes_correctly(); + } + } + + @Test + public void should_Batch_loads_occurring_within_futures() { + List> loadCalls = new ArrayList<>(); + DataLoader identityLoader = idLoader(newOptions(), loadCalls, dispatcher); + + Supplier nullValue = () -> null; + + final AtomicBoolean v4Called = new AtomicBoolean(); + + CompletableFuture.supplyAsync(nullValue).thenAccept(v1 -> { + LOGGER.debug("before load({}) called", "a"); + identityLoader.load("a"); + LOGGER.debug("after load({}) called", "a"); + CompletableFuture.supplyAsync(nullValue).thenAccept(v2 -> { + LOGGER.debug("before load({}) called", "b"); + identityLoader.load("b"); + LOGGER.debug("after load({}) called", "b"); + CompletableFuture.supplyAsync(nullValue).thenAccept(v3 -> { + LOGGER.debug("before load({}) called", "c"); + identityLoader.load("c"); + LOGGER.debug("after load({}) called", "c"); + CompletableFuture.supplyAsync(nullValue).thenAccept( + v4 -> { + LOGGER.debug("before load({}) called", "d"); + identityLoader.load("d"); + LOGGER.debug("after load({}) called", "d"); + v4Called.set(true); + LOGGER.debug("v4Called = true"); + }); + }); + }); + }); + + LOGGER.debug("awaiting for v4Called == true ..."); + await().untilTrue(v4Called); + LOGGER.debug("woke up!"); + +// identityLoader.dispatchAndJoin(); + + assertThat(flatList(loadCalls), equalTo(asList("a", "b", "c", "d"))); +// assertThat(loadCalls, equalTo( +// singletonList(asList("a", "b", "c", "d")))); + } + + @Test + public void can_call_a_loader_from_a_loader() throws Exception { + Set deepLoadCalls = new HashSet<>(); + DataLoader deepLoader = new AutoDataLoader(keys -> { + deepLoadCalls.addAll(keys); + return CompletableFuture.completedFuture(keys); + }, dispatcher); + + List> aLoadCalls = new ArrayList<>(); + DataLoader aLoader = new AutoDataLoader<>(keys -> { + aLoadCalls.add(keys); + return deepLoader.loadMany(keys); + }, dispatcher); + + List> bLoadCalls = new ArrayList<>(); + DataLoader bLoader = new AutoDataLoader<>(keys -> { + bLoadCalls.add(keys); + return deepLoader.loadMany(keys); + }, dispatcher); + + CompletableFuture a1 = aLoader.load("A1"); + CompletableFuture a2 = aLoader.load("A2"); + CompletableFuture b1 = bLoader.load("B1"); + CompletableFuture b2 = bLoader.load("B2"); +// +// CompletableFuture.allOf( +// aLoader.dispatch(), +// deepLoader.dispatch(), +// bLoader.dispatch(), +// deepLoader.dispatch() +// ).join(); + await().until(allOf(a1, a2, b1, b2)::isDone); + + assertThat(a1.get(), equalTo("A1")); + assertThat(a2.get(), equalTo("A2")); + assertThat(b1.get(), equalTo("B1")); + assertThat(b2.get(), equalTo("B2")); + + assertThat(aLoadCalls, equalTo( + singletonList(asList("A1", "A2")))); + + assertThat(bLoadCalls, equalTo( + singletonList(asList("B1", "B2")))); + + assertThat(deepLoadCalls, equalTo(new HashSet<>(asList("A1", "A2", "B1", "B2")))); + } + + @Test + @Ignore + public void should_allow_composition_of_data_loader_calls() throws Exception { + UserManager userManager = new UserManager(); + + BatchLoader userBatchLoader = userIds -> CompletableFuture + .supplyAsync(() -> userIds + .stream() + .map(userManager::loadUserById) + .collect(Collectors.toList())); + AutoDataLoader userLoader = new AutoDataLoader<>(userBatchLoader, dispatcher); + + AtomicBoolean gandalfCalled = new AtomicBoolean(false); + AtomicBoolean sarumanCalled = new AtomicBoolean(false); + + CompletableFuture gandalf = userLoader.load(1L) + .thenAccept(user -> userLoader.load(user.getInvitedByID()) + .thenAccept(invitedBy -> { + gandalfCalled.set(true); + assertThat(invitedBy.getName(), equalTo("Manwë")); + })); + + CompletableFuture saruman = userLoader.load(2L) + .thenAccept(user -> userLoader.load(user.getInvitedByID()) + .thenAccept(invitedBy -> { + sarumanCalled.set(true); + assertThat(invitedBy.getName(), equalTo("Aulë")); + })); + + await().untilTrue(gandalfCalled); + await().untilTrue(sarumanCalled); + + List allResults = userLoader.dispatch().join(); + assertThat(allResults.size(), equalTo(4)); + } + + + private static CacheKey getJsonObjectCacheMapFn() { + return key -> key.stream() + .map(entry -> entry.getKey() + ":" + entry.getValue()) + .sorted() + .collect(Collectors.joining()); + } + + private static AutoDataLoader idLoader(AutoDataLoaderOptions options, List> loadCalls, Dispatcher dispatcher) { + return new AutoDataLoader<>(keys -> { + loadCalls.add(new ArrayList<>(keys)); + @SuppressWarnings("unchecked") + List values = keys.stream() + .map(k -> (V) k) + .collect(Collectors.toList()); + return CompletableFuture.completedFuture(values); + }, options.setDispatcher(dispatcher)); + } + + private static AutoDataLoader idLoaderBlowsUps( + AutoDataLoaderOptions options, List> loadCalls, Dispatcher dispatcher) { + return new AutoDataLoader<>(keys -> { + loadCalls.add(new ArrayList<>(keys)); + return TestKit.futureError(); + }, options.setDispatcher(dispatcher)); + } + + private static AutoDataLoader idLoaderAllExceptions( + AutoDataLoaderOptions options, List> loadCalls, Dispatcher dispatcher) { + return new AutoDataLoader<>(keys -> { + loadCalls.add(new ArrayList<>(keys)); + + List errors = keys.stream().map(k -> new IllegalStateException("Error")).collect(Collectors.toList()); + return CompletableFuture.completedFuture(errors); + }, options.setDispatcher(dispatcher)); + } + + private static AutoDataLoader idLoaderOddEvenExceptions( + AutoDataLoaderOptions options, List> loadCalls, Dispatcher dispatcher) { + return new AutoDataLoader<>(keys -> { + loadCalls.add(new ArrayList<>(keys)); + + List 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.setDispatcher(dispatcher)); + } + + + private BatchLoader keysAsValues() { + return CompletableFuture::completedFuture; + } + + private static AutoDataLoaderOptions newOptions () { + return new AutoDataLoaderOptions(); + } + + private static List flatList (List> list) { + return list + .stream() + .flatMap(Collection::stream) + .collect(Collectors.toList()); + } +} +