[Fix #782] Adding MVStore persistence

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

Author: fjtirado

experimental/lambda/src/main/java/io/serverlessworkflow/impl/executors/func/JavaForExecutorBuilder.java

@@ -18,29 +18,23 @@
 import static io.serverlessworkflow.impl.executors.func.JavaFuncUtils.safeObject;
 
 import io.serverlessworkflow.api.types.ForTask;
-import io.serverlessworkflow.api.types.Workflow;
 import io.serverlessworkflow.api.types.func.ForTaskFunction;
 import io.serverlessworkflow.api.types.func.LoopPredicateIndex;
 import io.serverlessworkflow.api.types.func.TypedFunction;
-import io.serverlessworkflow.impl.WorkflowApplication;
+import io.serverlessworkflow.impl.WorkflowDefinition;
 import io.serverlessworkflow.impl.WorkflowMutablePosition;
 import io.serverlessworkflow.impl.WorkflowPredicate;
 import io.serverlessworkflow.impl.WorkflowValueResolver;
 import io.serverlessworkflow.impl.executors.ForExecutor.ForExecutorBuilder;
 import io.serverlessworkflow.impl.expressions.ExpressionDescriptor;
-import io.serverlessworkflow.impl.resources.ResourceLoader;
 import java.util.Collection;
 import java.util.Optional;
 
 public class JavaForExecutorBuilder extends ForExecutorBuilder {
 
   protected JavaForExecutorBuilder(
-      WorkflowMutablePosition position,
-      ForTask task,
-      Workflow workflow,
-      WorkflowApplication application,
-      ResourceLoader resourceLoader) {
-    super(position, task, workflow, application, resourceLoader);
+      WorkflowMutablePosition position, ForTask task, WorkflowDefinition definition) {
+    super(position, task, definition);
   }
 
   @Override

experimental/lambda/src/main/java/io/serverlessworkflow/impl/executors/func/JavaListenExecutorBuilder.java

@@ -19,24 +19,18 @@
 
 import io.serverlessworkflow.api.types.ListenTask;
 import io.serverlessworkflow.api.types.Until;
-import io.serverlessworkflow.api.types.Workflow;
 import io.serverlessworkflow.api.types.func.UntilPredicate;
-import io.serverlessworkflow.impl.WorkflowApplication;
+import io.serverlessworkflow.impl.WorkflowDefinition;
 import io.serverlessworkflow.impl.WorkflowMutablePosition;
 import io.serverlessworkflow.impl.WorkflowPredicate;
 import io.serverlessworkflow.impl.executors.ListenExecutor.ListenExecutorBuilder;
 import io.serverlessworkflow.impl.expressions.ExpressionDescriptor;
-import io.serverlessworkflow.impl.resources.ResourceLoader;
 
 public class JavaListenExecutorBuilder extends ListenExecutorBuilder {
 
   protected JavaListenExecutorBuilder(
-      WorkflowMutablePosition position,
-      ListenTask task,
-      Workflow workflow,
-      WorkflowApplication application,
-      ResourceLoader resourceLoader) {
-    super(position, task, workflow, application, resourceLoader);
+      WorkflowMutablePosition position, ListenTask task, WorkflowDefinition definition) {
+    super(position, task, definition);
   }
 
   @Override

experimental/lambda/src/main/java/io/serverlessworkflow/impl/executors/func/JavaSwitchExecutorBuilder.java

@@ -19,25 +19,19 @@
 
 import io.serverlessworkflow.api.types.SwitchCase;
 import io.serverlessworkflow.api.types.SwitchTask;
-import io.serverlessworkflow.api.types.Workflow;
 import io.serverlessworkflow.api.types.func.SwitchCaseFunction;
-import io.serverlessworkflow.impl.WorkflowApplication;
+import io.serverlessworkflow.impl.WorkflowDefinition;
 import io.serverlessworkflow.impl.WorkflowMutablePosition;
 import io.serverlessworkflow.impl.WorkflowPredicate;
 import io.serverlessworkflow.impl.executors.SwitchExecutor.SwitchExecutorBuilder;
 import io.serverlessworkflow.impl.expressions.ExpressionDescriptor;
-import io.serverlessworkflow.impl.resources.ResourceLoader;
 import java.util.Optional;
 
 public class JavaSwitchExecutorBuilder extends SwitchExecutorBuilder {
 
   protected JavaSwitchExecutorBuilder(
-      WorkflowMutablePosition position,
-      SwitchTask task,
-      Workflow workflow,
-      WorkflowApplication application,
-      ResourceLoader resourceLoader) {
-    super(position, task, workflow, application, resourceLoader);
+      WorkflowMutablePosition position, SwitchTask task, WorkflowDefinition definition) {
+    super(position, task, definition);
   }
 
   @Override

experimental/lambda/src/main/java/io/serverlessworkflow/impl/executors/func/JavaTaskExecutorFactory.java

@@ -17,32 +17,23 @@
 
 import io.serverlessworkflow.api.types.Task;
 import io.serverlessworkflow.api.types.TaskBase;
-import io.serverlessworkflow.api.types.Workflow;
-import io.serverlessworkflow.impl.WorkflowApplication;
+import io.serverlessworkflow.impl.WorkflowDefinition;
 import io.serverlessworkflow.impl.WorkflowMutablePosition;
 import io.serverlessworkflow.impl.executors.DefaultTaskExecutorFactory;
 import io.serverlessworkflow.impl.executors.TaskExecutorBuilder;
-import io.serverlessworkflow.impl.resources.ResourceLoader;
 
 public class JavaTaskExecutorFactory extends DefaultTaskExecutorFactory {
 
   public TaskExecutorBuilder<? extends TaskBase> getTaskExecutor(
-      WorkflowMutablePosition position,
-      Task task,
-      Workflow workflow,
-      WorkflowApplication application,
-      ResourceLoader resourceLoader) {
+      WorkflowMutablePosition position, Task task, WorkflowDefinition definition) {
     if (task.getForTask() != null) {
-      return new JavaForExecutorBuilder(
-          position, task.getForTask(), workflow, application, resourceLoader);
+      return new JavaForExecutorBuilder(position, task.getForTask(), definition);
     } else if (task.getSwitchTask() != null) {
-      return new JavaSwitchExecutorBuilder(
-          position, task.getSwitchTask(), workflow, application, resourceLoader);
+      return new JavaSwitchExecutorBuilder(position, task.getSwitchTask(), definition);
     } else if (task.getListenTask() != null) {
-      return new JavaListenExecutorBuilder(
-          position, task.getListenTask(), workflow, application, resourceLoader);
+      return new JavaListenExecutorBuilder(position, task.getListenTask(), definition);
     } else {
-      return super.getTaskExecutor(position, task, workflow, application, resourceLoader);
+      return super.getTaskExecutor(position, task, definition);
     }
   }
 }

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

@@ -61,7 +61,7 @@ public class WorkflowApplication implements AutoCloseable {
   private final Collection<EventPublisher> eventPublishers;
   private final boolean lifeCycleCEPublishingEnabled;
 
-  private WorkflowApplication(Builder builder) {
+  protected WorkflowApplication(Builder builder) {
     this.taskFactory = builder.taskFactory;
     this.exprFactory = builder.exprFactory;
     this.resourceLoaderFactory = builder.resourceLoaderFactory;
@@ -149,7 +149,7 @@ public SchemaValidator getValidator(SchemaInline inline) {
         () -> new RuntimeDescriptor("reference impl", "1.0.0_alpha", Collections.emptyMap());
     private boolean lifeCycleCEPublishingEnabled = true;
 
-    private Builder() {}
+    protected Builder() {}
 
     public Builder withListener(WorkflowExecutionListener listener) {
       listeners.add(listener);
@@ -254,7 +254,11 @@ public Map<WorkflowDefinitionId, WorkflowDefinition> workflowDefinitions() {
 
   public WorkflowDefinition workflowDefinition(Workflow workflow) {
     return definitions.computeIfAbsent(
-        WorkflowDefinitionId.of(workflow), k -> WorkflowDefinition.of(this, workflow));
+        WorkflowDefinitionId.of(workflow), k -> createDefinition(workflow));
+  }
+
+  protected WorkflowDefinition createDefinition(Workflow workflow) {
+    return WorkflowDefinition.of(this, workflow);
   }
 
   @Override

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

@@ -36,11 +36,13 @@ public class WorkflowDefinition implements AutoCloseable, WorkflowDefinitionData
   private Optional<WorkflowFilter> outputFilter = Optional.empty();
   private final WorkflowApplication application;
   private final TaskExecutor<?> taskExecutor;
+  private ResourceLoader resourceLoader;
 
   private WorkflowDefinition(
       WorkflowApplication application, Workflow workflow, ResourceLoader resourceLoader) {
     this.workflow = workflow;
     this.application = application;
+    this.resourceLoader = resourceLoader;
     if (workflow.getInput() != null) {
       Input input = workflow.getInput();
       this.inputSchemaValidator =
@@ -55,11 +57,7 @@ private WorkflowDefinition(
     }
     this.taskExecutor =
         TaskExecutorHelper.createExecutorList(
-            application.positionFactory().get(),
-            workflow.getDo(),
-            workflow,
-            application,
-            resourceLoader);
+            application.positionFactory().get(), workflow.getDo(), this);
   }
 
   static WorkflowDefinition of(WorkflowApplication application, Workflow workflow) {
@@ -72,7 +70,10 @@ static WorkflowDefinition of(WorkflowApplication application, Workflow workflow,
   }
 
   public WorkflowInstance instance(Object input) {
-    return new WorkflowMutableInstance(this, application.modelFactory().fromAny(input));
+    WorkflowModel inputModel = application.modelFactory().fromAny(input);
+    inputSchemaValidator().ifPresent(v -> v.validate(inputModel));
+    return new WorkflowMutableInstance(
+        this, application().idFactory().get(), inputModel, WorkflowStatus.PENDING);
   }
 
   Optional<SchemaValidator> inputSchemaValidator() {
@@ -106,7 +107,9 @@ public WorkflowApplication application() {
   }
 
   @Override
-  public void close() {
-    // TODO close resourcers hold for uncompleted process instances, if any
+  public void close() {}
+
+  public ResourceLoader resourceLoader() {
+    return resourceLoader;
   }
 }

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

@@ -49,18 +49,18 @@ public class WorkflowMutableInstance implements WorkflowInstance {
   private CompletableFuture<WorkflowModel> completableFuture;
   private Map<CompletableFuture<TaskContext>, TaskContext> suspended;
 
-  WorkflowMutableInstance(WorkflowDefinition definition, WorkflowModel input) {
-    this.id = definition.application().idFactory().get();
+  public WorkflowMutableInstance(
+      WorkflowDefinition definition, String id, WorkflowModel input, WorkflowStatus status) {
+    this.id = id;
     this.input = input;
-    this.status = new AtomicReference<>(WorkflowStatus.PENDING);
-    definition.inputSchemaValidator().ifPresent(v -> v.validate(input));
+    this.status = new AtomicReference<>(status);
     this.workflowContext = new WorkflowContext(definition, this);
   }
 
   @Override
   public CompletableFuture<WorkflowModel> start() {
-    this.startedAt = Instant.now();
-    this.status.set(WorkflowStatus.RUNNING);
+    startedAt = Instant.now();
+    status.set(WorkflowStatus.RUNNING);
     publishEvent(
         workflowContext, l -> l.onWorkflowStarted(new WorkflowStartedEvent(workflowContext)));
     this.completableFuture =
@@ -139,12 +139,14 @@ public WorkflowModel output() {
 
   @Override
   public <T> T outputAs(Class<T> clazz) {
-    return output
-        .as(clazz)
-        .orElseThrow(
-            () ->
-                new IllegalArgumentException(
-                    "Output " + output + " cannot be converted to class " + clazz));
+    return output != null
+        ? output
+            .as(clazz)
+            .orElseThrow(
+                () ->
+                    new IllegalArgumentException(
+                        "Output " + output + " cannot be converted to class " + clazz))
+        : null;
   }
 
   public void status(WorkflowStatus state) {
@@ -234,6 +236,13 @@ 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 {

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

@@ -28,6 +28,7 @@
 import io.serverlessworkflow.impl.TaskContext;
 import io.serverlessworkflow.impl.WorkflowApplication;
 import io.serverlessworkflow.impl.WorkflowContext;
+import io.serverlessworkflow.impl.WorkflowDefinition;
 import io.serverlessworkflow.impl.WorkflowFilter;
 import io.serverlessworkflow.impl.WorkflowModel;
 import io.serverlessworkflow.impl.WorkflowMutablePosition;
@@ -81,17 +82,13 @@ public abstract static class AbstractTaskExecutorBuilder<
     private V instance;
 
     protected AbstractTaskExecutorBuilder(
-        WorkflowMutablePosition position,
-        T task,
-        Workflow workflow,
-        WorkflowApplication application,
-        ResourceLoader resourceLoader) {
-      this.workflow = workflow;
+        WorkflowMutablePosition position, T task, WorkflowDefinition definition) {
+      this.workflow = definition.workflow();
       this.taskName = position.last().toString();
       this.position = position;
       this.task = task;
-      this.application = application;
-      this.resourceLoader = resourceLoader;
+      this.application = definition.application();
+      this.resourceLoader = definition.resourceLoader();
       if (task.getInput() != null) {
         Input input = task.getInput();
         this.inputProcessor = buildWorkflowFilter(application, input.getFrom());
@@ -174,16 +171,18 @@ protected AbstractTaskExecutor(AbstractTaskExecutorBuilder<T, ?> builder) {
 
   protected final CompletableFuture<TaskContext> executeNext(
       CompletableFuture<TaskContext> future, WorkflowContext workflow) {
-    return future.thenCompose(
-        t -> {
-          TransitionInfo transition = t.transition();
-          if (transition.isEndNode()) {
-            workflow.instance().status(WorkflowStatus.COMPLETED);
-          } else if (transition.next() != null) {
-            return transition.next().apply(workflow, t.parent(), t.output());
-          }
-          return CompletableFuture.completedFuture(t);
-        });
+    return future.thenCompose(t -> executeNext(workflow, t));
+  }
+
+  private CompletableFuture<TaskContext> executeNext(
+      WorkflowContext workflow, TaskContext taskContext) {
+    TransitionInfo transition = taskContext.transition();
+    if (transition.isEndNode()) {
+      workflow.instance().status(WorkflowStatus.COMPLETED);
+    } else if (transition.next() != null) {
+      return transition.next().apply(workflow, taskContext.parent(), taskContext.output());
+    }
+    return CompletableFuture.completedFuture(taskContext);
   }
 
   @Override

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

@@ -16,13 +16,11 @@
 package io.serverlessworkflow.impl.executors;
 
 import io.serverlessworkflow.api.types.TaskBase;
-import io.serverlessworkflow.api.types.Workflow;
 import io.serverlessworkflow.impl.TaskContext;
-import io.serverlessworkflow.impl.WorkflowApplication;
 import io.serverlessworkflow.impl.WorkflowContext;
+import io.serverlessworkflow.impl.WorkflowDefinition;
 import io.serverlessworkflow.impl.WorkflowModel;
 import io.serverlessworkflow.impl.WorkflowMutablePosition;
-import io.serverlessworkflow.impl.resources.ResourceLoader;
 import java.util.concurrent.CompletableFuture;
 
 public class CallTaskExecutor<T extends TaskBase> extends RegularTaskExecutor<T> {
@@ -36,11 +34,9 @@ public static class CallTaskExecutorBuilder<T extends TaskBase>
     protected CallTaskExecutorBuilder(
         WorkflowMutablePosition position,
         T task,
-        Workflow workflow,
-        WorkflowApplication application,
-        ResourceLoader resourceLoader,
+        WorkflowDefinition definition,
         CallableTask<T> callable) {
-      super(position, task, workflow, application, resourceLoader);
+      super(position, task, definition);
       this.callable = callable;
       callable.init(task, workflow, application, resourceLoader);
     }