Merge pull request #152 from meiliang86/child-wf-stub

fix issues in Async promise execution

Author: meiliang86

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

@@ -35,13 +35,14 @@ class ChildWorkflowStubImpl implements ChildWorkflowStub {
   private final String workflowType;
   private final ChildWorkflowOptions options;
   private final WorkflowInterceptor decisionContext;
-  private CompletablePromise<WorkflowExecution> execution;
+  private final CompletablePromise<WorkflowExecution> execution;
 
   ChildWorkflowStubImpl(
       String workflowType, ChildWorkflowOptions options, WorkflowInterceptor decisionContext) {
     this.workflowType = Objects.requireNonNull(workflowType);
     this.options = new ChildWorkflowOptions.Builder(options).validateAndBuildWithDefaults();
     this.decisionContext = Objects.requireNonNull(decisionContext);
+    this.execution = Workflow.newPromise();
   }
 
   @Override
@@ -80,20 +81,12 @@ public <R> R execute(Class<R> returnType, Object... args) {
   public <R> Promise<R> executeAsync(Class<R> returnType, Object... args) {
     WorkflowResult<R> result =
         decisionContext.executeChildWorkflow(workflowType, returnType, args, options);
-    execution = Workflow.newPromise();
     execution.completeFrom(result.getWorkflowExecution());
     return result.getResult();
   }
 
   @Override
   public void signal(String signalName, Object... args) {
-    if (execution == null) {
-      throw new IllegalStateException(
-          "This stub cannot be used to signal a workflow"
-              + " without starting it first. "
-              + "To signal a workflow execution that was started elsewhere "
-              + "use a stub created through Workflow.newExternalWorkflowStub");
-    }
     Promise<Void> signaled =
         decisionContext.signalExternalWorkflow(execution.get(), signalName, args);
     if (AsyncInternal.isAsync()) {

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

@@ -167,7 +167,7 @@ public static <T> T newExternalWorkflowStub(
             new ExternalWorkflowInvocationHandler(execution, getWorkflowInterceptor()));
   }
 
-  public static Promise<WorkflowExecution> getChildWorkflowExecution(Object workflowStub) {
+  public static Promise<WorkflowExecution> getWorkflowExecution(Object workflowStub) {
     if (workflowStub instanceof WorkflowStub) {
       return ((WorkflowStub) workflowStub).__getWorkflowExecution();
     }

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

@@ -545,6 +545,7 @@ public void childWorkflowStarted(ChildWorkflowExecutionStartedEventAttributes a)
           String childId = a.getWorkflowExecution().getWorkflowId();
           StateMachine<ChildWorkflowData> child = getChildWorkflow(childId);
           child.action(StateMachines.Action.START, ctx, a, 0);
+          scheduleDecision(ctx);
           // No need to lock until completion as child workflow might skip
           // time as well
           ctx.unlockTimer();

src/main/java/com/uber/cadence/workflow/Async.java

@@ -148,7 +148,7 @@ public static Promise<Void> procedure(Functions.Proc procedure) {
    * @return Promise that contains procedure result or failure
    */
   public static <A1> Promise<Void> procedure(Functions.Proc1<A1> procedure, A1 arg1) {
-    return procedure(() -> procedure.apply(arg1));
+    return AsyncInternal.procedure(procedure, arg1);
   }
 
   /**

src/main/java/com/uber/cadence/workflow/Functions.java

@@ -57,7 +57,7 @@ public interface Func6<T1, T2, T3, T4, T5, T6, R> extends Serializable {
   }
 
   @FunctionalInterface
-  public interface Proc {
+  public interface Proc extends Serializable {
     void apply();
   }
 

src/main/java/com/uber/cadence/workflow/Workflow.java

@@ -443,7 +443,7 @@ public static <R> R newExternalWorkflowStub(
    * as child workflow start is asynchronous.
    */
   public static Promise<WorkflowExecution> getWorkflowExecution(Object childWorkflowStub) {
-    return WorkflowInternal.getChildWorkflowExecution(childWorkflowStub);
+    return WorkflowInternal.getWorkflowExecution(childWorkflowStub);
   }
 
   /**

src/test/java/com/uber/cadence/workflow/WorkflowTest.java

@@ -19,6 +19,8 @@
 
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNotEquals;
+import static org.junit.Assert.assertNotNull;
 import static org.junit.Assert.assertNull;
 import static org.junit.Assert.assertTrue;
 import static org.junit.Assert.fail;
@@ -46,6 +48,7 @@
 import com.uber.cadence.common.RetryOptions;
 import com.uber.cadence.converter.JsonDataConverter;
 import com.uber.cadence.internal.sync.DeterministicRunnerTest;
+import com.uber.cadence.serviceclient.WorkflowServiceTChannel;
 import com.uber.cadence.testing.TestEnvironmentOptions;
 import com.uber.cadence.testing.TestEnvironmentOptions.Builder;
 import com.uber.cadence.testing.TestWorkflowEnvironment;
@@ -140,6 +143,7 @@ protected void failed(Throwable e, Description description) {
   private TestWorkflowEnvironment testEnvironment;
   private ScheduledExecutorService scheduledExecutor;
   private List<ScheduledFuture<?>> delayedCallbacks = new ArrayList<>();
+  private static WorkflowServiceTChannel service = new WorkflowServiceTChannel();
 
   private static WorkflowOptions.Builder newWorkflowOptionsBuilder(String taskList) {
     return new WorkflowOptions.Builder()
@@ -173,7 +177,7 @@ public void setUp() {
     if (useExternalService) {
       WorkerOptions workerOptions =
           new WorkerOptions.Builder().setInterceptorFactory(tracer).build();
-      worker = new Worker(DOMAIN, taskList, workerOptions);
+      worker = new Worker(service, DOMAIN, taskList, workerOptions);
       workflowClient = WorkflowClient.newInstance(DOMAIN);
       WorkflowClientOptions clientOptions =
           new WorkflowClientOptions.Builder()
@@ -1011,6 +1015,73 @@ public void testChildAsyncWorkflow() {
     assertEquals(null, client.execute(taskList));
   }
 
+  // This workflow is designed specifically for testing some internal logic in Async.procedure
+  // and ChildWorkflowStubImpl. See comments on testChildAsyncLambdaWorkflow for more details.
+  public interface WaitOnSignalWorkflow {
+
+    @WorkflowMethod()
+    void execute();
+
+    @SignalMethod
+    void signal(String value);
+  }
+
+  public static class TestWaitOnSignalWorkflowImpl implements WaitOnSignalWorkflow {
+    private final CompletablePromise<String> signal = Workflow.newPromise();
+
+    @Override
+    public void execute() {
+      signal.get();
+    }
+
+    @Override
+    public void signal(String value) {
+      signal.complete(value);
+    }
+  }
+
+  public static class TestChildAsyncLambdaWorkflow implements TestWorkflow1 {
+
+    @Override
+    public String execute(String taskList) {
+      ChildWorkflowOptions workflowOptions =
+          new ChildWorkflowOptions.Builder()
+              .setExecutionStartToCloseTimeout(Duration.ofSeconds(100))
+              .setTaskStartToCloseTimeout(Duration.ofSeconds(60))
+              .setTaskList(taskList)
+              .build();
+
+      WaitOnSignalWorkflow child =
+          Workflow.newChildWorkflowStub(WaitOnSignalWorkflow.class, workflowOptions);
+      Promise<Void> promise = Async.procedure(() -> child.execute());
+      Promise<WorkflowExecution> executionPromise = Workflow.getWorkflowExecution(child);
+      assertNotNull(executionPromise);
+      WorkflowExecution execution = executionPromise.get();
+      assertNotEquals("", execution.getWorkflowId());
+      assertNotEquals("", execution.getRunId());
+      child.signal("test");
+
+      promise.get();
+      return null;
+    }
+  }
+
+  // The purpose of this test is to exercise the lambda execution logic inside Async.procedure(),
+  // which executes on a different thread than workflow-main. This is different than executing
+  // classes that implements the workflow method interface, which executes on the workflow main
+  // thread.
+  @Test
+  public void testChildAsyncLambdaWorkflow() {
+    startWorkerFor(TestChildAsyncLambdaWorkflow.class, TestWaitOnSignalWorkflowImpl.class);
+
+    WorkflowOptions.Builder options = new WorkflowOptions.Builder();
+    options.setExecutionStartToCloseTimeout(Duration.ofSeconds(200));
+    options.setTaskStartToCloseTimeout(Duration.ofSeconds(60));
+    options.setTaskList(taskList);
+    TestWorkflow1 client = workflowClient.newWorkflowStub(TestWorkflow1.class, options.build());
+    assertEquals(null, client.execute(taskList));
+  }
+
   public static class TestUntypedChildStubWorkflow implements TestWorkflow1 {
 
     @Override