[Fix #782] Refining approach

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

Author: fjtirado

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

@@ -15,6 +15,8 @@
  */
 package io.serverlessworkflow.impl;
 
+import java.util.Objects;
+
 public class StringBufferWorkflowPosition implements WorkflowMutablePosition {
 
   private StringBuilder sb;
@@ -67,4 +69,18 @@ public Object last() {
     int indexOf = sb.lastIndexOf("/");
     return indexOf != -1 ? jsonPointer().substring(indexOf + 1) : "";
   }
+
+  @Override
+  public int hashCode() {
+    return Objects.hash(sb);
+  }
+
+  @Override
+  public boolean equals(Object obj) {
+    if (this == obj) return true;
+    if (obj == null) return false;
+    if (getClass() != obj.getClass()) return false;
+    StringBufferWorkflowPosition other = (StringBufferWorkflowPosition) obj;
+    return Objects.equals(sb, other.sb);
+  }
 }

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

@@ -37,6 +37,7 @@ public class TaskContext implements TaskContextData {
   private WorkflowModel rawOutput;
   private Instant completedAt;
   private TransitionInfo transition;
+  private boolean completed;
 
   public TaskContext(
       WorkflowModel input,
@@ -109,6 +110,7 @@ public WorkflowModel rawOutput() {
 
   public TaskContext output(WorkflowModel output) {
     this.output = output;
+    this.completed = true;
     return this;
   }
 
@@ -159,6 +161,10 @@ public TaskContext transition(TransitionInfo transition) {
     return this;
   }
 
+  public boolean isCompleted() {
+    return completed;
+  }
+
   @Override
   public String toString() {
     return "TaskContext [position="

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

@@ -25,6 +25,8 @@
 import io.serverlessworkflow.impl.resources.ResourceLoader;
 import io.serverlessworkflow.impl.schema.SchemaValidator;
 import java.nio.file.Path;
+import java.util.HashMap;
+import java.util.Map;
 import java.util.Optional;
 
 public class WorkflowDefinition implements AutoCloseable, WorkflowDefinitionData {
@@ -37,6 +39,7 @@ public class WorkflowDefinition implements AutoCloseable, WorkflowDefinitionData
   private final WorkflowApplication application;
   private final TaskExecutor<?> taskExecutor;
   private final ResourceLoader resourceLoader;
+  private final Map<String, TaskExecutor<?>> executors = new HashMap<>();
 
   private WorkflowDefinition(
       WorkflowApplication application, Workflow workflow, ResourceLoader resourceLoader) {
@@ -72,8 +75,7 @@ static WorkflowDefinition of(WorkflowApplication application, Workflow workflow,
   public WorkflowInstance instance(Object input) {
     WorkflowModel inputModel = application.modelFactory().fromAny(input);
     inputSchemaValidator().ifPresent(v -> v.validate(inputModel));
-    return new WorkflowMutableInstance(
-        this, application().idFactory().get(), inputModel, WorkflowStatus.PENDING);
+    return new WorkflowMutableInstance(this, application().idFactory().get(), inputModel);
   }
 
   Optional<SchemaValidator> inputSchemaValidator() {
@@ -110,6 +112,14 @@ public ResourceLoader resourceLoader() {
     return resourceLoader;
   }
 
+  public TaskExecutor<?> taskExecutor(String jsonPointer) {
+    return executors.get(jsonPointer);
+  }
+
+  public void addTaskExecutor(WorkflowMutablePosition position, TaskExecutor<?> taskExecutor) {
+    executors.put(position.jsonPointer(), taskExecutor);
+  }
+
   @Override
   public void close() {}
 }

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

@@ -38,32 +38,43 @@
 public class WorkflowMutableInstance implements WorkflowInstance {
 
   protected final AtomicReference<WorkflowStatus> status;
-  private final String id;
-  private final WorkflowModel input;
+  protected final String id;
+  protected final WorkflowModel input;
+
+  protected final WorkflowContext workflowContext;
+  protected Instant startedAt;
+
+  protected AtomicReference<CompletableFuture<WorkflowModel>> futureRef = new AtomicReference<>();
+  protected Instant completedAt;
 
-  private WorkflowContext workflowContext;
-  private Instant startedAt;
-  private Instant completedAt;
-  private volatile WorkflowModel output;
   private Lock statusLock = new ReentrantLock();
-  private CompletableFuture<WorkflowModel> completableFuture;
   private Map<CompletableFuture<TaskContext>, TaskContext> suspended;
 
-  public WorkflowMutableInstance(
-      WorkflowDefinition definition, String id, WorkflowModel input, WorkflowStatus status) {
+  public WorkflowMutableInstance(WorkflowDefinition definition, String id, WorkflowModel input) {
     this.id = id;
     this.input = input;
-    this.status = new AtomicReference<>(status);
+    this.status = new AtomicReference<>(WorkflowStatus.PENDING);
     this.workflowContext = new WorkflowContext(definition, this);
   }
 
   @Override
   public CompletableFuture<WorkflowModel> start() {
-    startedAt = Instant.now();
-    status.set(WorkflowStatus.RUNNING);
-    publishEvent(
-        workflowContext, l -> l.onWorkflowStarted(new WorkflowStartedEvent(workflowContext)));
-    this.completableFuture =
+    return startExecution(
+        () -> {
+          startedAt = Instant.now();
+          status.set(WorkflowStatus.RUNNING);
+          publishEvent(
+              workflowContext, l -> l.onWorkflowStarted(new WorkflowStartedEvent(workflowContext)));
+        });
+  }
+
+  protected final CompletableFuture<WorkflowModel> startExecution(Runnable runnable) {
+    CompletableFuture<WorkflowModel> future = futureRef.get();
+    if (future != null) {
+      return future;
+    }
+    runnable.run();
+    future =
         TaskExecutorHelper.processTaskList(
                 workflowContext.definition().startTask(),
                 workflowContext,
@@ -75,7 +86,8 @@ public CompletableFuture<WorkflowModel> start() {
                     .orElse(input))
             .whenComplete(this::whenFailed)
             .thenApply(this::whenSuccess);
-    return completableFuture;
+    futureRef.set(future);
+    return future;
   }
 
   private void whenFailed(WorkflowModel result, Throwable ex) {
@@ -94,7 +106,7 @@ private void handleException(Throwable ex) {
   }
 
   private WorkflowModel whenSuccess(WorkflowModel node) {
-    output =
+    WorkflowModel output =
         workflowContext
             .definition()
             .outputFilter()
@@ -103,7 +115,8 @@ private WorkflowModel whenSuccess(WorkflowModel node) {
     workflowContext.definition().outputSchemaValidator().ifPresent(v -> v.validate(output));
     status.set(WorkflowStatus.COMPLETED);
     publishEvent(
-        workflowContext, l -> l.onWorkflowCompleted(new WorkflowCompletedEvent(workflowContext)));
+        workflowContext,
+        l -> l.onWorkflowCompleted(new WorkflowCompletedEvent(workflowContext, output)));
     return output;
   }
 
@@ -134,19 +147,17 @@ public WorkflowStatus status() {
 
   @Override
   public WorkflowModel output() {
-    return output;
+    return futureRef.get().join();
   }
 
   @Override
   public <T> T outputAs(Class<T> clazz) {
-    return output != null
-        ? output
-            .as(clazz)
-            .orElseThrow(
-                () ->
-                    new IllegalArgumentException(
-                        "Output " + output + " cannot be converted to class " + clazz))
-        : null;
+    return output()
+        .as(clazz)
+        .orElseThrow(
+            () ->
+                new IllegalArgumentException(
+                    "Output " + output() + " cannot be converted to class " + clazz));
   }
 
   public void status(WorkflowStatus state) {
@@ -236,13 +247,6 @@ public CompletableFuture<TaskContext> suspendedCheck(TaskContext t) {
     return CompletableFuture.completedFuture(t);
   }
 
-  // internal purposes only, not to be invoked directly by users of the API
-  public void restore(
-      WorkflowPosition position, WorkflowModel model, WorkflowModel context, Instant startedAt) {
-    this.startedAt = startedAt;
-    workflowContext.context(context);
-  }
-
   @Override
   public boolean cancel() {
     try {
@@ -260,4 +264,6 @@ public boolean cancel() {
       statusLock.unlock();
     }
   }
+
+  public void restoreContext(WorkflowDefinition definition, TaskContext context) {}
 }

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

@@ -78,11 +78,13 @@ public abstract static class AbstractTaskExecutorBuilder<
     protected final WorkflowApplication application;
     protected final Workflow workflow;
     protected final ResourceLoader resourceLoader;
+    private final WorkflowDefinition definition;
 
     private V instance;
 
     protected AbstractTaskExecutorBuilder(
         WorkflowMutablePosition position, T task, WorkflowDefinition definition) {
+      this.definition = definition;
       this.workflow = definition.workflow();
       this.taskName = position.last().toString();
       this.position = position;
@@ -147,6 +149,7 @@ public V build() {
       if (instance == null) {
         instance = buildInstance();
         buildTransition(instance);
+        definition.addTaskExecutor(position, instance);
       }
       return instance;
     }
@@ -189,8 +192,9 @@ private CompletableFuture<TaskContext> executeNext(
   public CompletableFuture<TaskContext> apply(
       WorkflowContext workflowContext, Optional<TaskContext> parentContext, WorkflowModel input) {
     TaskContext taskContext = new TaskContext(input, position, parentContext, taskName, task);
+    workflowContext.instance().restoreContext(workflowContext.definition(), taskContext);
     CompletableFuture<TaskContext> completable = CompletableFuture.completedFuture(taskContext);
-    if (!TaskExecutorHelper.isActive(workflowContext)) {
+    if (taskContext.isCompleted() && !TaskExecutorHelper.isActive(workflowContext)) {
       return completable;
     }
     if (ifFilter.map(f -> f.test(workflowContext, taskContext, input)).orElse(true)) {
@@ -256,6 +260,10 @@ private void handleException(
     }
   }
 
+  public WorkflowPosition position() {
+    return position;
+  }
+
   protected abstract TransitionInfo getSkipTransition();
 
   protected abstract CompletableFuture<TaskContext> execute(

impl/core/src/main/java/io/serverlessworkflow/impl/lifecycle/WorkflowCompletedEvent.java

@@ -16,10 +16,18 @@
 package io.serverlessworkflow.impl.lifecycle;
 
 import io.serverlessworkflow.impl.WorkflowContextData;
+import io.serverlessworkflow.impl.WorkflowModel;
 
 public class WorkflowCompletedEvent extends WorkflowEvent {
 
-  public WorkflowCompletedEvent(WorkflowContextData workflow) {
+  private WorkflowModel output;
+
+  public WorkflowCompletedEvent(WorkflowContextData workflow, WorkflowModel output) {
     super(workflow);
+    this.output = output;
+  }
+
+  public WorkflowModel output() {
+    return output;
   }
 }

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

@@ -225,7 +225,8 @@ public void onWorkflowCompleted(WorkflowCompletedEvent event) {
             builder()
                 .withData(
                     cloudEventData(
-                        new WorkflowCompletedCEData(id(ev), ref(ev), ev.eventDate(), output(ev)),
+                        new WorkflowCompletedCEData(
+                            id(ev), ref(ev), ev.eventDate(), from(event.output())),
                         this::convert))
                 .withType(WORKFLOW_COMPLETED)
                 .build());
@@ -328,10 +329,6 @@ private static String pos(TaskEvent ev) {
     return ev.taskContext().position().jsonPointer();
   }
 
-  private static Object output(WorkflowEvent ev) {
-    return from(ev.workflowContext().instanceData().output());
-  }
-
   private static Object output(TaskEvent ev) {
     return from(ev.taskContext().output());
   }

impl/persistence/api/pom.xml

@@ -11,6 +11,7 @@
           <dependency>
             <groupId>io.serverlessworkflow</groupId>
             <artifactId>serverlessworkflow-impl-core</artifactId>
-        </dependency>
+            <version>8.0.0-SNAPSHOT</version>
+          </dependency>
     </dependencies>
 </project>