Workflow Implementation - Continues...

examples/src/main/java/io/dapr/examples/workflows/DemoActivityInput.java

@@ -0,0 +1,34 @@
+/*
+ * Copyright 2023 The Dapr Authors
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License 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.dapr.examples.workflows;
+
+public class DemoActivityInput {
+
+  private String message;
+
+  public DemoActivityInput() {
+  }
+
+  public DemoActivityInput(String message) {
+    this.message = message;
+  }
+
+  public String getMessage() {
+    return message;
+  }
+
+  public void setMessage(String message) {
+    this.message = message;
+  }
+}

examples/src/main/java/io/dapr/examples/workflows/DemoActivityOutput.java

@@ -0,0 +1,44 @@
+/*
+ * Copyright 2023 The Dapr Authors
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License 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.dapr.examples.workflows;
+
+public class DemoActivityOutput {
+
+  private String originalMessage;
+  private String newMessage;
+
+  public DemoActivityOutput() {
+  }
+
+  public DemoActivityOutput(String originalMessage, String newMessage) {
+    this.originalMessage = originalMessage;
+    this.newMessage = newMessage;
+  }
+
+  public String getOriginalMessage() {
+    return originalMessage;
+  }
+
+  public void setOriginalMessage(String originalMessage) {
+    this.originalMessage = originalMessage;
+  }
+
+  public String getNewMessage() {
+    return newMessage;
+  }
+
+  public void setNewMessage(String newMessage) {
+    this.newMessage = newMessage;
+  }
+}

examples/src/main/java/io/dapr/examples/workflows/DemoSubWorkflow.java

@@ -0,0 +1,39 @@
+/*
+ * Copyright 2023 The Dapr Authors
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License 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.dapr.examples.workflows;
+
+import io.dapr.workflows.Workflow;
+import io.dapr.workflows.WorkflowStub;
+
+/**
+ * Implementation of the DemoWorkflow for the server side.
+ */
+public class DemoSubWorkflow extends Workflow {
+  @Override
+  public WorkflowStub create() {
+    return ctx -> {
+
+      var logger = ctx.getLogger();
+      logger.info("Child-Workflow> Started: " + ctx.getName());
+      logger.info("Child-Workflow> Instance ID: " + ctx.getInstanceId());
+      logger.info("Child-Workflow> Current Time: " + ctx.getCurrentInstant());
+
+      var input = ctx.getInput(String.class);
+      logger.info("Child-Workflow> Input: " + input);
+
+      logger.info("Child-Workflow> Completed");
+      ctx.complete("result: " + input);
+    };
+  }
+}

examples/src/main/java/io/dapr/examples/workflows/DemoWorkflow.java

@@ -13,31 +13,114 @@
 
 package io.dapr.examples.workflows;
 
+import com.microsoft.durabletask.CompositeTaskFailedException;
+import com.microsoft.durabletask.Task;
 import com.microsoft.durabletask.TaskCanceledException;
 import io.dapr.workflows.Workflow;
 import io.dapr.workflows.WorkflowStub;
 
 import java.time.Duration;
