wip

Author: antmendoza

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

@@ -2,6 +2,8 @@
 
 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. 

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

@@ -30,7 +30,7 @@ public interface WorkflowTaskManager {
   @WorkflowMethod
   void execute(List<Task> inputPendingTask, List<String> inputTaskToComplete);
 
-  @SignalMethod
+  @UpdateMethod
   void createTask(Task task);
 
   @UpdateMethod

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

@@ -30,7 +30,7 @@ public class WorkflowTaskManagerImpl implements WorkflowTaskManager {
 
   private List<Task> pendingTask;
 
-  private List<String> taskToComplete;
+  private List<String> tasksToComplete;
 
   @Override
   public void execute(List<Task> inputPendingTask, List<String> inputTaskToComplete) {
@@ -46,23 +46,27 @@ public void execute(List<Task> inputPendingTask, List<String> inputTaskToComplet
               // Wait until one task is added / removed
               currentTask.size() != pendingTask.size()
                   // or there are pending task to complete
-                  || !taskToComplete.isEmpty());
+                  || !tasksToComplete.isEmpty());
 
-      if (!taskToComplete.isEmpty()) {
+      if (!tasksToComplete.isEmpty()) {
 
-        final String taskToken = taskToComplete.remove(0);
-        final String externalWorkflowId = new StringTokenizer(taskToken, "-").nextToken();
+        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);
       }
 
       if (Workflow.getInfo().isContinueAsNewSuggested()) {
         Workflow.newContinueAsNewStub(WorkflowTaskManager.class)
-            .execute(pendingTask, taskToComplete);
+            .execute(pendingTask, tasksToComplete);
       }
     }
   }
@@ -73,10 +77,10 @@ private Optional<Task> getPendingTaskWithToken(final String taskToken) {
   }
 
   private void initTaskToComplete(final List<String> tasks) {
-    if (taskToComplete == null) {
-      taskToComplete = new ArrayList<>();
+    if (tasksToComplete == null) {
+      tasksToComplete = new ArrayList<>();
     }
-    taskToComplete.addAll(tasks);
+    tasksToComplete.addAll(tasks);
   }
 
   private void initPendingTasks(final List<Task> tasks) {
@@ -89,14 +93,19 @@ 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) {
 
-    taskToComplete.add(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/WorkflowWithTasksImpl.java

@@ -71,7 +71,7 @@ private static class TaskToken {
 
     public String getNext() {
 
-      return Workflow.getInfo().getWorkflowId() + "-" + taskToken++;
+      return Workflow.getInfo().getWorkflowId() + "_" + taskToken++;
     }
   }
 }

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

@@ -19,18 +19,17 @@
 
 package io.temporal.samples.taskinteraction.activity;
 
-import static io.temporal.samples.taskinteraction.worker.Worker.TASK_QUEUE;
-
+import io.temporal.activity.Activity;
 import io.temporal.client.WorkflowClient;
+import io.temporal.client.WorkflowExecutionAlreadyStarted;
 import io.temporal.client.WorkflowOptions;
-import io.temporal.client.WorkflowStub;
 import io.temporal.samples.taskinteraction.Task;
 import io.temporal.samples.taskinteraction.WorkflowTaskManager;
 import java.util.ArrayList;
 
 public class ActivityTaskImpl implements ActivityTask {
 
-  private WorkflowClient workflowClient;
+  private final WorkflowClient workflowClient;
 
   public ActivityTaskImpl(WorkflowClient workflowClient) {
     this.workflowClient = workflowClient;
@@ -39,15 +38,23 @@ public ActivityTaskImpl(WorkflowClient workflowClient) {
   @Override
   public void createTask(Task task) {
 
-    WorkflowStub taskManager =
-        workflowClient.newUntypedWorkflowStub(
-            WorkflowTaskManager.class.getSimpleName(),
-            WorkflowOptions.newBuilder()
-                .setWorkflowId(WorkflowTaskManager.WORKFLOW_ID)
-                .setTaskQueue(TASK_QUEUE)
-                .build());
+    final String taskQueue = Activity.getExecutionContext().getInfo().getActivityTaskQueue();
+
+    final WorkflowOptions workflowOptions =
+        WorkflowOptions.newBuilder()
+            .setWorkflowId(WorkflowTaskManager.WORKFLOW_ID)
+            .setTaskQueue(taskQueue)
+            .build();
+
+    final WorkflowTaskManager taskManager =
+        workflowClient.newWorkflowStub(WorkflowTaskManager.class, workflowOptions);
+    try {
+      WorkflowClient.start(taskManager::execute, new ArrayList<>(), new ArrayList<>());
+    } catch (WorkflowExecutionAlreadyStarted e) {
+      // expected exception if workflow was started by a previous activity execution.
+      // This will be handled differently once updateWithStart is implemented
+    }
 
-    taskManager.signalWithStart(
-        "createTask", new Object[] {task}, new Object[] {new ArrayList<>(), new ArrayList<>()});
+    taskManager.createTask(task);
   }
 }

core/src/main/java/io/temporal/samples/taskinteraction/worker/Worker.java

@@ -29,7 +29,7 @@
 
 public class Worker {
 
-  public static final String TASK_QUEUE = "TaskWorkflowImplTaskQueue";
+  public static final String TASK_QUEUE = "TaskInteractionQueue";
 
   public static void main(String[] args) {