[Fix #1211] Fixing cancel behaviour

Signed-off-by: fjtirado <ftirados@redhat.com>

Author: fjtirado

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

@@ -26,6 +26,8 @@
 import io.serverlessworkflow.impl.lifecycle.WorkflowStatusEvent;
 import io.serverlessworkflow.impl.lifecycle.WorkflowSuspendedEvent;
 import java.time.Instant;
+import java.util.ArrayList;
+import java.util.Collection;
 import java.util.Map;
 import java.util.Optional;
 import java.util.concurrent.CancellationException;
@@ -54,6 +56,8 @@ public class WorkflowMutableInstance implements WorkflowInstance {
   private Lock statusLock = new ReentrantLock();
   private Map<CompletableFuture<TaskContext>, TaskContext> suspended;
 
+  private Collection<CompletableFuture<?>> cancelables = new ArrayList<>();
+
   protected WorkflowMutableInstance(WorkflowDefinition definition, String id, WorkflowModel input) {
     this.id = id;
     this.input = input;
@@ -280,7 +284,8 @@ public boolean cancel() {
     try {
       statusLock.lock();
       if (TaskExecutorHelper.isActive(status.get())) {
-        futureRef.get().cancel(true);
+        cancelables.forEach(t -> t.cancel(true));
+        cancelables.clear();
         status(WorkflowStatus.CANCELLED);
         publishEvent(
             workflowContext,
@@ -294,6 +299,30 @@ public boolean cancel() {
     }
   }
 
+  public void addCancelable(CompletableFuture<?> cancelable) {
+    try {
+      statusLock.lock();
+      if (status.get() == WorkflowStatus.CANCELLED) {
+        statusLock.unlock();
+        cancelable.cancel(true);
+      } else {
+        cancelables.add(cancelable);
+        statusLock.unlock();
+        cancelable.thenAccept(
+            __ -> {
+              try {
+                statusLock.lock();
+                cancelables.remove(cancelable);
+              } finally {
+                statusLock.unlock();
+              }
+            });
+      }
+    } finally {
+      statusLock.unlock();
+    }
+  }
+
   public <T> T additionalObject(String key, Supplier<T> supplier) {
     return (T) additionalObjects.computeIfAbsent(key, k -> supplier.get());
   }

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

@@ -173,6 +173,7 @@ protected CompletableFuture<WorkflowModel> internalExecute(
                     processCe(converter.apply(ce), output, workflow, taskContext, future)),
             workflow,
             taskContext);
+    workflow.instance().addCancelable(info.completableFuture());
     return info.completableFuture()
         .whenComplete((__, e) -> info.registrations().forEach(eventConsumer::unregister))
         .thenApply(__ -> output);

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

@@ -69,8 +69,7 @@ void testEventListened(
 
   @ParameterizedTest
   @MethodSource("wrongEventListenerParameters")
-  void testWrongEvent(
-      String listen, String emit, JsonNode expectedResult, Object emitInput, Object listenInput)
+  void testWrongEvent(String listen, String emit, Object emitInput, Object listenInput)
       throws IOException {
     WorkflowDefinition listenDefinition =
         appl.workflowDefinition(WorkflowReader.readWorkflowFromClasspath(listen));
@@ -83,7 +82,7 @@ void testWrongEvent(
     assertThat(waitingInstance.status()).isEqualTo(WorkflowStatus.WAITING);
     waitingInstance.cancel();
     assertThat(waitingInstance.status()).isEqualTo(WorkflowStatus.CANCELLED);
-    assertThat(future).isCancelled();
+    assertThat(future).isDone();
   }
 
   @ParameterizedTest
@@ -140,7 +139,6 @@ private static Stream<Arguments> wrongEventListenerParameters() {
         Arguments.of(
             "workflows-samples/listen-to-any-filter.yaml",
             "workflows-samples/emit-doctor.yaml",
-            doctor(),
             Map.of("temperature", 38),
             Map.of("threshold", 39)));
   }

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

@@ -45,7 +45,6 @@
 import java.util.Map;
 import java.util.Objects;
 import java.util.Optional;
-import java.util.concurrent.CancellationException;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.CompletionException;
 import java.util.concurrent.CopyOnWriteArrayList;
@@ -169,7 +168,7 @@ void testCancel() throws IOException, InterruptedException {
             .instance(Map.of());
     CompletableFuture<WorkflowModel> future = instance.start();
     instance.cancel();
-    assertThat(catchThrowableOfType(CancellationException.class, () -> future.get().asMap()))
+    assertThat(catchThrowableOfType(ExecutionException.class, () -> future.get().asMap()))
         .isNotNull();
     assertThat(instance.status()).isEqualTo(WorkflowStatus.CANCELLED);
     assertPojoInCE("io.serverlessworkflow.task.cancelled.v1", TaskCancelledCEData.class);