Example for tasks interaction

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

@@ -0,0 +1,47 @@
+# Demo tasks interaction
+
+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 waits to receive a Signal. 
+- The _external service_ signal back the main 
+workflow to unblock it.
+
+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 ([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
+
+```bash
+./gradlew -q execute -PmainClass=io.temporal.samples.taskinteraction.worker.Worker
+```
+
+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))
+
+```
+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
+```
+
+- 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/Task.java

@@ -0,0 +1,70 @@
+/*
+ *  Copyright (c) 2020 Temporal Technologies, Inc. All Rights Reserved
+ *
+ *  Copyright 2012-2016 Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ *
+ *  Modifications copyright (C) 2017 Uber Technologies, Inc.
+ *
+ *  Licensed under the Apache License, Version 2.0 (the "License"). You may not
+ *  use this file except in compliance with the License. A copy of the License is
+ *  located at
+ *
+ *  http://aws.amazon.com/apache2.0
+ *
+ *  or in the "license" file accompanying this file. This file is distributed on
+ *  an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either
+ *  express or implied. See the License for the specific language governing
+ *  permissions and limitations under the License.
+ */
+
+package io.temporal.samples.taskinteraction;
+
+public class Task {
+
+  private String token;
+  private TaskTitle title;
+
+  public Task() {}
+
+  public Task(String token, TaskTitle title) {
+    this.token = token;
+    this.title = title;
+  }
+
+  public String getToken() {
+    return token;
+  }
+
+  public TaskTitle getTitle() {
+    return title;
+  }
+
+  @Override
+  public String toString() {
+    return "Task{" + "token='" + token + '\'' + ", title=" + title + '}';
+  }
+
+  public static class TaskTitle {
+    private String value;
+
+    public TaskTitle() {}
+
+    public TaskTitle(final String value) {
+
+      this.value = value;
+    }
+
+    public String getValue() {
+      return value;
+    }
+
+    public void setValue(final String value) {
+      this.value = value;
+    }
+
+    @Override
+    public String toString() {
+      return "TaskTitle{" + "value='" + value + '\'' + '}';
+    }
+  }
+}

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

@@ -0,0 +1,30 @@
+/*
+ *  Copyright (c) 2020 Temporal Technologies, Inc. All Rights Reserved
+ *
+ *  Copyright 2012-2016 Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ *
+ *  Modifications copyright (C) 2017 Uber Technologies, Inc.
+ *
+ *  Licensed under the Apache License, Version 2.0 (the "License"). You may not
+ *  use this file except in compliance with the License. A copy of the License is
+ *  located at
+ *
+ *  http://aws.amazon.com/apache2.0
+ *
+ *  or in the "license" file accompanying this file. This file is distributed on
+ *  an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either
+ *  express or implied. See the License for the specific language governing
+ *  permissions and limitations under the License.
+ */
+
+package io.temporal.samples.taskinteraction;
+
+import io.temporal.workflow.SignalMethod;
+import io.temporal.workflow.WorkflowInterface;
+
+@WorkflowInterface
+public interface TaskClient {
+
+  @SignalMethod
+  void completeTaskByToken(String taskToken);
+}

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

@@ -0,0 +1,72 @@
+/*
+ *  Copyright (c) 2020 Temporal Technologies, Inc. All Rights Reserved
+ *
+ *  Copyright 2012-2016 Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ *
+ *  Modifications copyright (C) 2017 Uber Technologies, Inc.
+ *
+ *  Licensed under the Apache License, Version 2.0 (the "License"). You may not
+ *  use this file except in compliance with the License. A copy of the License is
+ *  located at
+ *
+ *  http://aws.amazon.com/apache2.0
+ *
+ *  or in the "license" file accompanying this file. This file is distributed on
+ *  an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either
+ *  express or implied. See the License for the specific language governing
+ *  permissions and limitations under the License.
+ */
+
+package io.temporal.samples.taskinteraction;
+
+import io.temporal.activity.ActivityOptions;
+import io.temporal.samples.taskinteraction.activity.ActivityTask;
+import io.temporal.workflow.CompletablePromise;
+import io.temporal.workflow.Workflow;
+import java.time.Duration;
+import java.util.HashMap;
+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 =
+      Workflow.newActivityStub(
+          ActivityTask.class,
+          ActivityOptions.newBuilder().setStartToCloseTimeout(Duration.ofSeconds(5)).build());
+
+  private final Map<String, CompletablePromise<R>> pendingPromises = new HashMap<>();
+  private final Logger logger = Workflow.getLogger(TaskService.class);
+  private final TaskClient listener =
+      taskToken -> {
+        logger.info("Completing task with token: " + taskToken);
+
+        final CompletablePromise<R> completablePromise = pendingPromises.get(taskToken);
+        completablePromise.complete(null);
+      };
+
+  public TaskService() {
+    Workflow.registerListener(listener);
+  }
+
+  public void executeTask(Task task) {
+
+    logger.info("Before creating task : " + task);
+    final String token = task.getToken();
+    activity.createTask(task);
+
+    logger.info("Task created: " + task);
+
+    final CompletablePromise<R> promise = Workflow.newPromise();
+    pendingPromises.put(token, promise);
+
+    // Wait promise to complete or fail
+    promise.get();
+
+    logger.info("Task completed: " + task);
+  }
+}

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

