wip

Author: antmendoza

core/src/main/java/io/temporal/samples/taskinteraction/WorkflowTaskManagerImpl.java

@@ -50,15 +50,12 @@ public void execute(List<Task> inputPendingTask, List<String> inputTaskToComplet
 
       if (!tasksToComplete.isEmpty()) {
 
-        System.out.println("tasksToComplete >>>>> " + tasksToComplete);
-
         final String taskToken = tasksToComplete.remove(0);
         final String externalWorkflowId = new StringTokenizer(taskToken, "_").nextToken();
 
         Workflow.newExternalWorkflowStub(TaskClient.class, externalWorkflowId)
             .completeTaskByToken(taskToken);
 
-        System.out.println("getPendingTaskWithToken >>>>> " + taskToken);
         final Task task = getPendingTaskWithToken(taskToken).get();
 
         pendingTask.remove(task);
@@ -93,18 +90,13 @@ private void initPendingTasks(final List<Task> tasks) {
 
   @Override
   public void createTask(Task task) {
-    System.out.println("creating task " + task);
-
     initPendingTasks(new ArrayList<>());
     pendingTask.add(task);
   }
 
   @Override
   public void completeTaskByToken(String taskToken) {
 
-    System.out.println("completeTaskByToken adding taskToken >>>>> " + taskToken);
-    System.out.println("completeTaskByToken >>>>> " + tasksToComplete);
-
     tasksToComplete.add(taskToken);
 
     Workflow.await(

core/src/main/java/io/temporal/samples/taskinteraction/client/StartWorkflow.java

@@ -33,16 +33,6 @@ public static void main(String[] args) throws InterruptedException {
     final WorkflowServiceStubs service = WorkflowServiceStubs.newLocalServiceStubs();
     final WorkflowClient client = WorkflowClient.newInstance(service);
 
-    extracted(client);
-    while (true) {
-
-      extracted(client);
-
-      Thread.sleep(200);
-    }
-  }
-
-  private static void extracted(final WorkflowClient client) {
     final WorkflowWithTasks workflow =
         client.newWorkflowStub(
             WorkflowWithTasks.class,
@@ -51,9 +41,13 @@ private static void extracted(final WorkflowClient client) {
                 .setTaskQueue(TASK_QUEUE)
                 .build());
 
-    System.out.println("Starting workflow " + WorkflowWithTasks.WORKFLOW_ID);
+    System.out.println("Starting workflow: " + WorkflowWithTasks.WORKFLOW_ID);
 
     // Execute workflow waiting for it to complete.
-    WorkflowClient.execute(workflow::execute);
+    workflow.execute();
+
+    System.out.println("Workflow completed: " + WorkflowWithTasks.WORKFLOW_ID);
+
+    System.exit(0);
   }
 }

core/src/test/java/io/temporal/samples/taskinteraction/WorkflowWithTasksImplTest.java

@@ -27,20 +27,27 @@
 import io.temporal.api.workflowservice.v1.DescribeWorkflowExecutionResponse;
 import io.temporal.client.WorkflowClient;
 import io.temporal.client.WorkflowOptions;
+import io.temporal.common.interceptors.*;
 import io.temporal.samples.taskinteraction.activity.ActivityTaskImpl;
 import io.temporal.testing.TestWorkflowRule;
-import java.time.Duration;
+import io.temporal.worker.WorkerFactoryOptions;
+import io.temporal.workflow.Workflow;
 import java.util.List;
+import java.util.concurrent.CompletableFuture;
 import org.junit.Rule;
 import org.junit.Test;
 
 public class WorkflowWithTasksImplTest {
 
+  private MyWorkerInterceptor myWorkerInterceptor = new MyWorkerInterceptor();
+
   @Rule
   public TestWorkflowRule testWorkflowRule =
       TestWorkflowRule.newBuilder()
-          // .setNamespace("default")
-          // .setUseExternalService(true)
+          .setWorkerFactoryOptions(
+              WorkerFactoryOptions.newBuilder()
+                  .setWorkerInterceptors(myWorkerInterceptor)
+                  .validateAndBuildWithDefaults())
           .setDoNotStart(true)
           .build();
 
@@ -56,6 +63,7 @@ public void testEnd2End() {
     testWorkflowRule
         .getWorker()
         .registerActivitiesImplementations(new ActivityTaskImpl(workflowClient));
+
     testWorkflowRule.getTestEnvironment().start();
 
     WorkflowWithTasks workflow =
@@ -70,23 +78,25 @@ public void testEnd2End() {
 
     WorkflowExecution execution = WorkflowClient.start(workflow::execute);
 
-    // TODO
-    testWorkflowRule.getTestEnvironment().sleep(Duration.ofSeconds(2));
+    // Wait until the first two tasks from WorkflowWithTasks are created in WorkflowTaskManager
+    myWorkerInterceptor.waitUntilTwoCreateTaskInvocations();
 
     WorkflowTaskManager workflowManager =
         workflowClient.newWorkflowStub(WorkflowTaskManager.class, WorkflowTaskManager.WORKFLOW_ID);
 
     final List<Task> pendingTask = getPendingTask(workflowManager);
     assertEquals(2, pendingTask.size());
 
-    // Let's complete the two pending task. Send update to the workflow that holds and keep tasks
-    // state
+    // Complete the two pending task created in parallel from `WorkflowWithTasks`.
+    // Send update to the workflow that keeps tasks state, that will signal back
+    // the `WorkflowWithTasks` execution
     workflowManager.completeTaskByToken(pendingTask.get(0).getToken());
     workflowManager.completeTaskByToken(pendingTask.get(1).getToken());
 
-    // TODO
-    testWorkflowRule.getTestEnvironment().sleep(Duration.ofSeconds(2));
+    // Wait until the last task in WorkflowWithTasks is created in WorkflowTaskManager
+    myWorkerInterceptor.waitUntilThreeInvocationsOfCreateTask();
 
+    // Complete the last task in `WorkflowWithTasks`
     assertEquals(1, getPendingTask(workflowManager).size());
     workflowManager.completeTaskByToken(getPendingTask(workflowManager).get(0).getToken());
 
@@ -115,4 +125,59 @@ private DescribeWorkflowExecutionResponse getDescribeWorkflowExecutionResponse(
   private static List<Task> getPendingTask(final WorkflowTaskManager workflowManager) {
     return workflowManager.getPendingTask();
   }
+
+  private class MyWorkerInterceptor extends WorkerInterceptorBase {
+
+    private int createTaskInvocations = 0;
+
+    private CompletableFuture<Void> waitUntilTwoInvocationsOfCreateTask;
+
+    private CompletableFuture<Void> waitUntilThreeInvocationsOfCreateTask;
+
+    public MyWorkerInterceptor() {
+      waitUntilTwoInvocationsOfCreateTask = new CompletableFuture<>();
+      waitUntilThreeInvocationsOfCreateTask = new CompletableFuture<>();
+    }
+
+    public Void waitUntilTwoCreateTaskInvocations() {
+      return getFromCompletableFuture(waitUntilTwoInvocationsOfCreateTask);
+    }
+
+    public Void waitUntilThreeInvocationsOfCreateTask() {
+      return getFromCompletableFuture(waitUntilThreeInvocationsOfCreateTask);
+    }
+
+    private Void getFromCompletableFuture(final CompletableFuture<Void> completableFuture) {
+      try {
+        return completableFuture.get();
+      } catch (Exception e) {
+        throw new RuntimeException(e);
+      }
+    }
+
+    @Override
+    public WorkflowInboundCallsInterceptor interceptWorkflow(
+        final WorkflowInboundCallsInterceptor next) {
+      return new WorkflowInboundCallsInterceptorBase(next) {
+        @Override
+        public UpdateOutput executeUpdate(final UpdateInput input) {
+          if (input.getUpdateName().equals("createTask")
+              && Workflow.getInfo()
+                  .getWorkflowType()
+                  .equals(WorkflowTaskManager.class.getSimpleName())) {
+            createTaskInvocations++;
+            if (createTaskInvocations == 2) {
+              waitUntilTwoInvocationsOfCreateTask.complete(null);
+            }
+
+            if (createTaskInvocations == 3) {
+              waitUntilThreeInvocationsOfCreateTask.complete(null);
+            }
+          }
+
+          return super.executeUpdate(input);
+        }
+      };
+    }
+  }
 }