[Fix #782] Added suspend support

Signed-off-by: fjtirado <[email protected]>

Author: fjtirado

impl/core/src/main/java/io/serverlessworkflow/impl/WorkflowMutableInstance.java

@@ -147,17 +147,21 @@ public WorkflowStatus status() {
 
   @Override
   public WorkflowModel output() {
-    return futureRef.get().join();
+    CompletableFuture<WorkflowModel> future = futureRef.get();
+    return future != null ? future.join() : null;
   }
 
   @Override
   public <T> T outputAs(Class<T> clazz) {
-    return output()
-        .as(clazz)
-        .orElseThrow(
-            () ->
-                new IllegalArgumentException(
-                    "Output " + output() + " cannot be converted to class " + clazz));
+    WorkflowModel output = output();
+    return output != null
+        ? output
+            .as(clazz)
+            .orElseThrow(
+                () ->
+                    new IllegalArgumentException(
+                        "Output " + output + " cannot be converted to class " + clazz))
+        : null;
   }
 
   public void status(WorkflowStatus state) {
@@ -182,8 +186,7 @@ public boolean suspend() {
     try {
       statusLock.lock();
       if (TaskExecutorHelper.isActive(status.get()) && suspended == null) {
-        suspended = new ConcurrentHashMap<>();
-        status.set(WorkflowStatus.SUSPENDED);
+        internalSuspend();
         publishEvent(
             workflowContext,
             l -> l.onWorkflowSuspended(new WorkflowSuspendedEvent(workflowContext)));
@@ -196,6 +199,11 @@ public boolean suspend() {
     }
   }
 
+  protected final void internalSuspend() {
+    suspended = new ConcurrentHashMap<>();
+    status.set(WorkflowStatus.SUSPENDED);
+  }
+
   @Override
   public boolean resume() {
     try {

impl/persistence/api/src/main/java/io/serverlessworkflow/impl/persistence/WorkflowPersistenceInstance.java

@@ -20,6 +20,7 @@
 import io.serverlessworkflow.impl.WorkflowDefinition;
 import io.serverlessworkflow.impl.WorkflowModel;
 import io.serverlessworkflow.impl.WorkflowMutableInstance;
+import io.serverlessworkflow.impl.WorkflowStatus;
 import io.serverlessworkflow.impl.executors.TransitionInfo;
 import java.util.concurrent.CompletableFuture;
 
@@ -37,6 +38,9 @@ public CompletableFuture<WorkflowModel> start() {
     return startExecution(
         () -> {
           startedAt = info.startedAt();
+          if (info.status() == WorkflowStatus.SUSPENDED) {
+            internalSuspend();
+          }
         });
   }
 

impl/test/test.db renamed to impl/test/db-samples/running.db

@@ -19,6 +19,7 @@
 
 import io.serverlessworkflow.impl.WorkflowApplication;
 import io.serverlessworkflow.impl.WorkflowDefinition;
+import io.serverlessworkflow.impl.WorkflowInstance;
 import io.serverlessworkflow.impl.persistence.PersistenceApplicationBuilder;
 import io.serverlessworkflow.impl.persistence.PersistenceInstanceHandlers;
 import io.serverlessworkflow.impl.persistence.bigmap.BytesMapPersistenceInstanceHandlers;
@@ -31,9 +32,14 @@
 public class DBGenerator {
 
   public static void main(String[] args) throws IOException {
-    Files.deleteIfExists(Path.of("test.db"));
+    runInstance("db-samples/running.db", false);
+    runInstance("db-samples/suspended.db", true);
+  }
+
+  private static void runInstance(String dbName, boolean suspend) throws IOException {
+    Files.deleteIfExists(Path.of(dbName));
     try (PersistenceInstanceHandlers factories =
-            BytesMapPersistenceInstanceHandlers.builder(new MVStorePersistenceStore("test.db"))
+            BytesMapPersistenceInstanceHandlers.builder(new MVStorePersistenceStore(dbName))
                 .build();
         WorkflowApplication application =
             PersistenceApplicationBuilder.builder(
@@ -43,7 +49,11 @@ public static void main(String[] args) throws IOException {
       WorkflowDefinition definition =
           application.workflowDefinition(
               readWorkflowFromClasspath("workflows-samples/set-listen-to-any.yaml"));
-      definition.instance(Map.of()).start();
+      WorkflowInstance instance = definition.instance(Map.of());
+      instance.start();
+      if (suspend) {
+        instance.suspend();
+      }
     }
   }
 }

impl/test/db-samples/suspended.db

@@ -44,15 +44,24 @@ void testWaitingInstance() throws IOException {
 
       WorkflowInstance instance = definition.instance(Map.of());
       instance.start();
-      assertThat(taskCounter.taskCounter(instance.id()).orElseThrow().completed()).isEqualTo(1);
+      assertThat(taskCounter.taskCounter(instance.id()).completed()).isEqualTo(1);
     }
   }
 
   @Test
   void testRestoreWaitingInstance() throws IOException {
+    runIt("db-samples/running.db", WorkflowStatus.WAITING);
+  }
+
+  @Test
+  void testRestoreSuspendedInstance() throws IOException {
+    runIt("db-samples/suspended.db", WorkflowStatus.SUSPENDED);
+  }
+
+  private void runIt(String dbName, WorkflowStatus expectedStatus) throws IOException {
     TaskCounterPerInstanceListener taskCounter = new TaskCounterPerInstanceListener();
     try (PersistenceInstanceHandlers handlers =
-            BytesMapPersistenceInstanceHandlers.builder(new MVStorePersistenceStore("test.db"))
+            BytesMapPersistenceInstanceHandlers.builder(new MVStorePersistenceStore(dbName))
                 .build();
         WorkflowApplication application =
             PersistenceApplicationBuilder.builder(
@@ -71,9 +80,8 @@ void testRestoreWaitingInstance() throws IOException {
           .singleElement()
           .satisfies(
               instance -> {
-                assertThat(instance.status()).isEqualTo(WorkflowStatus.WAITING);
-                assertThat(taskCounter.taskCounter(instance.id()).orElseThrow().completed())
-                    .isEqualTo(0);
+                assertThat(instance.status()).isEqualTo(expectedStatus);
+                assertThat(taskCounter.taskCounter(instance.id()).completed()).isEqualTo(0);
               });
     }
   }

impl/test/src/test/java/io/serverlessworkflow/impl/test/DBGenerator.java

@@ -20,7 +20,6 @@
 import io.serverlessworkflow.impl.lifecycle.WorkflowEvent;
 import io.serverlessworkflow.impl.lifecycle.WorkflowExecutionListener;
 import java.util.Map;
-import java.util.Optional;
 import java.util.concurrent.ConcurrentHashMap;
 
 public class TaskCounterPerInstanceListener implements WorkflowExecutionListener {
@@ -53,12 +52,11 @@ public void onTaskStarted(TaskStartedEvent ev) {
   }
 
   private TaskCounter taskCounter(WorkflowEvent ev) {
-    return taskCounter.computeIfAbsent(
-        ev.workflowContext().instanceData().id(), k -> new TaskCounter());
+    return taskCounter(ev.workflowContext().instanceData().id());
   }
 
-  public Optional<TaskCounter> taskCounter(String instanceId) {
-    return Optional.ofNullable(taskCounter.get(instanceId));
+  public TaskCounter taskCounter(String instanceId) {
+    return taskCounter.computeIfAbsent(instanceId, k -> new TaskCounter());
   }
 
   public void onTaskCompleted(TaskCompletedEvent ev) {