Added WorkflowInterceptor (#137)

* Switched continueAsNew to ContinueAsNewOptions

* Added Workflow.continueAsNew

* Updated version in README

* Added WorkflowInterceptor interface

* Migrated SyncDecisionContext to implement WorkflowInterceptor

* Fully migrated WorkflowInternal to use WorkflowInterceptor interface

* Threaded WorkflowInterceptor from Worker options

* Renamed WorkflowInterceptor.signalWorkflow to signalExternalWorkflow. Added tracing intercetor

* Added unit testing of the interceptor

Author: mfateev

README.md

@@ -29,7 +29,7 @@ Add *cadence-client* as a dependency to your *pom.xml*:
     <dependency>
       <groupId>com.uber</groupId>
       <artifactId>cadence-client</artifactId>
-      <version>0.1.0</version>
+      <version>0.2.0-SPANPSHOT</version>
     </dependency>
 
 # Overview

src/main/java/com/uber/cadence/activity/ActivityOptions.java

@@ -17,7 +17,7 @@
 
 package com.uber.cadence.activity;
 
-import static com.uber.cadence.internal.common.InternalUtils.roundUpToSeconds;
+import static com.uber.cadence.internal.common.OptionsUtils.roundUpToSeconds;
 
 import com.uber.cadence.common.RetryOptions;
 import java.time.Duration;

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

@@ -17,7 +17,7 @@
 
 package com.uber.cadence.client;
 
-import static com.uber.cadence.internal.common.InternalUtils.roundUpToSeconds;
+import static com.uber.cadence.internal.common.OptionsUtils.roundUpToSeconds;
 
 import com.uber.cadence.ChildPolicy;
 import com.uber.cadence.WorkflowIdReusePolicy;

src/main/java/com/uber/cadence/internal/common/InternalUtils.java

@@ -19,13 +19,10 @@
 
 import com.uber.cadence.workflow.WorkflowMethod;
 import java.lang.reflect.Method;
-import java.time.Duration;
 
 /** Utility functions shared by the implementation code. */
 public final class InternalUtils {
 
-  public static final float SECOND = 1000f;
-
   /**
    * Used to construct default name of an activity or workflow type from a method it implements.
    *
@@ -67,29 +64,6 @@ public static Method getWorkflowMethod(Class<?> workflowInterface) {
     return result;
   }
 
-  /**
-   * Convert milliseconds to seconds rounding up. Used by timers to ensure that they never fire
-   * earlier than requested.
-   */
-  public static Duration roundUpToSeconds(Duration duration, Duration defaultValue) {
-    if (duration == null) {
-      return defaultValue;
-    }
-    return roundUpToSeconds(duration);
-  }
-
-  /**
-   * Round durations to seconds rounding up. As all timeouts and timers resolution is in seconds
-   * ensures that nothing times out or fires before the requested time.
-   */
-  public static Duration roundUpToSeconds(Duration duration) {
-    if (duration == null) {
-      return Duration.ZERO;
-    }
-    Duration result = Duration.ofMillis((long) (Math.ceil(duration.toMillis() / SECOND) * SECOND));
-    return result;
-  }
-
   /** Prohibit instantiation */
   private InternalUtils() {}
 }

src/main/java/com/uber/cadence/internal/common/OptionsUtils.java

@@ -23,6 +23,7 @@
 public final class OptionsUtils {
 
   public static final Duration DEFAULT_TASK_START_TO_CLOSE_TIMEOUT = Duration.ofSeconds(10);
+  public static final float SECOND = 1000f;
 
   /** Merges value from annotation and option. Option value takes precedence. */
   public static <G> G merge(G annotation, G options, Class<G> type) {
@@ -51,6 +52,29 @@ public static Duration merge(long aSeconds, Duration o) {
     return aSeconds == 0 ? null : Duration.ofSeconds(aSeconds);
   }
 
+  /**
+   * Convert milliseconds to seconds rounding up. Used by timers to ensure that they never fire
+   * earlier than requested.
+   */
+  public static Duration roundUpToSeconds(Duration duration, Duration defaultValue) {
+    if (duration == null) {
+      return defaultValue;
+    }
+    return roundUpToSeconds(duration);
+  }
+
+  /**
+   * Round durations to seconds rounding up. As all timeouts and timers resolution is in seconds
+   * ensures that nothing times out or fires before the requested time.
+   */
+  public static Duration roundUpToSeconds(Duration duration) {
+    if (duration == null) {
+      return Duration.ZERO;
+    }
+    Duration result = Duration.ofMillis((long) (Math.ceil(duration.toMillis() / SECOND) * SECOND));
+    return result;
+  }
+
   /** Prohibits instantiation. */
   private OptionsUtils() {}
 }

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

@@ -27,8 +27,15 @@ public final class ContinueAsNewWorkflowExecutionParameters {
   private String taskList;
   private int taskStartToCloseTimeoutSeconds;
   private ChildPolicy childPolicy;
+  private String workflowType;
 
-  public ContinueAsNewWorkflowExecutionParameters() {}
+  public void setWorkflowType(String workflowType) {
+    this.workflowType = workflowType;
+  }
+
+  public String getWorkflowType() {
+    return workflowType;
+  }
 
   public int getExecutionStartToCloseTimeoutSeconds() {
     return executionStartToCloseTimeoutSeconds;

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

@@ -19,6 +19,7 @@
 
 import com.uber.cadence.WorkflowExecution;
 import com.uber.cadence.WorkflowType;
+import com.uber.cadence.workflow.Promise;
 import java.time.Duration;
 import java.util.function.BiConsumer;
 import java.util.function.Consumer;
@@ -87,7 +88,7 @@ Consumer<Exception> startChildWorkflow(
   Consumer<Exception> signalWorkflowExecution(
       SignalExternalWorkflowParameters signalParameters, BiConsumer<Void, Exception> callback);
 
-  void requestCancelWorkflowExecution(WorkflowExecution execution);
+  Promise<Void> requestCancelWorkflowExecution(WorkflowExecution execution);
 
   void continueAsNewOnCompletion(ContinueAsNewWorkflowExecutionParameters parameters);
 

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

@@ -24,11 +24,13 @@
 import com.uber.cadence.WorkflowExecution;
 import com.uber.cadence.WorkflowExecutionStartedEventAttributes;
 import com.uber.cadence.WorkflowType;
+import com.uber.cadence.workflow.Promise;
+import com.uber.cadence.workflow.Workflow;
 import java.time.Duration;
 import java.util.function.BiConsumer;
 import java.util.function.Consumer;
 
-final class DecisionContextImpl implements DecisionContext, HistoryEventHandler {
+public final class DecisionContextImpl implements DecisionContext, HistoryEventHandler {
 
   private final ActivityDecisionContext activityClient;
 
@@ -135,8 +137,10 @@ public Consumer<Exception> signalWorkflowExecution(
   }
 
   @Override
-  public void requestCancelWorkflowExecution(WorkflowExecution execution) {
+  public Promise<Void> requestCancelWorkflowExecution(WorkflowExecution execution) {
     workflowClient.requestCancelWorkflowExecution(execution);
+    // TODO: Make promise return success or failure of the cancellation request.
+    return Workflow.newPromise(null);
   }
 
   @Override

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

@@ -49,6 +49,7 @@
 import com.uber.cadence.TimerCanceledEventAttributes;
 import com.uber.cadence.TimerStartedEventAttributes;
 import com.uber.cadence.WorkflowExecutionStartedEventAttributes;
+import com.uber.cadence.WorkflowType;
 import com.uber.cadence.internal.common.WorkflowExecutionUtils;
 import com.uber.cadence.internal.worker.WorkflowExecutionException;
 import java.nio.charset.StandardCharsets;
@@ -335,8 +336,13 @@ void continueAsNewWorkflowExecution(ContinueAsNewWorkflowExecutionParameters con
         task.getHistory().getEvents().get(0).getWorkflowExecutionStartedEventAttributes();
     ContinueAsNewWorkflowExecutionDecisionAttributes attributes =
         new ContinueAsNewWorkflowExecutionDecisionAttributes();
-    attributes.setWorkflowType(task.getWorkflowType());
     attributes.setInput(continueParameters.getInput());
+    String workflowType = continueParameters.getWorkflowType();
+    if (workflowType != null && !workflowType.isEmpty()) {
+      attributes.setWorkflowType(new WorkflowType().setName(workflowType));
+    } else {
+      attributes.setWorkflowType(task.getWorkflowType());
+    }
     int executionStartToClose = continueParameters.getExecutionStartToCloseTimeoutSeconds();
     if (executionStartToClose == 0) {
       executionStartToClose = startedEvent.getExecutionStartToCloseTimeoutSeconds();

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

@@ -24,7 +24,7 @@
 import com.uber.cadence.TimerStartedEventAttributes;
 import com.uber.cadence.WorkflowExecutionSignaledEventAttributes;
 import com.uber.cadence.WorkflowQuery;
-import com.uber.cadence.internal.common.InternalUtils;
+import com.uber.cadence.internal.common.OptionsUtils;
 import com.uber.cadence.internal.worker.WorkflowExecutionException;
 import com.uber.cadence.workflow.Functions;
 import java.time.Duration;
@@ -278,7 +278,7 @@ private void updateTimers() {
     // Round up to the nearest second as we don't want to deliver a timer
     // earlier than requested.
     long delaySeconds =
-        InternalUtils.roundUpToSeconds(Duration.ofMillis(delayMilliseconds)).getSeconds();
+        OptionsUtils.roundUpToSeconds(Duration.ofMillis(delayMilliseconds)).getSeconds();
     if (timerCancellationHandler != null) {
       timerCancellationHandler.accept(null);
       timerCancellationHandler = null;