[Fix #782] More changes after testing

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

Author: fjtirado

impl/README.md

@@ -213,6 +213,10 @@ As shown in previous examples, to start a new workflow instance, first a [Workfl
 
 Once started, and before it completes, a workflow instance execution can be suspended or cancelled. Once cancelled, a workflow instance is done, while a suspended one might be resumed. 
 
+## Persistent 
+
+Workflow progress might be recorded into DB. See [details](persistence/README.md)
+
 ## Fluent Java DSL
 
 Prefer building workflows programmatically with type-safe builders and recipes?

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/WorkflowUtils.java

@@ -144,10 +144,12 @@ public static String toString(UriTemplate template) {
   }
 
   public static void safeClose(AutoCloseable closeable) {
-    try {
-      closeable.close();
-    } catch (Exception ex) {
-      logger.warn("Error closing resource {}", closeable.getClass().getName(), ex);
+    if (closeable != null) {
+      try {
+        closeable.close();
+      } catch (Exception ex) {
+        logger.warn("Error closing resource {}", closeable.getClass().getName(), ex);
+      }
     }
   }
 }

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/persistence/README.md

@@ -0,0 +1,13 @@
+[![Gitpod ready-to-code](https://img.shields.io/badge/Gitpod-ready--to--code-blue?logo=gitpod)](https://gitpod.io/#https://github.com/serverlessworkflow/sdk-java)
+
+# Serverless Workflow Specification — Java SDK (Reference Implementation)- Persistence
+
+Workflow persistence aim is to be able to restore workflow instances execution in the event of a JVM stop. To do that, progress of every running instance is persisted into the underlying DB by using life cycle events. Later on, when a new JVM is instantiated, the application is expected to manually start those instances that are not longer being executed by any other JVM, using the information previously stored. 
+
+Currently, persistence structure has been layout for key-value store dbs, plus one concrete implementation using [H2 MVStore](mvstore/README.md). Next step will do the same for relational dbs (table layout plus concrete implementation using Postgresql). 
+
+Map of key values has been given precedence because, when persisting the status of a running workflow instance, the number of writes are usually large, while read only operations are only performed when the JVM starts up. This give a performance edge for this kind of db over relational ones.
+
+---
+
+*Questions or ideas? PRs and issues welcome!*

impl/persistence/api/src/main/java/io/serverlessworkflow/impl/marshaller/DefaultBufferFactory.java

@@ -24,14 +24,19 @@ public class DefaultBufferFactory implements WorkflowBufferFactory {
 
   private final Collection<CustomObjectMarshaller> marshallers;
 
+  private static class DefaultBufferFactoryHolder {
+    private static DefaultBufferFactory instance =
+        new DefaultBufferFactory(
+            ServiceLoader.load(CustomObjectMarshaller.class).stream()
+                .map(ServiceLoader.Provider::get)
+                .toList());
+  }
+
   public static DefaultBufferFactory factory() {
-    return new DefaultBufferFactory(
-        ServiceLoader.load(CustomObjectMarshaller.class).stream()
-            .map(ServiceLoader.Provider::get)
-            .toList());
+    return DefaultBufferFactoryHolder.instance;
   }
 
-  public DefaultBufferFactory(Collection<CustomObjectMarshaller> marshallers) {
+  protected DefaultBufferFactory(Collection<CustomObjectMarshaller> marshallers) {
     this.marshallers = marshallers;
   }
 

impl/persistence/api/src/main/java/io/serverlessworkflow/impl/persistence/PersistenceApplicationBuilder.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.persistence;
+
+import io.serverlessworkflow.impl.WorkflowApplication;
+import io.serverlessworkflow.impl.WorkflowApplication.Builder;
+
+public class PersistenceApplicationBuilder {
+
+  public static PersistenceApplicationBuilder builder(
+      WorkflowApplication.Builder builder, PersistenceInstanceWriter writer) {
+    return new PersistenceApplicationBuilder(builder, writer);
+  }
+
+  private final PersistenceInstanceWriter writer;
+  private final WorkflowApplication.Builder appBuilder;
+
+  protected PersistenceApplicationBuilder(Builder appBuilder, PersistenceInstanceWriter writer) {
+    this.appBuilder = appBuilder;
+    this.writer = writer;
+  }
+
+  public WorkflowApplication build() {
+    appBuilder.withListener(new WorkflowPersistenceListener(writer));
+    return appBuilder.build();
+  }
+}

impl/persistence/api/src/main/java/io/serverlessworkflow/impl/persistence/PersistenceInstanceHandlers.java

@@ -0,0 +1,44 @@
+/*
+ * 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.persistence;
+
+import static io.serverlessworkflow.impl.WorkflowUtils.safeClose;
+
+public abstract class PersistenceInstanceHandlers implements AutoCloseable {
+
+  protected final PersistenceInstanceWriter writer;
+  protected final PersistenceInstanceReader reader;
+
+  protected PersistenceInstanceHandlers(
+      PersistenceInstanceWriter writer, PersistenceInstanceReader reader) {
+    this.writer = writer;
+    this.reader = reader;
+  }
+
+  public PersistenceInstanceWriter writer() {
+    return writer;
+  }
+
+  public PersistenceInstanceReader reader() {
+    return reader;
+  }
+
+  @Override
+  public void close() {
+    safeClose(writer);
+    safeClose(reader);
+  }
+}

impl/persistence/api/src/main/java/io/serverlessworkflow/impl/persistence/WorkflowPersistenceRestorer.java renamed to impl/persistence/api/src/main/java/io/serverlessworkflow/impl/persistence/PersistenceInstanceReader.java

@@ -21,13 +21,12 @@
 import java.util.Map;
 import java.util.Optional;
 
-public interface WorkflowPersistenceRestorer extends AutoCloseable {
-  Map<String, WorkflowInstance> restoreAll(WorkflowDefinition definition);
+public interface PersistenceInstanceReader extends AutoCloseable {
+  Map<String, WorkflowInstance> readAll(WorkflowDefinition definition);
 
-  Map<String, WorkflowInstance> restore(
-      WorkflowDefinition definition, Collection<String> instanceIds);
+  Map<String, WorkflowInstance> read(WorkflowDefinition definition, Collection<String> instanceIds);
 
-  Optional<WorkflowInstance> restore(WorkflowDefinition definition, String instanceId);
+  Optional<WorkflowInstance> read(WorkflowDefinition definition, String instanceId);
 
   @Override
   default void close() {}

impl/persistence/api/src/main/java/io/serverlessworkflow/impl/persistence/WorkflowPersistenceWriter.java renamed to impl/persistence/api/src/main/java/io/serverlessworkflow/impl/persistence/PersistenceInstanceWriter.java

@@ -18,7 +18,7 @@
 import io.serverlessworkflow.impl.TaskContextData;
 import io.serverlessworkflow.impl.WorkflowContextData;
 
-public interface WorkflowPersistenceWriter extends AutoCloseable {
+public interface PersistenceInstanceWriter extends AutoCloseable {
 
   void started(WorkflowContextData workflowContext);