[Fix #782] Fixes after manual testing

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

Author: fjtirado

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

@@ -15,6 +15,8 @@
  */
 package io.serverlessworkflow.impl;
 
+import static io.serverlessworkflow.impl.WorkflowUtils.safeClose;
+
 import io.serverlessworkflow.api.types.SchemaInline;
 import io.serverlessworkflow.api.types.Workflow;
 import io.serverlessworkflow.impl.events.EventConsumer;
@@ -39,13 +41,9 @@
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ExecutorService;
 import java.util.stream.Collectors;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
 
 public class WorkflowApplication implements AutoCloseable {
 
-  private static final Logger logger = LoggerFactory.getLogger(WorkflowApplication.class);
-
   private final TaskExecutorFactory taskFactory;
   private final ExpressionFactory exprFactory;
   private final ResourceLoaderFactory resourceLoaderFactory;
@@ -271,14 +269,11 @@ public void close() {
       safeClose(definition);
     }
     definitions.clear();
-  }
 
-  private void safeClose(AutoCloseable closeable) {
-    try {
-      closeable.close();
-    } catch (Exception ex) {
-      logger.warn("Error closing resource {}", closeable.getClass().getName(), ex);
+    for (WorkflowExecutionListener listener : listeners) {
+      listener.close();
     }
+    listeners.clear();
   }
 
   public WorkflowPositionFactory positionFactory() {

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

@@ -50,7 +50,7 @@ public class WorkflowMutableInstance implements WorkflowInstance {
   private Lock statusLock = new ReentrantLock();
   private Map<CompletableFuture<TaskContext>, TaskContext> suspended;
 
-  public WorkflowMutableInstance(WorkflowDefinition definition, String id, WorkflowModel input) {
+  protected WorkflowMutableInstance(WorkflowDefinition definition, String id, WorkflowModel input) {
     this.id = id;
     this.input = input;
     this.status = new AtomicReference<>(WorkflowStatus.PENDING);
@@ -265,5 +265,5 @@ public boolean cancel() {
     }
   }
 
-  public void restoreContext(WorkflowDefinition definition, TaskContext context) {}
+  public void restoreContext(WorkflowContext workflow, TaskContext context) {}
 }

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

@@ -28,11 +28,15 @@
 import java.net.URI;
 import java.util.Map;
 import java.util.Optional;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 public class WorkflowUtils {
 
   private WorkflowUtils() {}
 
+  private static final Logger logger = LoggerFactory.getLogger(WorkflowUtils.class);
+
   public static Optional<SchemaValidator> getSchemaValidator(
       SchemaValidatorFactory validatorFactory, ResourceLoader resourceLoader, SchemaUnion schema) {
     if (schema != null) {
@@ -138,4 +142,12 @@ public static String toString(UriTemplate template) {
     URI uri = template.getLiteralUri();
     return uri != null ? uri.toString() : template.getLiteralUriTemplate();
   }
+
+  public static void safeClose(AutoCloseable closeable) {
+    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

@@ -192,7 +192,7 @@ 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);
+    workflowContext.instance().restoreContext(workflowContext, taskContext);
     CompletableFuture<TaskContext> completable = CompletableFuture.completedFuture(taskContext);
     if (taskContext.isCompleted() && !TaskExecutorHelper.isActive(workflowContext)) {
       return completable;

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

@@ -15,7 +15,7 @@
  */
 package io.serverlessworkflow.impl.lifecycle;
 
-public interface WorkflowExecutionListener {
+public interface WorkflowExecutionListener extends AutoCloseable {
 
   default void onWorkflowStarted(WorkflowStartedEvent ev) {}
 
@@ -42,4 +42,7 @@ default void onTaskSuspended(TaskSuspendedEvent ev) {}
   default void onTaskResumed(TaskResumedEvent ev) {}
 
   default void onTaskRetried(TaskRetriedEvent ev) {}
+
+  @Override
+  default void close() {}
 }

impl/jackson/src/main/java/io/serverlessworkflow/impl/expressions/jq/JacksonModel.java

@@ -16,6 +16,7 @@
 package io.serverlessworkflow.impl.expressions.jq;
 
 import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.annotation.JsonDeserialize;
 import com.fasterxml.jackson.databind.annotation.JsonSerialize;
 import com.fasterxml.jackson.databind.node.ArrayNode;
 import com.fasterxml.jackson.databind.node.BooleanNode;
@@ -29,6 +30,7 @@
 import java.util.Optional;
 
 @JsonSerialize(using = JacksonModelSerializer.class)
+@JsonDeserialize(using = JacksonModelDeserializer.class)
 public class JacksonModel implements WorkflowModel {
 
   protected JsonNode node;

impl/jackson/src/main/java/io/serverlessworkflow/impl/expressions/jq/JacksonModelDeserializer.java

@@ -0,0 +1,37 @@
+/*
+ * 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.expressions.jq;
+
+import com.fasterxml.jackson.core.JacksonException;
+import com.fasterxml.jackson.core.JsonParser;
+import com.fasterxml.jackson.databind.DeserializationContext;
+import com.fasterxml.jackson.databind.deser.std.StdDeserializer;
+import java.io.IOException;
+
+public class JacksonModelDeserializer extends StdDeserializer<JacksonModel> {
+
+  private static final long serialVersionUID = 1L;
+
+  protected JacksonModelDeserializer() {
+    super(JacksonModel.class);
+  }
+
+  @Override
+  public JacksonModel deserialize(JsonParser p, DeserializationContext ctxt)
+      throws IOException, JacksonException {
+    return new JacksonModel(p.readValueAsTree());
+  }
+}

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

@@ -108,8 +108,11 @@ public Object readObject() {
       case INSTANT:
         return readInstant();
 
-      default:
+      case CUSTOM:
         return readCustomObject();
+
+      default:
+        throw new IllegalStateException("Unsupported type " + type);
     }
   }
 

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

@@ -99,6 +99,7 @@ public WorkflowOutputBuffer writeObject(Object object) {
       writeType(Type.BYTES);
       writeBytes(bytes);
     } else {
+      writeType(Type.CUSTOM);
       writeCustomObject(object);
     }
     return this;
@@ -109,15 +110,14 @@ protected void writeClass(Class<?> objectClass) {
   }
 
   protected void writeCustomObject(Object object) {
-    customMarshallers.stream()
-        .filter(m -> m.getObjectClass().isAssignableFrom(object.getClass()))
-        .findFirst()
-        .ifPresentOrElse(
-            m -> {
-              writeClass(m.getObjectClass());
-              m.write(this, m.getObjectClass().cast(object));
-            },
-            () -> new IllegalArgumentException("Unsupported type " + object.getClass()));
+    CustomObjectMarshaller marshaller =
+        customMarshallers.stream()
+            .filter(m -> m.getObjectClass().isAssignableFrom(object.getClass()))
+            .findFirst()
+            .orElseThrow(
+                () -> new IllegalArgumentException("Unsupported type " + object.getClass()));
+    writeClass(marshaller.getObjectClass());
+    marshaller.write(this, marshaller.getObjectClass().cast(object));
   }
 
   protected void writeType(Type type) {

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

@@ -18,11 +18,19 @@
 import java.io.InputStream;
 import java.io.OutputStream;
 import java.util.Collection;
+import java.util.ServiceLoader;
 
 public class DefaultBufferFactory implements WorkflowBufferFactory {
 
   private final Collection<CustomObjectMarshaller> marshallers;
 
+  public static DefaultBufferFactory factory() {
+    return new DefaultBufferFactory(
+        ServiceLoader.load(CustomObjectMarshaller.class).stream()
+            .map(ServiceLoader.Provider::get)
+            .toList());
+  }
+
   public DefaultBufferFactory(Collection<CustomObjectMarshaller> marshallers) {
     this.marshallers = marshallers;
   }