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/Orchestrator.java

@@ -0,0 +1,142 @@
+package org.dataloader.orchestration;
+
+import org.dataloader.DataLoader;
+import org.dataloader.impl.Assertions;
+import org.dataloader.orchestration.executors.ImmediateExecutor;
+import org.dataloader.orchestration.observation.Tracker;
+import org.dataloader.orchestration.observation.TrackingObserver;
+
+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> Builder<K, V> orchestrate(DataLoader<K, V> dataLoader) {
+        return new Builder<>(dataLoader);
+    }
+
+    public Orchestrator(Builder<K, V> builder) {
+        this.tracker = new Tracker(builder.trackingObserver);
+        this.executor = builder.executor;
+        this.startingDL = builder.dataLoader;
+    }
+
+    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??");
+
+        // tell the tracker we are under way
+        getTracker().startingExecution();
+
+        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 stepIndex, Step<?, ?> step, CompletableFuture<Object> cf) {
+        cf.whenComplete((v, throwable) -> {
+            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  " + stepIndex + ": " + throwable.getMessage());
+                throwable.printStackTrace(System.out);
+            } else {
+                System.out.println("step " + stepIndex + " returned : " + v);
+            }
+            getTracker().loadCallComplete(stepIndex, step.dataLoader(), throwable);
+        });
+    }
+
+
+    public static class Builder<K, V> {
+        private Executor executor = ImmediateExecutor.INSTANCE;
+        private DataLoader<K, V> dataLoader;
+        private TrackingObserver trackingObserver;
+
+        Builder(DataLoader<K, V> dataLoader) {
+            this.dataLoader = dataLoader;
+        }
+
+        public Builder<K, V> executor(Executor executor) {
+            this.executor = executor;
+            return this;
+        }
+
+        public Builder<K, V> observer(TrackingObserver trackingObserver) {
+            this.trackingObserver = trackingObserver;
+            return this;
+        }
+
+        public Orchestrator<K, V> build() {
+            return new Orchestrator<>(this);
+        }
+    }
+
+}