[Fix #782] More changes after testing

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

Author: fjtirado

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

@@ -62,7 +62,6 @@ public CompletableFuture<WorkflowModel> start() {
     return startExecution(
         () -> {
           startedAt = Instant.now();
-          status.set(WorkflowStatus.RUNNING);
           publishEvent(
               workflowContext, l -> l.onWorkflowStarted(new WorkflowStartedEvent(workflowContext)));
         });
@@ -73,6 +72,7 @@ protected final CompletableFuture<WorkflowModel> startExecution(Runnable runnabl
     if (future != null) {
       return future;
     }
+    status.set(WorkflowStatus.RUNNING);
     runnable.run();
     future =
         TaskExecutorHelper.processTaskList(

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

@@ -194,10 +194,11 @@ public CompletableFuture<TaskContext> apply(
     TaskContext taskContext = new TaskContext(input, position, parentContext, taskName, task);
     workflowContext.instance().restoreContext(workflowContext, taskContext);
     CompletableFuture<TaskContext> completable = CompletableFuture.completedFuture(taskContext);
-    if (taskContext.isCompleted() && !TaskExecutorHelper.isActive(workflowContext)) {
+    if (!TaskExecutorHelper.isActive(workflowContext)) {
       return completable;
-    }
-    if (ifFilter.map(f -> f.test(workflowContext, taskContext, input)).orElse(true)) {
+    } else if (taskContext.isCompleted()) {
+      return executeNext(completable, workflowContext);
+    } else if (ifFilter.map(f -> f.test(workflowContext, taskContext, input)).orElse(true)) {
       return executeNext(
           completable
               .thenCompose(workflowContext.instance()::suspendedCheck)

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

@@ -22,17 +22,22 @@
 import io.serverlessworkflow.impl.persistence.bigmap.BytesBigMapApplicationBuilder;
 import io.serverlessworkflow.impl.persistence.mvstore.MVStorePersistenceStore;
 import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Path;
 import java.util.Map;
 
 public class DBGenerator {
 
   public static void main(String[] args) throws IOException {
+    Files.deleteIfExists(Path.of("test.db"));
     try (MVStorePersistenceStore store = new MVStorePersistenceStore("test.db");
         WorkflowApplication application =
-            BytesBigMapApplicationBuilder.builder(WorkflowApplication.builder(), store).build()) {
+            BytesBigMapApplicationBuilder.builder(
+                    WorkflowApplication.builder().withListener(new TraceExecutionListener()), store)
+                .build()) {
       WorkflowDefinition definition =
           application.workflowDefinition(
-              readWorkflowFromClasspath("workflows-samples/listen-to-any.yaml"));
+              readWorkflowFromClasspath("workflows-samples/set-listen-to-any.yaml"));
       definition.instance(Map.of()).start();
     }
   }

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

@@ -30,29 +30,55 @@
 import io.serverlessworkflow.impl.persistence.mvstore.MVStorePersistenceStore;
 import java.io.IOException;
 import java.util.Collection;
+import java.util.Map;
 import org.junit.jupiter.api.Test;
 
 public class MvStorePersistenceTest {
 
   @Test
-  public void testRestoreWaitingInstance() throws IOException {
+  void testWaitingInstance() throws IOException {
+    TaskCounterPerInstanceListener taskCounter = new TaskCounterPerInstanceListener();
+    try (WorkflowApplication application =
+        WorkflowApplication.builder().withListener(taskCounter).build()) {
+      WorkflowDefinition definition =
+          application.workflowDefinition(
+              readWorkflowFromClasspath("workflows-samples/set-listen-to-any.yaml"));
+
+      WorkflowInstance instance = definition.instance(Map.of());
+      instance.start();
+      assertThat(taskCounter.taskCounter(instance.id()).orElseThrow().completed()).isEqualTo(1);
+    }
+  }
+
+  @Test
+  void testRestoreWaitingInstance() throws IOException {
     WorkflowBufferFactory bufferFactory = DefaultBufferFactory.factory();
+    TaskCounterPerInstanceListener taskCounter = new TaskCounterPerInstanceListener();
     try (MVStorePersistenceStore store = new MVStorePersistenceStore("test.db");
         WorkflowApplication application =
-            BytesBigMapApplicationBuilder.builder(WorkflowApplication.builder(), store)
+            BytesBigMapApplicationBuilder.builder(
+                    WorkflowApplication.builder()
+                        .withListener(taskCounter)
+                        .withListener(new TraceExecutionListener()),
+                    store)
                 .withFactory(bufferFactory)
                 .build();
         WorkflowPersistenceRestorer restorer =
             new BytesBigMapPersistenceRestorer(store, bufferFactory); ) {
       WorkflowDefinition definition =
           application.workflowDefinition(
-              readWorkflowFromClasspath("workflows-samples/listen-to-any.yaml"));
+              readWorkflowFromClasspath("workflows-samples/set-listen-to-any.yaml"));
       Collection<WorkflowInstance> instances = restorer.restoreAll(definition).values();
       assertThat(instances).hasSize(1);
       instances.forEach(WorkflowInstance::start);
       assertThat(instances)
           .singleElement()
-          .satisfies((instance -> assertThat(instance.status()).isEqualTo(WorkflowStatus.WAITING)));
+          .satisfies(
+              instance -> {
+                assertThat(instance.status()).isEqualTo(WorkflowStatus.WAITING);
+                assertThat(taskCounter.taskCounter(instance.id()).orElseThrow().completed())
+                    .isEqualTo(0);
+              });
     }
   }
 }

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

@@ -0,0 +1,67 @@
+/*
+ * Copyright 2020-Present The Serverless Workflow Specification 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.serverlessworkflow.impl.test;
+
+import io.serverlessworkflow.impl.lifecycle.TaskCompletedEvent;
+import io.serverlessworkflow.impl.lifecycle.TaskStartedEvent;
+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 {
+
+  public static class TaskCounter {
+    private int started;
+    private int completed;
+
+    public void incStarted() {
+      started++;
+    }
+
+    public void incCompleted() {
+      completed++;
+    }
+
+    public int started() {
+      return started;
+    }
+
+    public int completed() {
+      return completed;
+    }
+  }
+
+  private Map<String, TaskCounter> taskCounter = new ConcurrentHashMap<>();
+
+  public void onTaskStarted(TaskStartedEvent ev) {
+    taskCounter(ev).incStarted();
+  }
+
+  private TaskCounter taskCounter(WorkflowEvent ev) {
+    return taskCounter.computeIfAbsent(
+        ev.workflowContext().instanceData().id(), k -> new TaskCounter());
+  }
+
+  public Optional<TaskCounter> taskCounter(String instanceId) {
+    return Optional.ofNullable(taskCounter.get(instanceId));
+  }
+
+  public void onTaskCompleted(TaskCompletedEvent ev) {
+    taskCounter(ev).incCompleted();
+  }
+}

impl/test/src/test/resources/workflows-samples/set-listen-to-any.yaml

@@ -0,0 +1,13 @@
+document:
+  dsl: '1.0.0-alpha5'
+  namespace: test
+  name: set-listen-to-any
+  version: '0.1.0'
+do:
+  - doSomethingBeforeEvent:
+      set: 
+        name: javierito
+  - callDoctor:
+      listen:
+        to:
+          any: []