POC- Orchestration of DLs

src/main/java/org/dataloader/orchestration/CF.java

@@ -0,0 +1,157 @@
+package org.dataloader.orchestration;
+
+import java.util.Iterator;
+import java.util.concurrent.Executor;
+import java.util.concurrent.LinkedBlockingDeque;
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.function.BiFunction;
+import java.util.function.Supplier;
+
+public class CF<T> {
+
+    private AtomicReference<Object> result = new AtomicReference<>();
+
+    private static final Object NULL = new Object();
+
+    private static class ExceptionWrapper {
+        private final Throwable throwable;
+
+        private ExceptionWrapper(Throwable throwable) {
+            this.throwable = throwable;
+        }
+    }
+
+    private final LinkedBlockingDeque<CompleteAction<?, ?>> dependedActions = new LinkedBlockingDeque<>();
+
+    private static class CompleteAction<T, V> implements Runnable {
+        Executor executor;
+        CF<V> toComplete;
+        CF<T> src;
+        BiFunction<? super T, Throwable, ? extends V> mapperFn;
+
+        public CompleteAction(CF<V> toComplete, CF<T> src, BiFunction<? super T, Throwable, ? extends V> mapperFn, Executor executor) {
+            this.toComplete = toComplete;
+            this.src = src;
+            this.mapperFn = mapperFn;
+            this.executor = executor;
+        }
+
+        public void execute() {
+            if (executor != null) {
+                executor.execute(this);
+            } else {
+                toComplete.completeViaMapper(mapperFn, src.result.get());
+            }
+
+        }
+
+        @Override
+        public void run() {
+            toComplete.completeViaMapper(mapperFn, src.result.get());
+            src.dependedActions.remove(this);
+        }
+    }
+
+    private CF() {
+
+    }
+
+    public <U> CF<U> newIncomplete() {
+        return new CF<U>();
+    }
+
+    public static <T> CF<T> newComplete(T completed) {
+        CF<T> result = new CF<>();
+        result.encodeAndSetResult(completed);
+        return result;
+    }
+
+    public static <T> CF<T> newExceptionally(Throwable e) {
+        CF<T> result = new CF<>();
+        result.encodeAndSetResult(e);
+        return result;
+    }
+
+    public static <T> CF<T> supplyAsync(Supplier<T> supplier,
+                                        Executor executor) {
+
+        CF<T> result = new CF<>();
+        executor.execute(() -> {
+            try {
+                result.encodeAndSetResult(supplier.get());
+            } catch (Throwable ex) {
+                result.encodeAndSetResult(ex);
+            }
+        });
+        return result;
+    }
+
+
+    public <U> CF<U> map(BiFunction<? super T, Throwable, ? extends U> fn) {
+        CF<U> newResult = new CF<>();
+        dependedActions.push(new CompleteAction<>(newResult, this, fn, null));
+        return newResult;
+    }
+
+    public <U> CF<U> mapAsync(BiFunction<? super T, Throwable, ? extends U> fn, Executor executor) {
+        CF<U> newResult = new CF<>();
+        dependedActions.push(new CompleteAction<>(newResult, this, fn, executor));
+        return newResult;
+    }
+
+
+    public <U> CF<U> compose(BiFunction<? super T, Throwable, ? extends U> fn, Executor executor) {
+        CF<U> newResult = new CF<>();
+        dependedActions.push(new CompleteAction<>(newResult, this, fn, executor));
+        return newResult;
+    }
+
+
+    public boolean complete(T value) {
+        boolean success = result.compareAndSet(null, value);
+        fireDependentActions();
+        return success;
+    }
+
+    private boolean encodeAndSetResult(Object rawValue) {
+        if (rawValue == null) {
+            return result.compareAndSet(null, NULL);
+        } else if (rawValue instanceof Throwable) {
+            return result.compareAndSet(null, new ExceptionWrapper((Throwable) rawValue));
+        } else {
+            return result.compareAndSet(null, rawValue);
+        }
+    }
+
+    private Object decodeResult(Object rawValue) {
+        if (rawValue instanceof ExceptionWrapper) {
+            return ((ExceptionWrapper) rawValue).throwable;
+        } else if (rawValue == NULL) {
+            return null;
+        } else {
+            return rawValue;
+        }
+    }
+
+    private void fireDependentActions() {
+        Iterator<CompleteAction<?, ?>> iterator = dependedActions.iterator();
+        while (iterator.hasNext()) {
+            iterator.next().execute();
+        }
+    }
+
+    private <T, U> void completeViaMapper(BiFunction<? super T, Throwable, ? extends U> fn, Object encodedResult) {
+        try {
+            Object decodedResult = decodeResult(encodedResult);
+            Object mappedResult = fn.apply(
+                    (T) decodedResult,
+                    decodedResult instanceof Throwable ? (Throwable) decodedResult : null
+            );
+            this.result.compareAndSet(null, mappedResult);
+        } catch (Throwable t) {
+            encodeAndSetResult(t);
+        }
+    }
+
+
+}

src/main/java/org/dataloader/orchestration/ImmediateExecutor.java

