Overriding context

flo-runner/src/main/java/com/spotify/flo/context/FloRunner.java

@@ -131,7 +131,11 @@ private EvalContext createContext() {
         ? persist(instrumentedContext)
         : instrumentedContext;
 
-    return MemoizingContext.composeWith(LoggingContext.composeWith(baseContext, logging));
+    return
+        MemoizingContext.composeWith(
+            OverridingContext.composeWith(
+                LoggingContext.composeWith(
+                    baseContext, logging)));
   }
 
   private EvalContext createRootContext() {

flo-scala/src/main/scala/com/spotify/flo/package.scala

@@ -26,7 +26,6 @@ package object flo {
   }
 
   def $[T]: TaskBuilder0[T] = currentBuilder
-  def ▫[T]: TaskBuilder0[T] = currentBuilder
 
   private def currentBuilder[T]: TaskBuilder0[T] = {
     val builder = dynamicBuilder.value

flo-workflow/src/main/java/com/spotify/flo/TaskContextStrict.java

@@ -20,6 +20,8 @@
 
 package com.spotify.flo;
 
+import java.util.Optional;
+
 /**
  * This is to be extended when building a {@link TaskContext} that is intended to work only with a
  * task whose returned value is of type {@link S}.
@@ -39,4 +41,15 @@ final public void onSuccess(Task<?> task, Object z) {
    */
   public void onSuccessStrict(Task<?> task, S z) {
   }
+
+  /**
+   * Perform a lookup of the value this task would have produced if it ran. Override to be able
+   * to short-circuit a task with a previously calculated value.
+   *
+   * @param task a task with a {@link TaskContextStrict} to lookup a value for
+   * @return the value for this {@link TaskContextStrict} (e.g. from a previous run)
+   */
+  public Optional<S> lookup(Task<S> task) {
+    return Optional.empty();
+  }
 }

flo-workflow/src/main/java/com/spotify/flo/context/OverridingContext.java

@@ -0,0 +1,72 @@
+/*-
+ * -\-\-
+ * Flo Workflow Definition
+ * --
+ * Copyright (C) 2016 - 2017 Spotify AB
+ * --
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ * 
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ * 
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ * -/-/-
+ */
+
+package com.spotify.flo.context;
+
+import static com.spotify.flo.Util.colored;
+
+import com.spotify.flo.EvalContext;
+import com.spotify.flo.Task;
+import com.spotify.flo.TaskContextStrict;
+import java.util.Optional;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * An {@link EvalContext} that may return a value without calling the processFn or evaluating its
+ * dependencies.
+ */
+public class OverridingContext extends ForwardingEvalContext {
+  private static final Logger LOG = LoggerFactory.getLogger(OverridingContext.class);
+
+  private OverridingContext(EvalContext delegate) {
+    super(delegate);
+  }
+
+  public static EvalContext composeWith(EvalContext baseContext) {
+    return new OverridingContext(baseContext);
+  }
+
+  @Override
+  public <T> Value<T> evaluateInternal(Task<T> task, EvalContext context) {
+    final Optional<TaskContextStrict<?, T>> taskContextStrict =
+        task.contexts().stream()
+            .filter(c -> c instanceof TaskContextStrict)
+            .<TaskContextStrict<?, T>>map(c -> (TaskContextStrict<?, T>) c)
+            .findFirst();
+
+    if (taskContextStrict.isPresent()) {
+      return context.value(() -> taskContextStrict.get().lookup(task))
+          .flatMap(value -> {
+            if (value.isPresent()) {
+              final T t = value.get();
+              LOG.debug("Not expanding {}, lookup = {}", colored(task.id()), t);
+              return context.immediateValue(t);
+            } else {
+              LOG.debug("Expanding {}", colored(task.id()));
+              return delegate.evaluateInternal(task, context);
+            }
+          });
+    } else {
+      LOG.debug("Expanding {}", colored(task.id()));
+      return delegate.evaluateInternal(task, context);
+    }
+  }
+}

flo-workflow/src/test/java/com/spotify/flo/context/OverridingContextTest.java

@@ -0,0 +1,134 @@
+/*-
+ * -\-\-
+ * Flo Workflow Definition
+ * --
+ * Copyright (C) 2016 - 2017 Spotify AB
+ * --
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ * 
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ * 
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ * -/-/-
+ */
+
+package com.spotify.flo.context;
+
+import static com.spotify.flo.EvalContext.sync;
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.is;
+
+import com.spotify.flo.AwaitValue;
+import com.spotify.flo.EvalContext;
+import com.spotify.flo.Task;
+import com.spotify.flo.TaskContextStrict;
+import com.spotify.flo.TaskId;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Optional;
+import org.junit.Before;
+import org.junit.Test;
+
+public class OverridingContextTest {
+
+  EvalContext context = OverridingContext.composeWith(sync());
+
+  Map<TaskId, Integer> lookup = new HashMap<>();
+
+  TaskContextStrict<String, Integer> taskContext = new TaskContextStrict<String, Integer>() {
+
+    @Override
+    public String provide(EvalContext evalContext) {
+      return null;
+    }
+
+    @Override
+    public Optional<Integer> lookup(Task<Integer> task) {
+      return Optional.ofNullable(lookup.get(task.id()));
+    }
+  };
+
+  int countUpstreamRuns = 0;
+  int countRootRuns = 0;
+
+  @Before
+  public void setUp() {
+    lookup.clear();
+  }
+
+  @Test
+  public void shouldRunEverythingIfLookupsNotFound() {
+    final Task<Integer> task = rootTaskWithUpstreams(
+        upstreamCharCount("1"),
+        upstreamCharCount("22")
+    );
+
+    context.evaluate(task);
+
+    assertThat(countUpstreamRuns, is(2));
+    assertThat(countRootRuns, is(1));
+  }
+
+  @Test
+  public void shouldSkipOnExistingLookup() throws InterruptedException {
+    final Task<Integer> task = rootTaskWithUpstreams(
+        upstreamCharCount("1"),
+        upstreamCharCount("22")
+    );
+
+    lookup.put(task.id(), 3);
+
+    final AwaitValue<Integer> val = new AwaitValue<>();
+    context.evaluate(task).consume(val);
+    final Integer value = val.awaitAndGet();
+
+    assertThat(value, is(3));
+    assertThat(countUpstreamRuns, is(0));
+    assertThat(countRootRuns, is(0));
+  }
+
+  @Test
+  public void shouldSkipUpstreamOnExistingLookup() throws InterruptedException {
+    final Task<Integer> upstream = upstreamCharCount("1");
+    final Task<Integer> task = rootTaskWithUpstreams(
+        upstream,
+        upstreamCharCount("22")
+    );
+
+    lookup.put(upstream.id(), 1);
+
+    final AwaitValue<Integer> val = new AwaitValue<>();
+    context.evaluate(task).consume(val);
+    final Integer value = val.awaitAndGet();
+
+    assertThat(value, is(3));
+    assertThat(countUpstreamRuns, is(1));
+    assertThat(countRootRuns, is(1));
+  }
+
+  Task<Integer> rootTaskWithUpstreams(Task<Integer>... upstreams) {
+    return Task.named("rootTask", "foo").ofType(Integer.class)
+        .context(taskContext)
+        .inputs(() -> Arrays.asList(upstreams))
+        .process((context, inputs) -> {
+          countRootRuns++;
+          return inputs.stream().mapToInt(i -> i).sum();
+        });
+  }
+
+  Task<Integer> upstreamCharCount(String s) {
+    return Task.named("charCount", s).ofType(Integer.class)
+        .context(taskContext)
+        .process((context) -> {
+          countUpstreamRuns++;
+          return s.length();
+        });
+  }
+}