+import java.util.Arrays;
+import java.util.List;
 
 /**
  * Implementation of the DemoWorkflow for the server side.
  */
 public class DemoWorkflow extends Workflow {
-
   @Override
   public WorkflowStub create() {
     return ctx -> {
       ctx.getLogger().info("Starting Workflow: " + ctx.getName());
       ctx.getLogger().info("Instance ID: " + ctx.getInstanceId());
-      ctx.getLogger().info("Waiting for event: 'myEvent'...");
+      ctx.getLogger().info("Current Orchestration Time: " + ctx.getCurrentInstant());
+      ctx.getLogger().info("Waiting for event: 'TimedOutEvent'...");
+      try {
+        ctx.waitForExternalEvent("TimedOutEvent", Duration.ofSeconds(10)).await();
+      } catch (TaskCanceledException e) {
+        ctx.getLogger().warn("Timed out");
+        ctx.getLogger().warn(e.getMessage());
+      }
+
+      ctx.getLogger().info("Waiting for event: 'TestEvent'...");
+      try {
+        ctx.waitForExternalEvent("TestEvent", Duration.ofSeconds(10)).await();
+        ctx.getLogger().info("Received TestEvent");
+      } catch (TaskCanceledException e) {
+        ctx.getLogger().warn("Timed out");
+        ctx.getLogger().warn(e.getMessage());
+      }
+
+      ctx.getLogger().info("Parallel Execution - Waiting for all tasks to finish...");
+      try {
+        Task<String> t1 = ctx.waitForExternalEvent("event1", Duration.ofSeconds(5), String.class);
+        Task<String> t2 = ctx.waitForExternalEvent("event2", Duration.ofSeconds(5), String.class);
+        Task<String> t3 = ctx.waitForExternalEvent("event3", Duration.ofSeconds(5), String.class);
+
+        List<String> results = ctx.allOf(Arrays.asList(t1, t2, t3)).await();
+        results.forEach(t -> ctx.getLogger().info("finished task: " + t));
+        ctx.getLogger().info("All tasks finished!");
+
+      } catch (CompositeTaskFailedException e) {
+        ctx.getLogger().warn(e.getMessage());
+        List<Exception> exceptions = e.getExceptions();
+        exceptions.forEach(ex -> ctx.getLogger().warn(ex.getMessage()));
+      }
+
+      ctx.getLogger().info("Parallel Execution - Waiting for any task to finish...");
       try {
-        ctx.waitForExternalEvent("myEvent", Duration.ofSeconds(10)).await();
-        ctx.getLogger().info("Received!");
+        Task<String> e1 = ctx.waitForExternalEvent("e1", Duration.ofSeconds(5), String.class);
+        Task<String> e2 = ctx.waitForExternalEvent("e2", Duration.ofSeconds(5), String.class);
+        Task<String> e3 = ctx.waitForExternalEvent("e3", Duration.ofSeconds(5), String.class);
+        Task<Void> timeoutTask = ctx.createTimer(Duration.ofSeconds(1));
+
+        Task<?> winner = ctx.anyOf(Arrays.asList(e1, e2, e3, timeoutTask)).await();
+        if (winner == timeoutTask) {
+          ctx.getLogger().info("All tasks timed out!");
+        } else {
+          ctx.getLogger().info("One of the tasks finished!");
+        }
       } catch (TaskCanceledException e) {
         ctx.getLogger().warn("Timed out");
         ctx.getLogger().warn(e.getMessage());
       }
-      ctx.complete("finished");
+
+      ctx.getLogger().info("Calling Activity...");
+      var input = new DemoActivityInput("Hello Activity!");
+      var output = ctx.callActivity(DemoWorkflowActivity.class.getName(), input, DemoActivityOutput.class).await();
+
+      ctx.getLogger().info("Activity returned: " + output);
+      ctx.getLogger().info("Activity returned: " + output.getNewMessage());
+      ctx.getLogger().info("Activity returned: " + output.getOriginalMessage());
+
+
+      boolean shouldComplete = true;
+      ctx.getLogger().info("Waiting for event: 'RestartEvent'...");
+      try {
+        ctx.waitForExternalEvent("RestartEvent", Duration.ofSeconds(10)).await();
+        ctx.getLogger().info("Received RestartEvent");
+        ctx.getLogger().info("Restarting Workflow by calling continueAsNew...");
+        ctx.continueAsNew("TestInputRestart", false);
+        shouldComplete = false;
+      } catch (TaskCanceledException e) {
+        ctx.getLogger().warn("Restart Timed out");
+        ctx.getLogger().warn(e.getMessage());
+      }
+
+      if (shouldComplete) {
+        ctx.getLogger().info("Child-Workflow> Calling ChildWorkflow...");
+        var childWorkflowInput = "Hello ChildWorkflow!";
+        var childWorkflowOutput =
+            ctx.callSubWorkflow(DemoSubWorkflow.class.getName(), childWorkflowInput, String.class).await();
+
+        ctx.getLogger().info("Child-Workflow> returned: " + childWorkflowOutput);
+
+        ctx.getLogger().info("Workflow finished");
+        ctx.complete("finished");
+
+        return;
+      }
+
+      ctx.getLogger().info("Workflow restarted");
     };
   }
 }

examples/src/main/java/io/dapr/examples/workflows/DemoWorkflowActivity.java

@@ -0,0 +1,53 @@
+/*
+ * Copyright 2023 The Dapr Authors
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License 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.dapr.examples.workflows;
+
+import com.fasterxml.jackson.annotation.JsonAutoDetect;
+import io.dapr.workflows.runtime.WorkflowActivity;
+import io.dapr.workflows.runtime.WorkflowActivityContext;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.concurrent.TimeUnit;
+
+@JsonAutoDetect(fieldVisibility = JsonAutoDetect.Visibility.ANY)
+public class DemoWorkflowActivity implements WorkflowActivity {
+
+  @Override
+  public DemoActivityOutput run(WorkflowActivityContext ctx) {
+    Logger logger = LoggerFactory.getLogger(DemoWorkflowActivity.class);
+    logger.info("Starting Activity: " + ctx.getName());
+
+    var message = ctx.getInput(DemoActivityInput.class).getMessage();
+    var newMessage = message + " World!, from Activity";
+    logger.info("Message Received from input: " + message);
+    logger.info("Sending message to output: " + newMessage);
+
+    logger.info("Sleeping for 5 seconds to simulate long running operation...");
+
+    try {
+      TimeUnit.SECONDS.sleep(5);
+    } catch (InterruptedException e) {
+      throw new RuntimeException(e);
+    }
+
+
+    logger.info("Activity finished");
+
+    var output = new DemoActivityOutput(message, newMessage);
+    logger.info("Activity returned: " + output);
+
+    return output;
+  }
+}