wip

Author: antmendoza

core/src/main/java/io/temporal/samples/taskinteraction/README.md

@@ -1,49 +1,47 @@
 # Demo tasks interaction
 
-This example demonstrate a generic implementation for "User Tasks" interaction in Temporal.
-
-TODO 
-
-Temporal does not have such concept of "human task", as BPM systems, but it can be easily implemented with 
-the pattern: 
-- The main workflow have an activity (or local activity) that send the request to an external service. 
-The external for this example is another workflow ([WorkflowTaskManagerImpl.java](WorkflowTaskManagerImpl.java)), 
+This example demonstrates a generic implementation for "User Tasks" interaction with Temporal, 
+which can be easily implemented as follows: 
+- The main workflow [WorkflowWithTasks](./WorkflowWithTasks.java) have an activity (or local activity) that send the request to an external service. 
+The _external service_, for this example, is another workflow ([WorkflowTaskManager](WorkflowTaskManager.java)), 
 that takes care of the task life-cicle.
-- The main workflow wait with `Workflow.await` to receive a Signal.The external service signal back the main 
+- The main workflow waits, with `Workflow.await`, to receive a Signal. 
+- The _external service_ signal back the main 
 workflow to unblock it.
 
-The three steps mentioned above are encapsulated in the class [TaskClient.java](./TaskClient.java)
+The two first steps mentioned above are encapsulated in the class [TaskService.java](./TaskService.java), to make it easily reusable.
 
 ## Run the sample
 
-- Schedule the main workflow execution
+- Schedule the main workflow execution ([WorkflowWithTasks](./WorkflowWithTasks.java)), the one that contains the _User Tasks_ 
 
 ```bash
 ./gradlew -q execute -PmainClass=io.temporal.samples.taskinteraction.client.StartWorkflow
 ```
 
-- Open other terminal and Start the Worker
+- Open other terminal and start the Worker
 
 ```bash
 ./gradlew -q execute -PmainClass=io.temporal.samples.taskinteraction.worker.Worker
 ```
 
-The worker will start the workflow execution and schedule the two activities: 
-
+You will notice, from the worker logs, that it start the main workflow and execute two activities, the
+two activities register two tasks to the external service ([WorkflowTaskManagerImpl.java](WorkflowTaskManagerImpl.java))
 
 ```
-06:08:22.927 {WorkflowWithTasks0.25382038076376945 } [workflow[WorkflowWithTasks0.25382038076376945]-1] INFO  i.t.s.taskinteraction.TaskService - Before creating task : Task{token='WorkflowWithTasks0.25382038076376945-1713845302806-1', title=TaskTitle{value='TODO 1'}} 
-06:08:22.958 {WorkflowWithTasks0.25382038076376945 } [workflow[WorkflowWithTasks0.25382038076376945]-2] INFO  i.t.s.taskinteraction.TaskService - Before creating task : Task{token='WorkflowWithTasks0.25382038076376945-1713845302806-2', title=TaskTitle{value='TODO 2'}} 
-06:08:23.039 {WorkflowWithTasks0.25382038076376945 } [workflow[WorkflowWithTasks0.25382038076376945]-1] INFO  i.t.s.taskinteraction.TaskService - Task created: Task{token='WorkflowWithTasks0.25382038076376945-1713845302806-1', title=TaskTitle{value='TODO 1'}} 
-06:08:23.039 {WorkflowWithTasks0.25382038076376945 } [workflow[WorkflowWithTasks0.25382038076376945]-2] INFO  i.t.s.taskinteraction.TaskService - Task created: Task{token='WorkflowWithTasks0.25382038076376945-1713845302806-2', title=TaskTitle{value='TODO 2'}} 
-
+07:19:39.528 {WorkflowWithTasks1714454371179 } [workflow[WorkflowWithTasks1714454371179]-1] INFO  i.t.s.taskinteraction.TaskService - Before creating task : Task{token='WorkflowWithTasks1714454371179_1', title=TaskTitle{value='TODO 1'}} 
+07:19:39.563 {WorkflowWithTasks1714454371179 } [workflow[WorkflowWithTasks1714454371179]-2] INFO  i.t.s.taskinteraction.TaskService - Before creating task : Task{token='WorkflowWithTasks1714454371179_2', title=TaskTitle{value='TODO 2'}} 
+07:19:39.683 {WorkflowWithTasks1714454371179 } [workflow[WorkflowWithTasks1714454371179]-1] INFO  i.t.s.taskinteraction.TaskService - Task created: Task{token='WorkflowWithTasks1714454371179_1', title=TaskTitle{value='TODO 1'}} 
+07:19:39.684 {WorkflowWithTasks1714454371179 } [workflow[WorkflowWithTasks1714454371179]-2] INFO  i.t.s.taskinteraction.TaskService - Task created: Task{token='WorkflowWithTasks1714454371179_2', title=TaskTit
 ```
 