@@ -0,0 +1,12 @@
+package org.dataloader.orchestration;
+
+import java.util.concurrent.Executor;
+
+class ImmediateExecutor implements Executor {
+    static final ImmediateExecutor INSTANCE = new ImmediateExecutor();
+
+    @Override
+    public void execute(Runnable command) {
+        command.run();
+    }
+}

src/main/java/org/dataloader/orchestration/ObservingExecutor.java

@@ -0,0 +1,23 @@
+package org.dataloader.orchestration;
+
+import java.util.concurrent.Executor;
+import java.util.function.Consumer;
+
+class ObservingExecutor<T> implements Executor {
+
+    private final Executor delegate;
+    private final T state;
+    private final Consumer<T> callback;
+
+    public ObservingExecutor(Executor delegate, T state, Consumer<T> callback) {
+        this.delegate = delegate;
+        this.state = state;
+        this.callback = callback;
+    }
+
+    @Override
+    public void execute(Runnable command) {
+        delegate.execute(command);
+        callback.accept(state);
+    }
+}

src/main/java/org/dataloader/orchestration/Orchestrator.java

@@ -0,0 +1,116 @@
+package org.dataloader.orchestration;
+
+import org.dataloader.DataLoader;
+import org.dataloader.impl.Assertions;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.Executor;
+import java.util.function.Function;
+
+public class Orchestrator<K, V> {
+
+    private final Executor executor;
+    private final Tracker tracker;
+    private final DataLoader<K, V> startingDL;
+    private final List<Step<?, ?>> steps = new ArrayList<>();
+
+    /**
+     * This will create a new {@link Orchestrator} that can allow multiple calls to multiple data-loader's
+     * to be orchestrated so they all run optimally.
+     *
+     * @param dataLoader the data loader to start with
+     * @param <K>        the key type
+     * @param <V>        the value type
+     * @return a new {@link Orchestrator}
+     */
+    public static <K, V> Orchestrator<K, V> orchestrate(DataLoader<K, V> dataLoader) {
+        return new Orchestrator<>(new Tracker(), dataLoader, ImmediateExecutor.INSTANCE);
+    }
+
+    // TODO - make this a builder
+    public static <K, V> Orchestrator<K, V> orchestrate(DataLoader<K, V> dataLoader, Executor executor) {
+        return new Orchestrator<>(new Tracker(), dataLoader, executor);
+    }
+
+    private Orchestrator(Tracker tracker, DataLoader<K, V> dataLoader, Executor executor) {
+        this.tracker = tracker;
+        this.startingDL = dataLoader;
+        this.executor = executor;
+    }
+
+    public Tracker getTracker() {
+        return tracker;
+    }
+
+    public Executor getExecutor() {
+        return executor;
+    }
+
+
+    public Step<K, V> load(K key) {
+        return load(key, null);
+    }
+
+    public Step<K, V> load(K key, Object keyContext) {
+        return Step.loadImpl(this, castAs(startingDL), key, keyContext);
+    }
+
+    static <T> T castAs(Object o) {
+        //noinspection unchecked
+        return (T) o;
+    }
+
+
+    <KT, VT> void record(Step<KT, VT> step) {
+        steps.add(step);
+        tracker.incrementStepCount();
+    }
+
+    /**
+     * This is the callback point saying to start the DataLoader loading process.
+     * <p>
+     * The type of object returned here depends on the value type of the last Step.  We cant be truly generic
+     * here and must be case.
+     *
+     * @param <VT> the value type
+     * @return the final composed value
+     */
+    <VT> CompletableFuture<VT> execute() {
+        Assertions.assertState(!steps.isEmpty(), () -> "How can the steps to run be empty??");
+        int index = 0;
+        Step<?, ?> firstStep = steps.get(index);
+
+        CompletableFuture<Object> currentCF = castAs(firstStep.codeToRun().apply(null)); // first load uses variable capture
+        whenComplete(index, firstStep, currentCF);
+
+        for (index++; index < steps.size(); index++) {
+            Step<?, ?> nextStep = steps.get(index);
+            Function<Object, CompletableFuture<?>> codeToRun = castAs(nextStep.codeToRun());
+            CompletableFuture<Object> nextCF = currentCF.thenCompose(value -> castAs(codeToRun.apply(value)));
+            currentCF = nextCF;
+
+            // side effect when this step is complete
+            whenComplete(index, nextStep, nextCF);
+        }
+        return castAs(currentCF);
+
+    }
+
+    private void whenComplete(int index, Step<?, ?> step, CompletableFuture<Object> cf) {
+        cf.whenComplete((v, throwable) -> {
+            getTracker().loadCallComplete(step.dataLoader());
+            // replace with instrumentation code
+            if (throwable != null) {
+                // TODO - should we be cancelling future steps here - no need for dispatch tracking if they will never run
+                System.out.println("A throwable has been thrown on step  " + index + ": " + throwable.getMessage());
+                throwable.printStackTrace(System.out);
+            } else {
+                System.out.println("step " + index + " returned : " + v);
+            }
+        });
+    }
+
+
+}