Added TestWorkflowService that implements Cadence service API. (#126)

* Minor comment fixes

* Added SelfAdvancingTimer

* All WorkflowTest tests are passing with both docker and in process service

Author: mfateev

build.gradle

@@ -43,6 +43,7 @@ dependencies {
     compile(group: 'org.apache.thrift', name: 'libthrift', version: '0.9.3')
     compile group: 'com.google.code.gson', name: 'gson', version: '2.8.2'
     testCompile group: 'junit', name: 'junit', version: '4.12'
+    testCompile group: 'com.googlecode.junit-toolbox', name: 'junit-toolbox', version: '2.4'
     testCompile group: 'ch.qos.logback', name: 'logback-classic', version: '1.2.3'
 }
 

src/main/java/com/uber/cadence/client/WorkflowException.java

@@ -31,11 +31,25 @@ public class WorkflowException extends RuntimeException {
 
   protected WorkflowException(
       String message, WorkflowExecution execution, String workflowType, Throwable cause) {
-    super(message, cause);
+    super(getMessage(message, execution, workflowType), cause);
     this.execution = execution;
     this.workflowType = workflowType;
   }
 
+  private static String getMessage(
+      String message, WorkflowExecution execution, String workflowType) {
+    StringBuilder result = new StringBuilder();
+    result.append(message);
+    if (workflowType != null) {
+      result.append(", WorkflowType=\"");
+      result.append(workflowType);
+      result.append("\"");
+    }
+    result.append(", WorkflowExecution=\"");
+    result.append(execution);
+    return result.toString();
+  }
+
   public WorkflowExecution getExecution() {
     return execution;
   }

src/main/java/com/uber/cadence/internal/replay/ReplayDecisionTaskHandler.java

@@ -25,6 +25,7 @@
 import com.uber.cadence.RespondQueryTaskCompletedRequest;
 import com.uber.cadence.WorkflowExecution;
 import com.uber.cadence.WorkflowType;
+import com.uber.cadence.internal.common.WorkflowExecutionUtils;
 import com.uber.cadence.internal.worker.DecisionTaskHandler;
 import com.uber.cadence.internal.worker.DecisionTaskWithHistoryIterator;
 import java.io.PrintWriter;
@@ -111,7 +112,18 @@ private Result handleDecisionTaskImpl(DecisionTaskWithHistoryIterator decisionTa
       DecisionsHelper decisionsHelper = decider.getDecisionsHelper();
       List<Decision> decisions = decisionsHelper.getDecisions();
       byte[] context = decisionsHelper.getWorkflowContextDataToReturn();
-      if (log.isDebugEnabled()) {
+      if (log.isTraceEnabled()) {
+        WorkflowExecution execution = decisionTask.getWorkflowExecution();
+        log.trace(
+            "WorkflowTask startedEventId="
+                + decisionTask.getStartedEventId()
+                + ", WorkflowID="
+                + execution.getWorkflowId()
+                + ", RunID="
+                + execution.getRunId()
+                + " completed with "
+                + WorkflowExecutionUtils.prettyPrintDecisions(decisions));
+      } else if (log.isDebugEnabled()) {
         WorkflowExecution execution = decisionTask.getWorkflowExecution();
         log.debug(
             "WorkflowTask startedEventId="

src/main/java/com/uber/cadence/internal/sync/DeterministicRunnerImpl.java

@@ -150,7 +150,7 @@ public void runUntilAllBlocked() throws Throwable {
               new WorkflowThreadImpl(
                   false, threadPool, this, nr.name, false, runnerCancellationScope, nr.runnable);
           // It is important to prepend threads as there are callbacks
-          // like signals that have to apply before any other threads.
+          // like signals that have to run before any other threads.
           // Otherwise signal might be never processed if it was received
           // after workflow decided to close.
           threads.addFirst(thread);

src/main/java/com/uber/cadence/internal/sync/SyncDecisionContext.java

@@ -35,6 +35,7 @@
 import com.uber.cadence.workflow.ActivityFailureException;
 import com.uber.cadence.workflow.ActivityTimeoutException;
 import com.uber.cadence.workflow.CancellationScope;
+import com.uber.cadence.workflow.ChildWorkflowException;
 import com.uber.cadence.workflow.ChildWorkflowFailureException;
 import com.uber.cadence.workflow.ChildWorkflowOptions;
 import com.uber.cadence.workflow.CompletablePromise;
@@ -49,7 +50,7 @@
 import java.util.concurrent.TimeUnit;
 import java.util.function.Consumer;
 
-class SyncDecisionContext implements ActivityExecutor {
+final class SyncDecisionContext implements ActivityExecutor {
   private final DecisionContext context;
   private DeterministicRunner runner;
   private final DataConverter converter;
@@ -237,7 +238,9 @@ private RuntimeException mapChildWorkflowException(Exception failure) {
     if (failure instanceof CancellationException) {
       return (CancellationException) failure;
     }
-
+    if (failure instanceof ChildWorkflowException) {
+      throw (ChildWorkflowException) failure;
+    }
     if (!(failure instanceof ChildWorkflowTaskFailedException)) {
       throw new IllegalArgumentException("Unexpected exception type: ", failure);
     }

src/main/java/com/uber/cadence/internal/sync/TestActivityEnvironmentInternal.java

@@ -165,7 +165,7 @@ public <T> Promise<T> executeActivity(
               .setWorkflowId("test-workflow-id")
               .setRunId(UUID.randomUUID().toString()));
       task.setActivityType(new ActivityType().setName(activityType));
-      IWorkflowService service = new WorkflowServiceWrapper(testEnvironmentOptions.getService());
+      IWorkflowService service = new WorkflowServiceWrapper(null);
       Result taskResult =
           activityTaskHandler.handle(service, testEnvironmentOptions.getDomain(), task);
       return Workflow.newPromise(getReply(task, taskResult, returnType));

src/main/java/com/uber/cadence/internal/sync/TestWorkflowEnvironmentInternal.java

@@ -117,6 +117,16 @@ public WorkflowClient newWorkflowClient() {
     return WorkflowClientInternal.newInstance(service, testEnvironmentOptions.getDomain(), options);
   }
 
+  @Override
+  public long currentTimeMillis() {
+    return service.currentTimeMillis();
+  }
+
+  @Override
+  public void registerDelayedCallback(Duration delay, Runnable r) {
+    service.registerDelayedCallback(delay, r);
+  }
+
   public IWorkflowService getWorkflowService() {
     return service;
   }
@@ -142,6 +152,10 @@ private WorkflowServiceWrapper() {
       impl = new TestWorkflowService();
     }
 
+    public long currentTimeMillis() {
+      return impl.currentTimeMillis();
+    }
+
     @Override
     public RecordActivityTaskHeartbeatResponse RecordActivityTaskHeartbeat(
         RecordActivityTaskHeartbeatRequest heartbeatRequest)
@@ -500,5 +514,9 @@ public void getDiagnostics(StringBuilder result) {
     public void close() {
       impl.close();
     }
+
+    public void registerDelayedCallback(Duration delay, Runnable r) {
+      impl.registerDelayedCallback(delay, r);
+    }
   }
 }

src/main/java/com/uber/cadence/internal/sync/WorkflowThreadImpl.java

@@ -161,7 +161,6 @@ public void setName(String name) {
     if (name == null) {
       name = "workflow-" + super.hashCode();
     }
-    log.trace(String.format("Workflow thread \"%s\" created", name));
     this.task = new RunnableWrapper(context, name, detached, parentCancellationScope, runnable);
   }
 
@@ -172,14 +171,11 @@ public boolean isDetached() {
 
   @Override
   public void cancel() {
-    log.trace(String.format("Workflow thread \"%s\" cancel called", getName()));
     task.cancellationScope.cancel();
   }
 
   @Override
   public void cancel(String reason) {
-    log.trace(
-        String.format("Workflow thread \"%s cancel called with \"%s\" reason", getName(), reason));
     task.cancellationScope.cancel(reason);
   }
 
@@ -203,7 +199,6 @@ public void start() {
     if (context.getStatus() != Status.CREATED) {
       throw new IllegalThreadStateException("already started");
     }
-    log.trace(String.format("Workflow thread \"%s\" started", getName()));
     context.setStatus(Status.RUNNING);
     try {
       taskFuture = threadPool.submit(task);

src/main/java/com/uber/cadence/internal/testservice/RequestContext.java

@@ -73,6 +73,9 @@ Runnable getCallback() {
   private final List<Timer> timers = new ArrayList<>();
   private long workflowCompletedAtEventId = -1;
   private boolean needDecision;
+  // How many times call SelfAdvancedTimer#lockTimeSkipping.
+  // Negative means how many times to call SelfAdvancedTimer#unlockTimeSkipping.
+  private int timerLocks;
 
   /**
    * Creates an instance of the RequestContext
@@ -95,6 +98,18 @@ void add(RequestContext ctx) {
     this.events.addAll(ctx.getEvents());
   }
 
+  void lockTimer() {
+    timerLocks++;
+  }
+
+  void unlockTimer() {
+    timerLocks--;
+  }
+
+  public int getTimerLocks() {
+    return timerLocks;
+  }
+
   long currentTimeInNanoseconds() {
     return clock.getAsLong() * NANOS_PER_MILLIS;
   }
@@ -193,4 +208,8 @@ void fireCallbacks() throws InternalServiceError {
   ExecutionId getExecutionId() {
     return executionId;
   }
+
+  public boolean isEmpty() {
+    return events.isEmpty() && activityTasks.isEmpty() && decisionTask == null && timers.isEmpty();
+  }
 }

src/main/java/com/uber/cadence/internal/testservice/SelfAdvancingTimer.java

@@ -0,0 +1,55 @@
+/*
+ *  Copyright 2012-2016 Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ *
+ *  Modifications copyright (C) 2017 Uber Technologies, Inc.
+ *
+ *  Licensed under the Apache License, Version 2.0 (the "License"). You may not
+ *  use this file except in compliance with the License. A copy of the License is
+ *  located at
+ *
+ *  http://aws.amazon.com/apache2.0
+ *
+ *  or in the "license" file accompanying this file. This file 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.uber.cadence.internal.testservice;
+
+import java.time.Duration;
+import java.util.function.LongSupplier;
+
+/**
+ * Timer service that automatically forwards current time to the next task time when is not locked
+ * through {@link #lockTimeSkipping()}.
+ */
+public interface SelfAdvancingTimer {
+
+  /**
+   * Schedule a task with a specified delay. The actual wait time is defined by the internal clock
+   * that might advance much faster than the wall clock.
+   */
+  void schedule(Duration delay, Runnable task);
+
+  /** Supplier that returns current time of the timer when called. */
+  LongSupplier getClock();
+
+  /**
+   * Prohibit automatic time skipping until {@link #unlockTimeSkipping()} is called. Locks and
+   * unlocks are counted.
+   */
+  void lockTimeSkipping();
+
+  void unlockTimeSkipping();
+
+  /**
+   * Update lock count. The same as calling lockTimeSkipping count number of times for positive
+   * count and unlockTimeSkipping for negative count.
+   *
+   * @param count
+   */
+  void updateLocks(int count);
+
+  void shutdown();
+}