-- Complete task in the "External system". This class will query and complete one of the 
-pending task in the external system, in this case a workflow, that will at the same time, 
-signal back the main workflow (the one that created the task and is waiting)
+- Now, we can start completing the tasks using the helper class [CompleteTask.java](./client/CompleteTask.java)
 
 ```bash
 ./gradlew -q execute -PmainClass=io.temporal.samples.taskinteraction.client.CompleteTask
 ```
+You can see from the implementation that [WorkflowWithTasksImpl](./WorkflowWithTasksImpl.java) has three task.
+- two in parallel using `Async.procedure`,
+- one blocking task at the end.
 
+This class needs to be run three times. After the three task are completed the main workflow completes.

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

@@ -28,6 +28,10 @@
 import java.util.Map;
 import org.slf4j.Logger;
 
+/**
+ * This class responsibility is to register the task in the external system and waits for the
+ * external system to signal back.
+ */
 public class TaskService<R> {
 
   private final ActivityTask activity =

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

@@ -20,11 +20,7 @@
 package io.temporal.samples.taskinteraction;
 
 import io.temporal.workflow.Workflow;
-import java.util.ArrayList;
-import java.util.List;
-import java.util.Optional;
-import java.util.StringTokenizer;
-import org.jetbrains.annotations.NotNull;
+import java.util.*;
 
 public class WorkflowTaskManagerImpl implements WorkflowTaskManager {
 
@@ -39,27 +35,21 @@ public void execute(List<Task> inputPendingTask, List<String> inputTaskToComplet
 
     while (true) {
 
-      final List<Task> currentTask = new ArrayList<>(pendingTask);
-
       Workflow.await(
           () ->
-              // Wait until one task is added / removed
-              currentTask.size() != pendingTask.size()
-                  // or there are pending task to complete
-                  || !tasksToComplete.isEmpty());
-
-      if (!tasksToComplete.isEmpty()) {
+              // Wait until there are pending task to complete
+              !tasksToComplete.isEmpty());
 
-        final String taskToken = tasksToComplete.remove(0);
-        final String externalWorkflowId = new StringTokenizer(taskToken, "_").nextToken();
+      final String taskToken = tasksToComplete.remove(0);
 
-        Workflow.newExternalWorkflowStub(TaskClient.class, externalWorkflowId)
-            .completeTaskByToken(taskToken);
+      // Find the workflow id of the workflow we have to signal back
+      final String externalWorkflowId = new StringTokenizer(taskToken, "_").nextToken();
 
-        final Task task = getPendingTaskWithToken(taskToken).get();
+      Workflow.newExternalWorkflowStub(TaskClient.class, externalWorkflowId)
+          .completeTaskByToken(taskToken);
 
-        pendingTask.remove(task);
-      }
+      final Task task = getPendingTaskWithToken(taskToken).get();
+      pendingTask.remove(task);
 
       if (Workflow.getInfo().isContinueAsNewSuggested()) {
         Workflow.newContinueAsNewStub(WorkflowTaskManager.class)
@@ -68,26 +58,6 @@ public void execute(List<Task> inputPendingTask, List<String> inputTaskToComplet
     }
   }
 
-  @NotNull
-  private Optional<Task> getPendingTaskWithToken(final String taskToken) {
-    return pendingTask.stream().filter((t) -> t.getToken().equals(taskToken)).findFirst();
-  }
-
-  private void initTaskToComplete(final List<String> tasks) {
-    if (tasksToComplete == null) {
-      tasksToComplete = new ArrayList<>();
-    }
-    tasksToComplete.addAll(tasks);
-  }
-
-  private void initPendingTasks(final List<Task> tasks) {
-
-    if (pendingTask == null) {
-      pendingTask = new ArrayList<>();
-    }
-    pendingTask.addAll(tasks);
-  }
-
   @Override
   public void createTask(Task task) {
     initPendingTasks(new ArrayList<>());
@@ -102,7 +72,7 @@ public void completeTaskByToken(String taskToken) {
     Workflow.await(
         () -> {
           final boolean taskCompleted =
-              getPendingTask().stream().filter((t) -> t.getToken() == taskToken).count() == 0;
+              getPendingTask().stream().noneMatch((t) -> Objects.equals(t.getToken(), taskToken));
 
           return taskCompleted;
         });
@@ -112,4 +82,23 @@ public void completeTaskByToken(String taskToken) {
   public List<Task> getPendingTask() {
     return pendingTask;
   }
+
+  private Optional<Task> getPendingTaskWithToken(final String taskToken) {
+    return pendingTask.stream().filter((t) -> t.getToken().equals(taskToken)).findFirst();
+  }
+
+  private void initTaskToComplete(final List<String> tasks) {
+    if (tasksToComplete == null) {
+      tasksToComplete = new ArrayList<>();
+    }
+    tasksToComplete.addAll(tasks);
+  }
+
+  private void initPendingTasks(final List<Task> tasks) {
+
+    if (pendingTask == null) {
+      pendingTask = new ArrayList<>();
+    }
+    pendingTask.addAll(tasks);
+  }
 }

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

@@ -25,6 +25,7 @@
 import java.util.Arrays;
 import org.slf4j.Logger;
 
+/** Workflow that creates three task and waits for them to complete */
 public class WorkflowWithTasksImpl implements WorkflowWithTasks {
 
   private final Logger logger = Workflow.getLogger(WorkflowWithTasksImpl.class);
@@ -35,12 +36,10 @@ public class WorkflowWithTasksImpl implements WorkflowWithTasks {
   public void execute() {
 
     // Schedule two "tasks" in parallel. The last parameter is the token the client needs
-    // to change the task state, and to complete the task eventually
-
+    // to unblock/complete the task. This token contains the workflowId that the
+    // client can use to create the workflow stub.
     final TaskToken taskToken = new TaskToken();
 
-    // Schedule two "tasks" in parallel. The last parameter is the token the client needs
-    // to change the task state, and ultimately to complete the task
     logger.info("About to create async tasks");
     final Promise<Void> task1 =
         Async.procedure(

core/src/main/java/io/temporal/samples/taskinteraction/activity/ActivityTaskImpl.java

@@ -55,6 +55,7 @@ public void createTask(Task task) {
       // This will be handled differently once updateWithStart is implemented
     }
 
+    // register the "task" to the external workflow that manages task lifecycle
     taskManager.createTask(task);
   }
 }

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

@@ -25,36 +25,33 @@
 import io.temporal.serviceclient.WorkflowServiceStubs;
 import java.util.List;
 
+/**
+ * This class helps to complete tasks in the external workflow. Queries for pending task and
+ * complete one of them
+ */
 public class CompleteTask {
 
   public static void main(String[] args) throws InterruptedException {
 
     final WorkflowServiceStubs service = WorkflowServiceStubs.newLocalServiceStubs();
     final WorkflowClient client = WorkflowClient.newInstance(service);
 
-    while (true) {
-
-      // WorkflowTaskManager keeps and manage workflow task lifecycle
-      final WorkflowTaskManager workflowTaskManager =
-          client.newWorkflowStub(WorkflowTaskManager.class, WorkflowTaskManager.WORKFLOW_ID);
+    // WorkflowTaskManager keeps and manage workflow task lifecycle
+    final WorkflowTaskManager workflowTaskManager =
+        client.newWorkflowStub(WorkflowTaskManager.class, WorkflowTaskManager.WORKFLOW_ID);
 
-      Thread.sleep(200);
-      final List<Task> pendingTask = getPendingTask(workflowTaskManager);
-      System.out.println("Pending task " + pendingTask);
+    Thread.sleep(200);
+    final List<Task> pendingTask = getPendingTask(workflowTaskManager);
+    System.out.println("Pending task " + pendingTask);
 
-      if (pendingTask.isEmpty()) {
-        // Thread.sleep(1000);
-        continue;
-      }
+    if (!pendingTask.isEmpty()) {
 
       final Task nextOpenTask = pendingTask.get(0);
       System.out.println("Completing task with token " + nextOpenTask);
       workflowTaskManager.completeTaskByToken(nextOpenTask.getToken());
-
-      System.out.println("Pending task " + getPendingTask(workflowTaskManager));
     }
 
-    //    System.exit(0);
+    System.out.println("Pending task " + getPendingTask(workflowTaskManager));
   }
 
   private static List<Task> getPendingTask(final WorkflowTaskManager workflowTaskManager) {

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

@@ -26,6 +26,7 @@
 import io.temporal.samples.taskinteraction.WorkflowWithTasks;
 import io.temporal.serviceclient.WorkflowServiceStubs;
 
+/** Client that start schedule WorkflowWithTasks. */
 public class StartWorkflow {
 
   public static void main(String[] args) throws InterruptedException {
@@ -43,7 +44,7 @@ public static void main(String[] args) throws InterruptedException {
 
     System.out.println("Starting workflow: " + WorkflowWithTasks.WORKFLOW_ID);
 
-    // Execute workflow waiting for it to complete.
+    // Schedule workflow and waiting for it to complete.
     workflow.execute();
 
     System.out.println("Workflow completed: " + WorkflowWithTasks.WORKFLOW_ID);