[Fix #782] Adding MVStore persistence

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

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

@@ -16,7 +16,6 @@
 package io.serverlessworkflow.impl;
 
 import com.github.f4b6a3.ulid.UlidCreator;
-import io.serverlessworkflow.api.types.Document;
 import io.serverlessworkflow.api.types.SchemaInline;
 import io.serverlessworkflow.api.types.Workflow;
 import io.serverlessworkflow.impl.events.EventConsumer;
@@ -62,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;
@@ -150,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);
@@ -249,15 +248,12 @@ public WorkflowApplication build() {
     }
   }
 
-  private static record WorkflowId(String namespace, String name, String version) {
-    static WorkflowId of(Document document) {
-      return new WorkflowId(document.getNamespace(), document.getName(), document.getVersion());
-    }
+  public WorkflowDefinition workflowDefinition(Workflow workflow) {
+    return definitions.computeIfAbsent(WorkflowId.of(workflow), k -> createDefinition(workflow));
   }
 
-  public WorkflowDefinition workflowDefinition(Workflow workflow) {
-    return definitions.computeIfAbsent(
-        WorkflowId.of(workflow.getDocument()), k -> WorkflowDefinition.of(this, 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/WorkflowId.java

@@ -0,0 +1,40 @@
+/*
+ * 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;
+
+import io.serverlessworkflow.api.types.Document;
+import io.serverlessworkflow.api.types.Workflow;
+
+public record WorkflowId(String namespace, String name, String version) {
+
+  public static WorkflowId of(Workflow workflow) {
+    Document document = workflow.getDocument();
+    return new WorkflowId(document.getNamespace(), document.getName(), document.getVersion());
+  }
+
+  public String identifier() {
+    return identifier(namespace, name, version);
+  }
+
+  public static String asString(Workflow workflow) {
+    Document document = workflow.getDocument();
+    return identifier(document.getNamespace(), document.getName(), document.getVersion());
+  }
+
+  private static String identifier(String namespace, String name, String version) {
+    return namespace + "-" + name + "-" + version;
+  }
+}

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

@@ -51,18 +51,18 @@ public class WorkflowMutableInstance implements WorkflowInstance {
   private TaskContext suspendedTask;
   private CompletableFuture<TaskContext> cancelled;
 
-  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 =
@@ -107,7 +107,7 @@ private WorkflowModel whenSuccess(WorkflowModel node) {
             .map(f -> f.apply(workflowContext, null, node))
             .orElse(node);
     workflowContext.definition().outputSchemaValidator().ifPresent(v -> v.validate(output));
-    status.compareAndSet(WorkflowStatus.RUNNING, WorkflowStatus.COMPLETED);
+    status.set(WorkflowStatus.COMPLETED);
     publishEvent(
         workflowContext, l -> l.onWorkflowCompleted(new WorkflowCompletedEvent(workflowContext)));
     return output;
@@ -145,12 +145,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) {
@@ -215,15 +217,9 @@ public boolean resume() {
     }
   }
 
-  public CompletableFuture<TaskContext> completedChecks(TaskContext t) {
+  public CompletableFuture<TaskContext> cancelCheck(TaskContext t) {
     try {
       statusLock.lock();
-      if (suspended != null) {
-        suspendedTask = t;
-        publishEvent(
-            workflowContext, l -> l.onTaskSuspended(new TaskSuspendedEvent(workflowContext, t)));
-        return suspended;
-      }
       if (cancelled != null) {
         cancelled = new CompletableFuture<TaskContext>();
         workflowContext.instance().status(WorkflowStatus.CANCELLED);
@@ -237,6 +233,29 @@ public CompletableFuture<TaskContext> completedChecks(TaskContext t) {
     return CompletableFuture.completedFuture(t);
   }
 
+  public CompletableFuture<TaskContext> suspendedCheck(TaskContext t) {
+    try {
+      statusLock.lock();
+      if (suspended != null) {
+        this.suspendedTask = t;
+        publishEvent(
+            workflowContext,
+            l -> l.onTaskSuspended(new TaskSuspendedEvent(workflowContext, suspendedTask)));
+        return suspended;
+      }
+    } finally {
+      statusLock.unlock();
+    }
+    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 {