Refactored InvocationHandlers to use Stubs internally. (#134)

* Added ExternalWorkflowInvocationHandler

* Migrated Workflow InvocationHandlers to use untyped stub

* Renamed Stubs

* Added ChildWorkflowStub and ExternalWorkflowStub to isAsync

Author: mfateev

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

@@ -201,7 +201,7 @@ static WorkflowClient newInstance(
    * @param options options used to start a workflow through returned stub
    * @return Stub that can be used to start workflow and later to signal or query it.
    */
-  UntypedWorkflowStub newUntypedWorkflowStub(String workflowType, WorkflowOptions options);
+  WorkflowStub newUntypedWorkflowStub(String workflowType, WorkflowOptions options);
 
   /**
    * Creates workflow untyped client stub for a known execution. Use it to send signals or queries
@@ -213,7 +213,7 @@ static WorkflowClient newInstance(
    * @param workflowType type of the workflow. Optional as it is used for error reporting only.
    * @return Stub that can be used to start workflow and later to signal or query it.
    */
-  UntypedWorkflowStub newUntypedWorkflowStub(
+  WorkflowStub newUntypedWorkflowStub(
       String workflowId, Optional<String> runId, Optional<String> workflowType);
 
   /**
@@ -224,8 +224,7 @@ UntypedWorkflowStub newUntypedWorkflowStub(
    * @param workflowType type of the workflow. Optional as it is used for error reporting only.
    * @return Stub that can be used to start workflow and later to signal or query it.
    */
-  UntypedWorkflowStub newUntypedWorkflowStub(
-      WorkflowExecution execution, Optional<String> workflowType);
+  WorkflowStub newUntypedWorkflowStub(WorkflowExecution execution, Optional<String> workflowType);
 
   /**
    * Creates new {@link ActivityCompletionClient} that can be used to complete activities

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

@@ -22,11 +22,11 @@
 
 public interface WorkflowClientInterceptor {
 
-  UntypedWorkflowStub newUntypedWorkflowStub(
-      String workflowType, WorkflowOptions options, UntypedWorkflowStub next);
+  WorkflowStub newUntypedWorkflowStub(
+      String workflowType, WorkflowOptions options, WorkflowStub next);
 
-  UntypedWorkflowStub newUntypedWorkflowStub(
-      WorkflowExecution execution, Optional<String> workflowType, UntypedWorkflowStub next);
+  WorkflowStub newUntypedWorkflowStub(
+      WorkflowExecution execution, Optional<String> workflowType, WorkflowStub next);
 
   ActivityCompletionClient newActivityCompletionClient(ActivityCompletionClient next);
 }

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

@@ -23,14 +23,14 @@
 public class WorkflowClientInterceptorBase implements WorkflowClientInterceptor {
 
   @Override
-  public UntypedWorkflowStub newUntypedWorkflowStub(
-      String workflowType, WorkflowOptions options, UntypedWorkflowStub next) {
+  public WorkflowStub newUntypedWorkflowStub(
+      String workflowType, WorkflowOptions options, WorkflowStub next) {
     return next;
   }
 
   @Override
-  public UntypedWorkflowStub newUntypedWorkflowStub(
-      WorkflowExecution execution, Optional<String> workflowType, UntypedWorkflowStub next) {
+  public WorkflowStub newUntypedWorkflowStub(
+      WorkflowExecution execution, Optional<String> workflowType, WorkflowStub next) {
     return next;
   }
 

src/main/java/com/uber/cadence/client/UntypedWorkflowStub.java renamed to src/main/java/com/uber/cadence/client/WorkflowStub.java

@@ -23,7 +23,13 @@
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
 
-public interface UntypedWorkflowStub {
+/**
+ * WorkflowStub is a client side stub to a single workflow instance. It can be used to start,
+ * signal, query, wait for completion and cancel a workflow execution. Created through {@link
+ * WorkflowClient#newUntypedWorkflowStub(String, WorkflowOptions)} or {@link
+ * WorkflowClient#newUntypedWorkflowStub(WorkflowExecution, Optional)}.
+ */
+public interface WorkflowStub {
 
   void signal(String signalName, Object... args);
 

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

@@ -17,6 +17,7 @@
 
 package com.uber.cadence.internal.common;
 
+import com.uber.cadence.workflow.WorkflowMethod;
 import java.lang.reflect.Method;
 import java.time.Duration;
 
@@ -34,6 +35,38 @@ public static String getSimpleName(Method method) {
     return method.getDeclaringClass().getSimpleName() + "::" + method.getName();
   }
 
+  public static String getWorkflowType(Method method, WorkflowMethod workflowMethod) {
+    String workflowName = workflowMethod.name();
+    if (workflowName.isEmpty()) {
+      return InternalUtils.getSimpleName(method);
+    } else {
+      return workflowName;
+    }
+  }
+
+  public static Method getWorkflowMethod(Class<?> workflowInterface) {
+    Method result = null;
+    for (Method m : workflowInterface.getMethods()) {
+      if (m.getAnnotation(WorkflowMethod.class) != null) {
+        if (result != null) {
+          throw new IllegalArgumentException(
+              "Workflow interface must have exactly one method "
+                  + "annotated with @WorkflowMethod. Found \""
+                  + result
+                  + "\" and \""
+                  + m
+                  + "\"");
+        }
+        result = m;
+      }
+    }
+    if (result == null) {
+      throw new IllegalArgumentException(
+          "Method annotated with @WorkflowMethod is not " + "found at " + workflowInterface);
+    }
+    return result;
+  }
+
   /**
    * Convert milliseconds to seconds rounding up. Used by timers to ensure that they never fire
    * earlier than requested.

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

@@ -22,7 +22,7 @@
 import com.uber.cadence.activity.MethodRetry;
 import com.uber.cadence.internal.common.InternalUtils;
 import com.uber.cadence.internal.sync.AsyncInternal.AsyncMarker;
-import com.uber.cadence.workflow.UntypedActivityStub;
+import com.uber.cadence.workflow.ActivityStub;
 import com.uber.cadence.workflow.Workflow;
 import java.lang.reflect.InvocationHandler;
 import java.lang.reflect.Method;
@@ -79,8 +79,7 @@ public Object invoke(Object proxy, Method method, Object[] args) {
         }
 
         ActivityOptions mergedOptions = ActivityOptions.merge(activityMethod, methodRetry, options);
-        UntypedActivityStub stub =
-            UntypedActivityStubImpl.newInstance(mergedOptions, activityExecutor);
+        ActivityStub stub = ActivityStubImpl.newInstance(mergedOptions, activityExecutor);
         function = (a) -> stub.execute(activityName, method.getReturnType(), a);
         methodFunctions.put(method, function);
       } catch (NoSuchMethodException e) {

src/main/java/com/uber/cadence/internal/sync/UntypedActivityStubImpl.java renamed to src/main/java/com/uber/cadence/internal/sync/ActivityStubImpl.java

@@ -20,23 +20,22 @@
 import com.google.common.base.Defaults;
 import com.uber.cadence.activity.ActivityOptions;
 import com.uber.cadence.workflow.ActivityException;
+import com.uber.cadence.workflow.ActivityStub;
 import com.uber.cadence.workflow.Promise;
-import com.uber.cadence.workflow.UntypedActivityStub;
 
 /** Supports calling activity by name and arguments without its strongly typed interface. */
-class UntypedActivityStubImpl implements UntypedActivityStub {
+class ActivityStubImpl implements ActivityStub {
 
   private final ActivityOptions options;
   private final ActivityExecutor activityExecutor;
 
-  static UntypedActivityStub newInstance(
-      ActivityOptions options, ActivityExecutor activityExecutor) {
+  static ActivityStub newInstance(ActivityOptions options, ActivityExecutor activityExecutor) {
     ActivityOptions validatedOptions =
         new ActivityOptions.Builder(options).validateAndBuildWithDefaults();
-    return new UntypedActivityStubImpl(validatedOptions, activityExecutor);
+    return new ActivityStubImpl(validatedOptions, activityExecutor);
   }
 
-  private UntypedActivityStubImpl(ActivityOptions options, ActivityExecutor activityExecutor) {
+  private ActivityStubImpl(ActivityOptions options, ActivityExecutor activityExecutor) {
     this.options = options;
     this.activityExecutor = activityExecutor;
   }

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

@@ -21,10 +21,12 @@
 
 import com.uber.cadence.common.RetryOptions;
 import com.uber.cadence.internal.common.LambdaUtils;
+import com.uber.cadence.workflow.ActivityStub;
+import com.uber.cadence.workflow.ChildWorkflowStub;
 import com.uber.cadence.workflow.CompletablePromise;
+import com.uber.cadence.workflow.ExternalWorkflowStub;
 import com.uber.cadence.workflow.Functions;
 import com.uber.cadence.workflow.Promise;
-import com.uber.cadence.workflow.UntypedActivityStub;
 import com.uber.cadence.workflow.Workflow;
 import java.lang.invoke.MethodHandleInfo;
 import java.lang.invoke.SerializedLambda;
@@ -303,7 +305,9 @@ private static <R> Promise<R> execute(boolean async, Functions.Func<R> func) {
   public static boolean isAsync(Object func) {
     SerializedLambda lambda = LambdaUtils.toSerializedLambda(func);
     Object target = getTarget(lambda);
-    return target instanceof UntypedActivityStub
+    return target instanceof ActivityStub
+        || target instanceof ChildWorkflowStub
+        || target instanceof ExternalWorkflowStub
         || (target instanceof AsyncMarker
             && lambda.getImplMethodKind() == MethodHandleInfo.REF_invokeInterface);
   }

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

@@ -17,52 +17,43 @@
 
 package com.uber.cadence.internal.sync;
 
-import com.google.common.base.Defaults;
-import com.uber.cadence.WorkflowExecution;
+import static com.uber.cadence.internal.common.InternalUtils.getWorkflowMethod;
+import static com.uber.cadence.internal.common.InternalUtils.getWorkflowType;
+
 import com.uber.cadence.activity.MethodRetry;
-import com.uber.cadence.converter.DataConverter;
 import com.uber.cadence.internal.common.InternalUtils;
-import com.uber.cadence.workflow.ChildWorkflowException;
 import com.uber.cadence.workflow.ChildWorkflowOptions;
-import com.uber.cadence.workflow.CompletablePromise;
-import com.uber.cadence.workflow.Promise;
+import com.uber.cadence.workflow.ChildWorkflowStub;
 import com.uber.cadence.workflow.QueryMethod;
-import com.uber.cadence.workflow.SignalExternalWorkflowException;
 import com.uber.cadence.workflow.SignalMethod;
-import com.uber.cadence.workflow.Workflow;
 import com.uber.cadence.workflow.WorkflowMethod;
 import java.lang.reflect.InvocationHandler;
 import java.lang.reflect.Method;
 
 /** Dynamic implementation of a strongly typed child workflow interface. */
 class ChildWorkflowInvocationHandler implements InvocationHandler {
 
-  private final ChildWorkflowOptions options;
-  private final SyncDecisionContext decisionContext;
-  private final DataConverter dataConverter;
-  private CompletablePromise<WorkflowExecution> execution = Workflow.newPromise();
-  private boolean startRequested;
+  private final ChildWorkflowStub stub;
 
   ChildWorkflowInvocationHandler(
-      ChildWorkflowOptions options, SyncDecisionContext decisionContext) {
-    this.options = options;
-    this.decisionContext = decisionContext;
-    dataConverter = decisionContext.getDataConverter();
-  }
+      Class<?> workflowInterface,
+      ChildWorkflowOptions options,
+      SyncDecisionContext decisionContext) {
+    Method workflowMethod = getWorkflowMethod(workflowInterface);
+    WorkflowMethod workflowAnnotation = workflowMethod.getAnnotation(WorkflowMethod.class);
+    String workflowType = getWorkflowType(workflowMethod, workflowAnnotation);
+    MethodRetry retryAnnotation = workflowMethod.getAnnotation(MethodRetry.class);
 
-  public ChildWorkflowInvocationHandler(
-      WorkflowExecution execution, SyncDecisionContext decisionContext) {
-    this.options = null;
-    this.decisionContext = decisionContext;
-    dataConverter = decisionContext.getDataConverter();
-    this.execution.complete(execution);
+    ChildWorkflowOptions merged =
+        ChildWorkflowOptions.merge(workflowAnnotation, retryAnnotation, options);
+    this.stub = new ChildWorkflowStubImpl(workflowType, merged, decisionContext);
   }
 
   @Override
   public Object invoke(Object proxy, Method method, Object[] args) {
     // Implement WorkflowStub
     if (method.getName().equals(WorkflowStub.GET_EXECUTION_METHOD_NAME)) {
-      return execution;
+      return stub.getExecution();
     }
     WorkflowMethod workflowMethod = method.getAnnotation(WorkflowMethod.class);
     QueryMethod queryMethod = method.getAnnotation(QueryMethod.class);
@@ -78,17 +69,12 @@ public Object invoke(Object proxy, Method method, Object[] args) {
               + "from @WorkflowMethod, @QueryMethod or @SignalMethod");
     }
     if (workflowMethod != null) {
-      if (startRequested) {
-        throw new IllegalStateException("Already started: " + execution);
-      }
-      startRequested = true;
-      return executeChildWorkflow(method, workflowMethod, args);
+      return stub.execute(method.getReturnType(), args);
     }
     if (queryMethod != null) {
-      if (execution == null) {
-        throw new IllegalStateException("Workflow not started yet");
-      }
-      return queryWorkflow(method, queryMethod, args);
+      throw new UnsupportedOperationException(
+          "Query is not supported from workflow to workflow. "
+              + "Use activity that perform the query instead.");
     }
     if (signalMethod != null) {
       signalWorkflow(method, signalMethod, args);
@@ -103,51 +89,6 @@ private void signalWorkflow(Method method, SignalMethod signalMethod, Object[] a
     if (signalName.isEmpty()) {
       signalName = InternalUtils.getSimpleName(method);
     }
-    Promise<Void> signalled = decisionContext.signalWorkflow(execution.get(), signalName, args);
-    if (AsyncInternal.isAsync()) {
-      AsyncInternal.setAsyncResult(signalled);
-      return;
-    }
-    try {
-      signalled.get();
-    } catch (SignalExternalWorkflowException e) {
-      // Reset stack to the current one. Otherwise it is very confusing to see a stack of
-      // an event handling method.
-      e.setStackTrace(Thread.currentThread().getStackTrace());
-      throw e;
-    }
-  }
-
-  private Object queryWorkflow(Method method, QueryMethod queryMethod, Object[] args) {
-    throw new UnsupportedOperationException(
-        "Query is not supported from workflow to workflow. "
-            + "Use activity that perform the query instead.");
-  }
-
-  private Object executeChildWorkflow(Method method, WorkflowMethod workflowMethod, Object[] args) {
-    String workflowType = workflowMethod.name();
-    if (workflowType.isEmpty()) {
-      workflowType = InternalUtils.getSimpleName(method);
-    }
-    byte[] input = dataConverter.toData(args);
-    MethodRetry retry = method.getAnnotation(MethodRetry.class);
-    ChildWorkflowOptions merged = ChildWorkflowOptions.merge(workflowMethod, retry, options);
-    Promise<byte[]> encodedResult =
-        decisionContext.executeChildWorkflow(workflowType, merged, input, execution);
-    Promise<?> result =
-        encodedResult.thenApply(
-            (encoded) -> dataConverter.fromData(encoded, method.getReturnType()));
-    if (AsyncInternal.isAsync()) {
-      AsyncInternal.setAsyncResult(result);
-      return Defaults.defaultValue(method.getReturnType());
-    }
-    try {
-      return result.get();
-    } catch (ChildWorkflowException e) {
-      // Reset stack to the current one. Otherwise it is very confusing to see a stack of
-      // an event handling method.
-      e.setStackTrace(Thread.currentThread().getStackTrace());
-      throw e;
-    }
+    stub.signal(signalName, args);
   }
 }