[Fix #1657] Integrate status change events with others

The idea is to not continue workflow execution till the listeners are
completed.

Signed-off-by: Francisco Javier Tirado Sarti <ftirados@ibm.com>

Author: fjtirado

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

@@ -17,23 +17,12 @@
 
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
-import java.util.concurrent.locks.Lock;
-import java.util.concurrent.locks.ReentrantLock;
 
 public class DefaultExecutorServiceFactory extends AbstractExecutorServiceHolder {
-
-  private Lock serviceLock = new ReentrantLock();
+  private ExecutorService service = Executors.newCachedThreadPool();
 
   @Override
   public ExecutorService get() {
-    try {
-      serviceLock.lock();
-      if (service == null) {
-        service = Executors.newCachedThreadPool();
-      }
-    } finally {
-      serviceLock.unlock();
-    }
     return service;
   }
 }

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

@@ -254,7 +254,7 @@ public SchemaValidator getValidator(SchemaInline inline) {
     private WorkflowPositionFactory positionFactory = () -> new QueueWorkflowPosition();
     private WorkflowInstanceIdFactory idFactory;
     private WorkflowScheduler scheduler;
-    private ExecutorServiceFactory executorFactory = new DefaultExecutorServiceFactory();
+    private ExecutorServiceFactory executorFactory;
     private EventConsumer<?, ?> eventConsumer;
     private Collection<EventPublisher> eventPublishers = new ArrayList<>();
     private RuntimeDescriptorFactory descriptorFactory =
@@ -458,7 +458,9 @@ public Builder withCronResolverFactory(CronResolverFactory cronResolverFactory)
     }
 
     public WorkflowApplication build() {
-
+      if (executorFactory == null) {
+        executorFactory = new DefaultExecutorServiceFactory();
+      }
       if (modelFactory == null) {
         modelFactory =
             loadFirst(WorkflowModelFactory.class)

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

@@ -73,6 +73,7 @@ protected WorkflowMutableInstance(WorkflowDefinition definition, String id, Work
   public CompletableFuture<WorkflowModel> start() {
     return startExecution(
         () -> {
+          status(WorkflowStatus.RUNNING);
           startedAt = Instant.now();
           return publishEvent(
               workflowContext, l -> l.onWorkflowStarted(new WorkflowStartedEvent(workflowContext)));
@@ -85,8 +86,6 @@ protected final CompletableFuture<WorkflowModel> startExecution(
     if (future != null) {
       return future;
     }
-    status(WorkflowStatus.RUNNING);
-
     future =
         runnable
             .get()
@@ -103,20 +102,23 @@ protected final CompletableFuture<WorkflowModel> startExecution(
                                 .orElse(input))
                         .whenComplete(this::setCompleteDate)
                         .thenApply(this::filterAndValidate)
-                        .whenComplete(this::handleException)
-                        .thenCompose(
-                            model ->
-                                publishEvent(
-                                        workflowContext,
-                                        l ->
-                                            l.onWorkflowCompleted(
-                                                new WorkflowCompletedEvent(workflowContext, model)))
-                                    .thenApply(__ -> model)))
+                        .thenCompose(this::publishEvents)
+                        .exceptionallyCompose(this::handleException))
             .whenComplete(this::cleanUp);
     futureRef.set(future);
     return future;
   }
 
+  private CompletableFuture<WorkflowModel> publishEvents(WorkflowModel model) {
+    return status(WorkflowStatus.COMPLETED)
+        .thenCompose(
+            __ ->
+                publishEvent(
+                    workflowContext,
+                    l -> l.onWorkflowCompleted(new WorkflowCompletedEvent(workflowContext, model))))
+        .thenApply(__ -> model);
+  }
+
   private void setCompleteDate(WorkflowModel result, Throwable ex) {
     completedAt = Instant.now();
   }
@@ -130,19 +132,19 @@ private void cleanUp(WorkflowModel result, Throwable ex) {
     workflowContext.definition().removeInstance(this);
   }
 
-  private void handleException(WorkflowModel result, Throwable exception) {
-    if (exception != null) {
-      final Throwable cause =
-          exception instanceof CompletionException ? exception.getCause() : exception;
-      if (!(cause instanceof CancellationException)) {
-        status(WorkflowStatus.FAULTED);
-        publishEvent(
-            workflowContext,
-            l -> l.onWorkflowFailed(new WorkflowFailedEvent(workflowContext, cause)));
-      }
-    } else {
-      status(WorkflowStatus.COMPLETED);
+  private CompletableFuture<WorkflowModel> handleException(Throwable exception) {
+    final Throwable cause =
+        exception instanceof CompletionException ? exception.getCause() : exception;
+    if (!(cause instanceof CancellationException)) {
+      return status(WorkflowStatus.FAULTED)
+          .thenCompose(
+              __ ->
+                  publishEvent(
+                      workflowContext,
+                      l -> l.onWorkflowFailed(new WorkflowFailedEvent(workflowContext, cause))))
+          .thenCompose(__ -> CompletableFuture.failedFuture(exception));
     }
+    return CompletableFuture.failedFuture(exception);
   }
 
   private WorkflowModel filterAndValidate(WorkflowModel model) {
@@ -208,15 +210,19 @@ public <T> T outputAs(Class<T> clazz) {
         : null;
   }
 
-  public void status(WorkflowStatus state) {
+  public CompletableFuture<?> status(WorkflowStatus state) {
     WorkflowStatus prevState = this.status.getAndSet(state);
-    if (prevState != state) {
-      publishEvent(
-          workflowContext,
-          l ->
-              l.onWorkflowStatusChanged(
-                  new WorkflowStatusEvent(workflowContext, prevState, state)));
-    }
+    return prevState != state
+        ? publishEvent(
+            workflowContext,
+            l ->
+                l.onWorkflowStatusChanged(
+                    new WorkflowStatusEvent(workflowContext, prevState, state)))
+        : CompletableFuture.completedFuture(null);
+  }
+
+  protected final void setStatus(WorkflowStatus state) {
+    this.status.set(state);
   }
 
   @Override
@@ -234,8 +240,9 @@ public String toString() {
 
   @Override
   public boolean suspend() {
-    boolean result = _suspend();
+    boolean result = internalSuspend();
     if (result) {
+      status(WorkflowStatus.SUSPENDED);
       publishEvent(
           workflowContext, l -> l.onWorkflowSuspended(new WorkflowSuspendedEvent(workflowContext)));
     }
@@ -244,19 +251,22 @@ public boolean suspend() {
 
   @Override
   public CompletableFuture<Boolean> suspendFuture() {
-    return _suspend()
-        ? publishEvent(
-                workflowContext,
-                l -> l.onWorkflowSuspended(new WorkflowSuspendedEvent(workflowContext)))
+    return internalSuspend()
+        ? status(WorkflowStatus.SUSPENDED)
+            .thenCompose(
+                __ ->
+                    publishEvent(
+                        workflowContext,
+                        l -> l.onWorkflowSuspended(new WorkflowSuspendedEvent(workflowContext))))
             .thenApply(__ -> true)
         : CompletableFuture.completedFuture(false);
   }
 
-  private boolean _suspend() {
+  private boolean internalSuspend() {
     try {
       statusLock.lock();
       if (TaskExecutorHelper.isActive(status.get()) && suspended == null) {
-        internalSuspend();
+        setSuspended();
         return true;
       } else {
         return false;
@@ -266,14 +276,13 @@ private boolean _suspend() {
     }
   }
 
-  protected final void internalSuspend() {
+  protected final void setSuspended() {
     suspended = new ConcurrentHashMap<>();
-    status(WorkflowStatus.SUSPENDED);
   }
 
   @Override
   public boolean resume() {
-    boolean result = _resume();
+    boolean result = internalResume();
     if (result) {
       publishEvent(
           workflowContext, l -> l.onWorkflowResumed(new WorkflowResumedEvent(workflowContext)));
@@ -283,15 +292,15 @@ public boolean resume() {
 
   @Override
   public CompletableFuture<Boolean> resumeFuture() {
-    return _resume()
+    return internalResume()
         ? publishEvent(
                 workflowContext,
                 l -> l.onWorkflowResumed(new WorkflowResumedEvent(workflowContext)))
             .thenApply(__ -> true)
         : CompletableFuture.completedFuture(false);
   }
 
-  private boolean _resume() {
+  private boolean internalResume() {
     boolean result;
     try {
       statusLock.lock();
@@ -327,25 +336,28 @@ public CompletableFuture<TaskContext> cancelCheck(TaskContext t) {
   }
 
   public CompletableFuture<TaskContext> suspendedCheck(TaskContext t) {
+    boolean isActive;
     try {
       statusLock.lock();
       if (suspended != null) {
         CompletableFuture<TaskContext> suspendedTask = new CompletableFuture<TaskContext>();
         suspended.put(suspendedTask, t);
         return suspendedTask;
-      } else if (TaskExecutorHelper.isActive(status.get())) {
-        status(WorkflowStatus.RUNNING);
       }
+      isActive = TaskExecutorHelper.isActive(status.get());
     } finally {
       statusLock.unlock();
     }
-    return CompletableFuture.completedFuture(t);
+    return isActive
+        ? status(WorkflowStatus.RUNNING).thenApply(__ -> t)
+        : CompletableFuture.completedFuture(t);
   }
 
   @Override
   public boolean cancel() {
-    boolean result = _cancel();
+    boolean result = internalCancel();
     if (result) {
+      status(WorkflowStatus.CANCELLED);
       publishEvent(
           workflowContext, l -> l.onWorkflowCancelled(new WorkflowCancelledEvent(workflowContext)));
     }
@@ -354,23 +366,25 @@ public boolean cancel() {
 
   @Override
   public CompletableFuture<Boolean> cancelFuture() {
-    return _cancel()
-        ? publishEvent(
-                workflowContext,
-                l -> l.onWorkflowCancelled(new WorkflowCancelledEvent(workflowContext)))
+    return internalCancel()
+        ? status(WorkflowStatus.CANCELLED)
+            .thenCompose(
+                __ ->
+                    publishEvent(
+                        workflowContext,
+                        l -> l.onWorkflowCancelled(new WorkflowCancelledEvent(workflowContext))))
             .thenApply(__ -> true)
         : CompletableFuture.completedFuture(false);
   }
 
-  private boolean _cancel() {
+  private boolean internalCancel() {
     boolean result;
     Collection<CompletableFuture<?>> toCancel = null;
     try {
       statusLock.lock();
       if (TaskExecutorHelper.isActive(status.get())) {
         toCancel = new ArrayList<>(cancelables);
         cancelables.clear();
-        status(WorkflowStatus.CANCELLED);
         result = true;
       } else {
         result = false;

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

@@ -193,7 +193,7 @@ private CompletableFuture<TaskContext> executeNext(
       WorkflowContext workflow, TaskContext taskContext) {
     TransitionInfo transition = taskContext.transition();
     if (transition.isEndNode()) {
-      workflow.instance().status(WorkflowStatus.COMPLETED);
+      return workflow.instance().status(WorkflowStatus.COMPLETED).thenApply(__ -> taskContext);
     } else if (transition.next() != null) {
       return transition.next().apply(workflow, taskContext.parent(), taskContext.output());
     }
@@ -241,15 +241,6 @@ public CompletableFuture<TaskContext> apply(
                   })
               .thenCompose(t -> execute(workflowContext, t))
               .thenCompose(workflowContext.instance()::cancelCheck)
-              .whenComplete(
-                  (t, e) -> {
-                    if (e != null) {
-                      handleException(
-                          workflowContext,
-                          taskContext,
-                          e instanceof CompletionException ? e.getCause() : e);
-                    }
-                  })
               .thenApply(
                   t -> {
                     outputProcessor.ifPresent(
@@ -272,7 +263,13 @@ public CompletableFuture<TaskContext> apply(
                               l ->
                                   l.onTaskCompleted(
                                       new TaskCompletedEvent(workflowContext, taskContext)))
-                          .thenApply(__ -> t));
+                          .thenApply(__ -> t))
+              .exceptionallyCompose(
+                  e ->
+                      handleException(
+                          workflowContext,
+                          taskContext,
+                          e instanceof CompletionException ? e.getCause() : e));
       if (timeout.isPresent()) {
         completable =
             completable
@@ -298,17 +295,21 @@ public CompletableFuture<TaskContext> apply(
     }
   }
 
-  private void handleException(
+  private CompletableFuture<TaskContext> handleException(
       WorkflowContext workflowContext, TaskContext taskContext, Throwable e) {
+    CompletableFuture<?> events;
     if (e instanceof CancellationException) {
-      publishEvent(
-          workflowContext,
-          l -> l.onTaskCancelled(new TaskCancelledEvent(workflowContext, taskContext)));
+      events =
+          publishEvent(
+              workflowContext,
+              l -> l.onTaskCancelled(new TaskCancelledEvent(workflowContext, taskContext)));
     } else {
-      publishEvent(
-          workflowContext,
-          l -> l.onTaskFailed(new TaskFailedEvent(workflowContext, taskContext, e)));
+      events =
+          publishEvent(
+              workflowContext,
+              l -> l.onTaskFailed(new TaskFailedEvent(workflowContext, taskContext, e)));
     }
+    return events.thenCompose(__ -> CompletableFuture.failedFuture(e));
   }
 
   public WorkflowPosition position() {