POC- Orchestration of DLs - tried and executor that can observe

Author: bbakerman

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

@@ -6,10 +6,12 @@
 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<>();
@@ -24,16 +26,26 @@ public class Orchestrator<K, V> {
      * @return a new {@link Orchestrator}
      */
     public static <K, V> Orchestrator<K, V> orchestrate(DataLoader<K, V> dataLoader) {
-        return new Orchestrator<>(new Tracker(), dataLoader);
+        return new Orchestrator<>(new Tracker(), dataLoader, ImmediateExecutor.INSTANCE);
     }
 
-    public Tracker getTracker() {
-        return tracker;
+    // 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) {
+    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;
     }
 
 

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

@@ -3,7 +3,10 @@
 import org.dataloader.DataLoader;
 
 import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.Executor;
+import java.util.function.Consumer;
 import java.util.function.Function;
+import java.util.function.Supplier;
 
 import static org.dataloader.orchestration.Orchestrator.castAs;
 
@@ -35,7 +38,11 @@ public Step<K, V> load(K key, Object keyContext) {
     }
 
     public Step<V, V> thenLoad(Function<V, K> codeToRun) {
-        return thenLoadImpl(orchestrator, dl, codeToRun);
+        return thenLoadImpl(orchestrator, dl, codeToRun, false);
+    }
+
+    public Step<V, V> thenLoadAsync(Function<V, K> codeToRun) {
+        return thenLoadImpl(orchestrator, dl, codeToRun, true);
     }
 
     static <K, V> Step<K, V> loadImpl(Orchestrator<?, ?> orchestrator, DataLoader<Object, Object> dl, K key, Object keyContext) {
@@ -49,16 +56,42 @@ static <K, V> Step<K, V> loadImpl(Orchestrator<?, ?> orchestrator, DataLoader<Ob
         return step;
     }
 
-    static <K, V> Step<V, V> thenLoadImpl(Orchestrator<?, ?> orchestrator, DataLoader<Object, Object> dl, Function<V, K> codeToRun) {
-        Function<V, CompletableFuture<V>> actualCodeToRun = v -> {
+    static <K, V> Step<V, V> thenLoadImpl(Orchestrator<?, ?> orchestrator, DataLoader<Object, Object> dl, Function<V, K> codeToRun, boolean async) {
+        Tracker tracker = orchestrator.getTracker();
+        Function<V, CompletableFuture<V>> actualCodeToRun;
+        if (async) {
+            actualCodeToRun = mkAsyncLoadLambda(orchestrator, dl, codeToRun, tracker);
+        } else {
+            actualCodeToRun = mkSyncLoadLambda(dl, codeToRun, tracker);
+        }
+        Step<V, V> step = new Step<>(orchestrator, dl, actualCodeToRun);
+        orchestrator.record(step);
+        return step;
+    }
+
+    private static <K, V> Function<V, CompletableFuture<V>> mkSyncLoadLambda(DataLoader<Object, Object> dl, Function<V, K> codeToRun, Tracker tracker) {
+        return v -> {
             K key = codeToRun.apply(v);
             CompletableFuture<V> cf = castAs(dl.load(key));
-            orchestrator.getTracker().loadCall(dl);
+            tracker.loadCall(dl);
             return cf;
         };
-        Step<V, V> step = new Step<>(orchestrator, dl, actualCodeToRun);
-        orchestrator.record(step);
-        return step;
+    }
+
+    private static <K, V> Function<V, CompletableFuture<V>> mkAsyncLoadLambda(Orchestrator<?, ?> orchestrator, DataLoader<Object, Object> dl, Function<V, K> codeToRun, Tracker tracker) {
+        return v -> {
+            Executor executor = orchestrator.getExecutor();
+            Consumer<String> callback = atSomePointWeNeedMoreStateButUsingStringForNowToMakeItCompile -> {
+                tracker.loadCall(dl);
+            };
+            ObservingExecutor<String> observingExecutor = new ObservingExecutor<>(executor, "state", callback);
+            Supplier<CompletableFuture<V>> dataLoaderCall = () -> {
+                K key = codeToRun.apply(v);
+                return castAs(dl.load(key));
+            };
+            return CompletableFuture.supplyAsync(dataLoaderCall, observingExecutor)
+                    .thenCompose(Function.identity());
+        };
     }
 
     public CompletableFuture<V> toCompletableFuture() {

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

@@ -31,6 +31,10 @@ public Step<K, V> load(K key, Object keyContext) {
     }
 
     public Step<V, V> thenLoad(Function<V, K> codeToRun) {
-        return Step.thenLoadImpl(orchestrator, castAs(dl), codeToRun);
+        return Step.thenLoadImpl(orchestrator, castAs(dl), codeToRun, false);
+    }
+
+    public Step<V, V> thenLoadAsync(Function<V, K> codeToRun) {
+        return Step.thenLoadImpl(orchestrator, castAs(dl), codeToRun, true);
     }
 }

src/test/java/org/dataloader/fixtures/TestKit.java

@@ -72,6 +72,10 @@ public static BatchLoader<String, String> reverseBatchLoader() {
         return keys -> CompletableFuture.completedFuture(keys.stream().map(TestKit::reverse).collect(toList()));
     }
 
+    public static BatchLoader<String, String> alternateCaseBatchLoader() {
+        return keys -> CompletableFuture.completedFuture(keys.stream().map(TestKit::alternateCase).collect(toList()));
+    }
+
     public static String reverse(String s) {
         StringBuilder sb = new StringBuilder();
         for (int i = s.length() - 1; i >= 0; i--) {
@@ -80,6 +84,19 @@ public static String reverse(String s) {
         return sb.toString();
     }
 
+    public static String alternateCase(String s) {
+        StringBuilder sb = new StringBuilder();
+        for (int i = 0; i <= s.length()-1; i++) {
+            char c = s.charAt(i);
+            if (i % 2 == 0) {
+                sb.append(Character.toLowerCase(c));
+            } else {
+                sb.append(Character.toUpperCase(c));
+            }
+        }
+        return sb.toString();
+    }
+
     public static <K, V> DataLoader<K, V> idLoader() {
         return idLoader(null, new ArrayList<>());
     }

src/test/java/org/dataloader/orchestration/OrchestratorTest.java

@@ -3,12 +3,18 @@
 import org.dataloader.DataLoader;
 import org.dataloader.DataLoaderOptions;
 import org.dataloader.DataLoaderRegistry;
+import org.dataloader.fixtures.TestKit;
 import org.junit.jupiter.api.Test;
 
 import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.Executor;
+import java.util.concurrent.ForkJoinPool;
+import java.util.concurrent.Semaphore;
 
 import static org.awaitility.Awaitility.await;
 import static org.dataloader.DataLoaderFactory.newDataLoader;
+import static org.dataloader.fixtures.TestKit.alternateCaseBatchLoader;
 import static org.dataloader.fixtures.TestKit.lowerCaseBatchLoader;
 import static org.dataloader.fixtures.TestKit.reverseBatchLoader;
 import static org.dataloader.fixtures.TestKit.upperCaseBatchLoader;
@@ -22,6 +28,7 @@ class OrchestratorTest {
     DataLoader<String, String> dlUpper = newDataLoader(upperCaseBatchLoader(), cachingAndBatchingOptions);
     DataLoader<String, String> dlLower = newDataLoader(lowerCaseBatchLoader(), cachingAndBatchingOptions);
     DataLoader<String, String> dlReverse = newDataLoader(reverseBatchLoader(), cachingAndBatchingOptions);
+    DataLoader<String, String> dlAlternateCase = newDataLoader(alternateCaseBatchLoader(), cachingAndBatchingOptions);
 
     @Test
     void canOrchestrate() {
@@ -30,6 +37,7 @@ void canOrchestrate() {
                 .register("upper", dlUpper)
                 .register("lower", dlLower)
                 .register("reverse", dlReverse)
+                .register("alternateCase", dlAlternateCase)
                 .build();
 
         Orchestrator<String, String> orchestrator = Orchestrator.orchestrate(dlUpper);
@@ -56,22 +64,25 @@ void canOrchestrateWhenNotInPerfectOrder() {
                 .register("reverse", dlReverse)
                 .register("lower", dlLower)
                 .register("upper", dlUpper)
+                .register("alternateCase", dlAlternateCase)
                 .build();
 
-        Orchestrator<String, String> orchestrator = Orchestrator.orchestrate(dlUpper);
+        ForkJoinPool forkJoinPool = ForkJoinPool.commonPool();
+        Orchestrator<String, String> orchestrator = Orchestrator.orchestrate(dlUpper, forkJoinPool);
         CompletableFuture<String> cf = orchestrator.load("aBc", null)
                 .with(dlLower).thenLoad(key1 -> key1)
                 .with(dlReverse).thenLoad(key -> key)
+                .with(dlAlternateCase).thenLoadAsync(key -> key)
                 .toCompletableFuture();
 
-        registry.dispatchAll();
-
-        assertThat(cf.isDone(), equalTo(false));
-
-        assertThat(orchestrator.getTracker().getOutstandingLoadCount(),equalTo(2));
+        for (int i = 0; i < 10; i++) {
+            TestKit.snooze(50); // TODO - hack or now
+            registry.dispatchAll();
+            System.out.println("Waiting for " + i + " to complete...");
+        }
 
         await().until(cf::isDone);
 
-        assertThat(cf.join(), equalTo("cba"));
+        assertThat(cf.join(), equalTo("cBa"));
     }
 }