@@ -0,0 +1,41 @@
+/*
+ *  Copyright (c) 2020 Temporal Technologies, Inc. All Rights Reserved
+ *
+ *  Copyright 2012-2016 Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ *
+ *  Modifications copyright (C) 2017 Uber Technologies, Inc.
+ *
+ *  Licensed under the Apache License, Version 2.0 (the "License"). You may not
+ *  use this file except in compliance with the License. A copy of the License is
+ *  located at
+ *
+ *  http://aws.amazon.com/apache2.0
+ *
+ *  or in the "license" file accompanying this file. This file is distributed on
+ *  an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either
+ *  express or implied. See the License for the specific language governing
+ *  permissions and limitations under the License.
+ */
+
+package io.temporal.samples.taskinteraction;
+
+import io.temporal.workflow.*;
+import java.util.List;
+
+@WorkflowInterface
+public interface WorkflowTaskManager {
+
+  String WORKFLOW_ID = WorkflowTaskManager.class.getSimpleName();
+
+  @WorkflowMethod
+  void execute(List<Task> inputPendingTask, List<String> inputTaskToComplete);
+
+  @UpdateMethod
+  void createTask(Task task);
+
+  @UpdateMethod
+  void completeTaskByToken(String taskToken);
+
+  @QueryMethod
+  List<Task> getPendingTask();
+}

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

@@ -0,0 +1,104 @@
+/*
+ *  Copyright (c) 2020 Temporal Technologies, Inc. All Rights Reserved
+ *
+ *  Copyright 2012-2016 Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ *
+ *  Modifications copyright (C) 2017 Uber Technologies, Inc.
+ *
+ *  Licensed under the Apache License, Version 2.0 (the "License"). You may not
+ *  use this file except in compliance with the License. A copy of the License is
+ *  located at
+ *
+ *  http://aws.amazon.com/apache2.0
+ *
+ *  or in the "license" file accompanying this file. This file is distributed on
+ *  an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either
+ *  express or implied. See the License for the specific language governing
+ *  permissions and limitations under the License.
+ */
+
+package io.temporal.samples.taskinteraction;
+
+import io.temporal.workflow.Workflow;
+import java.util.*;
+
+public class WorkflowTaskManagerImpl implements WorkflowTaskManager {
+
+  private List<Task> pendingTask;
+
+  private List<String> tasksToComplete;
+
+  @Override
+  public void execute(List<Task> inputPendingTask, List<String> inputTaskToComplete) {
+    initPendingTasks(inputPendingTask);
+    initTaskToComplete(inputTaskToComplete);
+
+    while (true) {
+
+      Workflow.await(
+          () ->
+              // Wait until there are pending task to complete
+              !tasksToComplete.isEmpty());
+
+      final String taskToken = tasksToComplete.remove(0);
+
+      // Find the workflow id of the workflow we have to signal back
+      final String externalWorkflowId = new StringTokenizer(taskToken, "_").nextToken();
+
+      Workflow.newExternalWorkflowStub(TaskClient.class, externalWorkflowId)
+          .completeTaskByToken(taskToken);
+
+      final Task task = getPendingTaskWithToken(taskToken).get();
+      pendingTask.remove(task);
+
+      if (Workflow.getInfo().isContinueAsNewSuggested()) {
+        Workflow.newContinueAsNewStub(WorkflowTaskManager.class)
+            .execute(pendingTask, tasksToComplete);
+      }
+    }
+  }
+
+  @Override
+  public void createTask(Task task) {
+    initPendingTasks(new ArrayList<>());
+    pendingTask.add(task);
+  }
+
+  @Override
+  public void completeTaskByToken(String taskToken) {
+
+    tasksToComplete.add(taskToken);
+
+    Workflow.await(
+        () -> {
+          final boolean taskCompleted =
+              getPendingTask().stream().noneMatch((t) -> Objects.equals(t.getToken(), taskToken));
+
+          return taskCompleted;
+        });
+  }
+
+  @Override
+  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);
+  }
+}