[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/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) {
@@ -110,6 +113,14 @@ public ResourceLoader resourceLoader() {
     return resourceLoader;
   }
 
+  public TaskExecutor<?> taskExecutor(WorkflowPosition position) {
+    return executors.get(position.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

@@ -49,7 +49,7 @@ public class WorkflowMutableInstance implements WorkflowInstance {
   private CompletableFuture<WorkflowModel> completableFuture;
   private Map<CompletableFuture<TaskContext>, TaskContext> suspended;
 
-  public WorkflowMutableInstance(
+  protected WorkflowMutableInstance(
       WorkflowDefinition definition, String id, WorkflowModel input, WorkflowStatus status) {
     this.id = id;
     this.input = input;
@@ -236,13 +236,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 {

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;
     }
@@ -256,6 +259,10 @@ private void handleException(
     }
   }
 
+  public WorkflowPosition position() {
+    return position;
+  }
+
   protected abstract TransitionInfo getSkipTransition();
 
   protected abstract CompletableFuture<TaskContext> execute(

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>

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

@@ -0,0 +1,137 @@
+/*
+ * 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.marshaller;
+
+import java.time.Instant;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.LinkedHashMap;
+import java.util.Map;
+
+public abstract class AbstractInputBuffer implements WorkflowInputBuffer {
+
+  private final Collection<CustomObjectMarshaller> customMarshallers;
+
+  protected AbstractInputBuffer(Collection<CustomObjectMarshaller> customMarshallers) {
+    this.customMarshallers = customMarshallers;
+  }
+
+  @Override
+  public <T extends Enum<T>> T readEnum(Class<T> enumClass) {
+    return Enum.valueOf(enumClass, readString());
+  }
+
+  @Override
+  public Instant readInstant() {
+    return Instant.ofEpochMilli(readLong());
+  }
+
+  @Override
+  public Map<String, Object> readMap() {
+    int size = readInt();
+    Map<String, Object> map = new LinkedHashMap<String, Object>(size);
+    while (size-- > 0) {
+      map.put(readString(), readObject());
+    }
+    return map;
+  }
+
+  @Override
+  public Collection<Object> readCollection() {
+    int size = readInt();
+    Collection<Object> col = new ArrayList<>(size);
+    while (size-- > 0) {
+      col.add(readObject());
+    }
+    return col;
+  }
+
+  protected Type readType() {
+    return Type.values()[readByte()];
+  }
+
+  @Override
+  public Object readObject() {
+
+    Type type = readType();
+
+    switch (type) {
+      case NULL:
+        return null;
+
+      case SHORT:
+        return readShort();
+
+      case LONG:
+        return readLong();
+
+      case INT:
+        return readInt();
+
+      case BYTE:
+        return readByte();
+
+      case BYTES:
+        return readBytes();
+
+      case FLOAT:
+        return readFloat();
+
+      case DOUBLE:
+        return readDouble();
+
+      case BOOLEAN:
+        return readBoolean();
+
+      case STRING:
+        return readString();
+
+      case MAP:
+        return readMap();
+
+      case COLLECTION:
+        return readCollection();
+
+      case INSTANT:
+        return readInstant();
+
+      default:
+        return readCustomObject();
+    }
+  }
+
+  protected Class<?> readClass() {
+    String className = readString();
+    try {
+      return Class.forName(className);
+    } catch (ClassNotFoundException ex) {
+      throw new IllegalStateException(ex);
+    }
+  }
+
+  protected Class<?> loadClass(String className) throws ClassNotFoundException {
+    return Class.forName(className);
+  }
+
+  protected Object readCustomObject() {
+    Class<?> objectClass = readClass();
+    return customMarshallers.stream()
+        .filter(m -> m.getObjectClass().isAssignableFrom(objectClass))
+        .findFirst()
+        .map(m -> m.read(this))
+        .orElseThrow(() -> new IllegalArgumentException("Unsupported type " + objectClass));
+  }
+}

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

@@ -0,0 +1,126 @@
+/*
+ * 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.marshaller;
+
+import java.time.Instant;
+import java.util.Collection;
+import java.util.Map;
+
+public abstract class AbstractOutputBuffer implements WorkflowOutputBuffer {
+
+  private final Collection<CustomObjectMarshaller> customMarshallers;
+
+  protected AbstractOutputBuffer(Collection<CustomObjectMarshaller> customMarshallers) {
+    this.customMarshallers = customMarshallers;
+  }
+
+  @Override
+  public WorkflowOutputBuffer writeInstant(Instant instant) {
+    writeLong(instant.getEpochSecond());
+    return this;
+  }
+
+  @Override
+  public <T extends Enum<T>> WorkflowOutputBuffer writeEnum(T value) {
+    writeString(value.name());
+    return this;
+  }
+
+  @Override
+  public WorkflowOutputBuffer writeMap(Map<String, Object> map) {
+    writeInt(map.size());
+    map.forEach(
+        (k, v) -> {
+          writeString(k);
+          writeObject(v);
+        });
+
+    return this;
+  }
+
+  @Override
+  public WorkflowOutputBuffer writeCollection(Collection<Object> col) {
+    writeInt(col.size());
+    col.forEach(this::writeObject);
+    return this;
+  }
+
+  @Override
+  public WorkflowOutputBuffer writeObject(Object object) {
+    if (object == null) {
+      writeType(Type.NULL);
+    } else if (object instanceof Short number) {
+      writeType(Type.SHORT);
+      writeShort(number);
+    } else if (object instanceof Integer number) {
+      writeType(Type.INT);
+      writeInt(number);
+    } else if (object instanceof Long number) {
+      writeType(Type.LONG);
+      writeLong(number);
+    } else if (object instanceof Byte number) {
+      writeType(Type.BYTE);
+      writeLong(number);
+    } else if (object instanceof Float number) {
+      writeType(Type.FLOAT);
+      writeFloat(number);
+    } else if (object instanceof Double number) {
+      writeType(Type.DOUBLE);
+      writeDouble(number);
+    } else if (object instanceof Boolean bool) {
+      writeType(Type.BOOLEAN);
+      writeBoolean(bool);
+    } else if (object instanceof String str) {
+      writeType(Type.STRING);
+      writeString(str);
+    } else if (object instanceof Map value) {
+      writeType(Type.MAP);
+      writeMap(value);
+    } else if (object instanceof Collection value) {
+      writeType(Type.COLLECTION);
+      writeCollection(value);
+    } else if (object instanceof Instant value) {
+      writeType(Type.INSTANT);
+      writeInstant(value);
+    } else if (object instanceof byte[] bytes) {
+      writeType(Type.BYTES);
+      writeBytes(bytes);
+    } else {
+      writeCustomObject(object);
+    }
+    return this;
+  }
+
+  protected void writeClass(Class<?> objectClass) {
+    writeString(objectClass.getCanonicalName());
+  }
+
+  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()));
+  }
+
+  protected void writeType(Type type) {
+    writeByte((byte) type.ordinal());
+  }
+}