[Fix #695] Suspending/cancelling/resume

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

Author: fjtirado

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

@@ -122,7 +122,6 @@ public WorkflowPosition position() {
     return position;
   }
 
-  @Override
   public Map<String, Object> variables() {
     return contextVariables;
   }
@@ -132,7 +131,6 @@ public Instant startedAt() {
     return startedAt;
   }
 
-  @Override
   public Optional<TaskContext> parent() {
     return parentContext;
   }

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

@@ -17,8 +17,6 @@
 
 import io.serverlessworkflow.api.types.TaskBase;
 import java.time.Instant;
-import java.util.Map;
-import java.util.Optional;
 
 public interface TaskContextData {
 
@@ -34,12 +32,8 @@ public interface TaskContextData {
 
   WorkflowPosition position();
 
-  Map<String, Object> variables();
-
   Instant startedAt();
 
-  Optional<TaskContext> parent();
-
   String taskName();
 
   Instant completedAt();

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

@@ -19,4 +19,10 @@
 
 public interface WorkflowInstance extends WorkflowInstanceData {
   CompletableFuture<WorkflowModel> start();
+
+  boolean suspend();
+
+  boolean cancel();
+
+  boolean resume();
 }

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

@@ -18,13 +18,18 @@
 import static io.serverlessworkflow.impl.lifecycle.LifecycleEventsUtils.publishEvent;
 
 import io.serverlessworkflow.impl.executors.TaskExecutorHelper;
+import io.serverlessworkflow.impl.lifecycle.WorkflowCancelledEvent;
 import io.serverlessworkflow.impl.lifecycle.WorkflowCompletedEvent;
 import io.serverlessworkflow.impl.lifecycle.WorkflowFailedEvent;
 import io.serverlessworkflow.impl.lifecycle.WorkflowStartedEvent;
 import java.time.Instant;
 import java.util.Optional;
+import java.util.concurrent.CancellationException;
 import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CompletionException;
 import java.util.concurrent.atomic.AtomicReference;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReentrantLock;
 
 public class WorkflowMutableInstance implements WorkflowInstance {
 
@@ -36,7 +41,12 @@ public class WorkflowMutableInstance implements WorkflowInstance {
   private Instant startedAt;
   private Instant completedAt;
   private volatile WorkflowModel output;
+  private Lock resumeLock = new ReentrantLock();
   private CompletableFuture<WorkflowModel> completableFuture;
+  private ResumeInfo resumeInfo;
+
+  private static record ResumeInfo(
+      CompletableFuture<TaskContext> incompletedFuture, TaskContext completedInfo) {}
 
   WorkflowMutableInstance(WorkflowDefinition definition, WorkflowModel input) {
     this.id = definition.application().idFactory().get();
@@ -70,7 +80,17 @@ public CompletableFuture<WorkflowModel> start() {
   private void whenFailed(WorkflowModel result, Throwable ex) {
     completedAt = Instant.now();
     if (ex != null) {
-      status.compareAndSet(WorkflowStatus.RUNNING, WorkflowStatus.FAULTED);
+      handleException(ex instanceof CompletionException ? ex = ex.getCause() : ex);
+    }
+  }
+
+  private void handleException(Throwable ex) {
+    if (ex instanceof CancellationException) {
+      status.set(WorkflowStatus.CANCELLED);
+      publishEvent(
+          workflowContext, l -> l.onWorkflowCancelled(new WorkflowCancelledEvent(workflowContext)));
+    } else {
+      status.set(WorkflowStatus.FAULTED);
       publishEvent(
           workflowContext, l -> l.onWorkflowFailed(new WorkflowFailedEvent(workflowContext, ex)));
     }
@@ -146,4 +166,56 @@ public String toString() {
         + completedAt
         + "]";
   }
+
+  @Override
+  public boolean suspend() {
+    try {
+      resumeLock.lock();
+      WorkflowStatus currentStatus = status.get();
+      if (currentStatus == WorkflowStatus.RUNNING || currentStatus == WorkflowStatus.WAITING) {
+        status.set(WorkflowStatus.SUSPENDED);
+        return true;
+      } else {
+        return false;
+      }
+    } finally {
+      resumeLock.unlock();
+    }
+  }
+
+  @Override
+  public boolean resume() {
+    try {
+      resumeLock.lock();
+      if (resumeInfo != null) {
+        resumeInfo.incompletedFuture().complete(resumeInfo.completedInfo());
+        resumeInfo = null;
+        return true;
+      } else {
+        return false;
+      }
+    } finally {
+      resumeLock.unlock();
+    }
+  }
+
+  public CompletableFuture<TaskContext> checkSuspended(TaskContext t) {
+    try {
+      resumeLock.lock();
+      if (status.get() == WorkflowStatus.SUSPENDED) {
+        CompletableFuture<TaskContext> incompletedFuture = new CompletableFuture<>();
+        resumeInfo = new ResumeInfo(incompletedFuture, t);
+        return incompletedFuture;
+      } else {
+        return CompletableFuture.completedFuture(t);
+      }
+    } finally {
+      resumeLock.unlock();
+    }
+  }
+
+  @Override
+  public boolean cancel() {
+    return completableFuture.cancel(true);
+  }
 }

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

@@ -21,5 +21,6 @@ public enum WorkflowStatus {
   WAITING,
   COMPLETED,
   FAULTED,
-  CANCELLED
+  CANCELLED,
+  SUSPENDED
 }

impl/core/src/main/java/io/serverlessworkflow/impl/executors/AbstractTaskExecutor.java

@@ -34,6 +34,7 @@
 import io.serverlessworkflow.impl.WorkflowPosition;
 import io.serverlessworkflow.impl.WorkflowPredicate;
 import io.serverlessworkflow.impl.WorkflowStatus;
+import io.serverlessworkflow.impl.lifecycle.TaskCancelledEvent;
 import io.serverlessworkflow.impl.lifecycle.TaskCompletedEvent;
 import io.serverlessworkflow.impl.lifecycle.TaskFailedEvent;
 import io.serverlessworkflow.impl.lifecycle.TaskStartedEvent;
@@ -43,7 +44,9 @@
 import java.util.Iterator;
 import java.util.Map;
 import java.util.Optional;
+import java.util.concurrent.CancellationException;
 import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CompletionException;
 
 public abstract class AbstractTaskExecutor<T extends TaskBase> implements TaskExecutor<T> {
 
@@ -204,12 +207,13 @@ public CompletableFuture<TaskContext> apply(
               .whenComplete(
                   (t, e) -> {
                     if (e != null) {
-                      publishEvent(
+                      handleException(
                           workflowContext,
-                          l ->
-                              l.onTaskFailed(new TaskFailedEvent(workflowContext, taskContext, e)));
+                          taskContext,
+                          e instanceof CompletionException ? e.getCause() : e);
                     }
                   })
+              .thenCompose(t -> workflowContext.instance().checkSuspended(t))
               .thenApply(
                   t -> {
                     outputProcessor.ifPresent(
@@ -235,6 +239,19 @@ public CompletableFuture<TaskContext> apply(
     }
   }
 
+  private void handleException(
+      WorkflowContext workflowContext, TaskContext taskContext, Throwable e) {
+    if (e instanceof CancellationException) {
+      publishEvent(
+          workflowContext,
+          l -> l.onTaskCancelled(new TaskCancelledEvent(workflowContext, taskContext)));
+    } else {
+      publishEvent(
+          workflowContext,
+          l -> l.onTaskFailed(new TaskFailedEvent(workflowContext, taskContext, e)));
+    }
+  }
+
   protected abstract TransitionInfo getSkipTransition();
 
   protected abstract CompletableFuture<TaskContext> execute(

impl/core/src/main/java/io/serverlessworkflow/impl/lifecycle/ce/WorkflowErrorCEData.java

@@ -21,7 +21,6 @@
 import io.serverlessworkflow.impl.lifecycle.WorkflowFailedEvent;
 import java.io.PrintWriter;
 import java.io.StringWriter;
-import java.util.concurrent.CompletionException;
 
 public record WorkflowErrorCEData(
     String type, Integer status, String instance, String title, String detail) {
@@ -35,10 +34,6 @@ public static WorkflowErrorCEData error(WorkflowFailedEvent ev) {
   }
 
   private static WorkflowErrorCEData error(Throwable cause) {
-
-    if (cause instanceof CompletionException) {
-      cause = cause.getCause();
-    }
     return cause instanceof WorkflowException ex ? error(ex) : commonError(cause);
   }