[Fix #782] Refining approach

Author: fjtirado

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;

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

@@ -0,0 +1,142 @@
+/*
+ * 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;
+  }
+
+  private enum TYPE {
+    BYTE,
+    BYTES,
+    INT,
+    SHORT,
+    LONG,
+    FLOAT,
+    DOUBLE,
+    BOOLEAN,
+    STRING,
+    INSTANT,
+    MAP,
+    COLLECTION,
+    NULL
+  }
+
+  @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()));
+  }
+
+  private void writeType(TYPE type) {
+    writeByte((byte) type.ordinal());
+  }
+}

impl/persistence/api/src/main/java/io/serverlessworkflow/impl/persistence/WorkflowMinimumPersistenceReader.java renamed to impl/persistence/api/src/main/java/io/serverlessworkflow/impl/marshaller/CustomObjectMarshaller.java

@@ -13,18 +13,12 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package io.serverlessworkflow.impl.persistence;
+package io.serverlessworkflow.impl.marshaller;
 
-import io.serverlessworkflow.impl.WorkflowDefinition;
-import io.serverlessworkflow.impl.WorkflowInstance;
-import java.util.stream.Stream;
+public interface CustomObjectMarshaller<T> {
+  void write(WorkflowOutputBuffer buffer, T object);
 
-public interface WorkflowMinimumPersistenceReader extends AutoCloseable {
+  T read(WorkflowInputBuffer buffer);
 
-  /**
-   * Allow streaming over all stored workflow instances for a certain definition
-   *
-   * @return
-   */
-  Stream<WorkflowInstance> all(WorkflowDefinition definition);
+  Class<T> getObjectClass();
 }

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

@@ -13,20 +13,27 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package io.serverlessworkflow.impl.persistence.bigmap;
+package io.serverlessworkflow.impl.marshaller;
 
 import java.io.InputStream;
 import java.io.OutputStream;
+import java.util.Collection;
 
 public class DefaultBufferFactory implements WorkflowBufferFactory {
 
+  private final Collection<CustomObjectMarshaller> marshallers;
+
+  public DefaultBufferFactory(Collection<CustomObjectMarshaller> marshallers) {
+    this.marshallers = marshallers;
+  }
+
   @Override
   public WorkflowInputBuffer input(InputStream input) {
-    return new DefaultInputBuffer(input);
+    return new DefaultInputBuffer(input, marshallers);
   }
 
   @Override
   public WorkflowOutputBuffer output(OutputStream output) {
-    return new DefaultOutputBuffer(output);
+    return new DefaultOutputBuffer(output, marshallers);
   }
 }

impl/persistence/bigmap/src/main/java/io/serverlessworkflow/impl/persistence/bigmap/DefaultInputBuffer.java renamed to impl/persistence/api/src/main/java/io/serverlessworkflow/impl/marshaller/DefaultInputBuffer.java

@@ -13,19 +13,20 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package io.serverlessworkflow.impl.persistence.bigmap;
+package io.serverlessworkflow.impl.marshaller;
 
 import java.io.DataInputStream;
 import java.io.IOException;
 import java.io.InputStream;
 import java.io.UncheckedIOException;
 import java.time.Instant;
+import java.util.Collection;
 
 public class DefaultInputBuffer implements WorkflowInputBuffer {
 
   private DataInputStream input;
 
-  public DefaultInputBuffer(InputStream in) {
+  public DefaultInputBuffer(InputStream in, Collection<CustomObjectMarshaller> marshallers) {
     input = new DataInputStream(in);
   }
 

impl/persistence/bigmap/src/main/java/io/serverlessworkflow/impl/persistence/bigmap/DefaultOutputBuffer.java renamed to impl/persistence/api/src/main/java/io/serverlessworkflow/impl/marshaller/DefaultOutputBuffer.java

@@ -13,19 +13,21 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package io.serverlessworkflow.impl.persistence.bigmap;
+package io.serverlessworkflow.impl.marshaller;
 
 import java.io.DataOutputStream;
 import java.io.IOException;
 import java.io.OutputStream;
 import java.io.UncheckedIOException;
-import java.time.Instant;
+import java.util.Collection;
 
-public class DefaultOutputBuffer implements WorkflowOutputBuffer {
+public class DefaultOutputBuffer extends AbstractOutputBuffer {
 
   private DataOutputStream output;
 
-  public DefaultOutputBuffer(OutputStream out) {
+  public DefaultOutputBuffer(
+      OutputStream out, Collection<CustomObjectMarshaller> customMarshallers) {
+    super(customMarshallers);
     output = new DataOutputStream(out);
   }
 
@@ -103,7 +105,6 @@ public WorkflowOutputBuffer writeBoolean(boolean bool) {
   public WorkflowOutputBuffer writeByte(byte one) {
     try {
       output.writeByte(one);
-      ;
     } catch (IOException e) {
       throw new UncheckedIOException(e);
     }
@@ -121,18 +122,6 @@ public WorkflowOutputBuffer writeBytes(byte[] bytes) {
     return this;
   }
 
-  @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 void close() {
     try {

impl/persistence/bigmap/src/main/java/io/serverlessworkflow/impl/persistence/bigmap/WorkflowBufferFactory.java renamed to impl/persistence/api/src/main/java/io/serverlessworkflow/impl/marshaller/WorkflowBufferFactory.java

@@ -13,7 +13,7 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package io.serverlessworkflow.impl.persistence.bigmap;
+package io.serverlessworkflow.impl.marshaller;
 
 import java.io.InputStream;
 import java.io.OutputStream;

impl/persistence/bigmap/src/main/java/io/serverlessworkflow/impl/persistence/bigmap/WorkflowInputBuffer.java renamed to impl/persistence/api/src/main/java/io/serverlessworkflow/impl/marshaller/WorkflowInputBuffer.java

@@ -13,7 +13,7 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package io.serverlessworkflow.impl.persistence.bigmap;
+package io.serverlessworkflow.impl.marshaller;
 
 import java.io.Closeable;
 import java.time.Instant;

impl/persistence/bigmap/src/main/java/io/serverlessworkflow/impl/persistence/bigmap/WorkflowOutputBuffer.java renamed to impl/persistence/api/src/main/java/io/serverlessworkflow/impl/marshaller/WorkflowOutputBuffer.java

@@ -13,12 +13,13 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package io.serverlessworkflow.impl.persistence.bigmap;
+package io.serverlessworkflow.impl.marshaller;
 
-import java.io.Closeable;
 import java.time.Instant;
+import java.util.Collection;
+import java.util.Map;
 
-public interface WorkflowOutputBuffer extends Closeable {
+public interface WorkflowOutputBuffer extends AutoCloseable {
 
   WorkflowOutputBuffer writeString(String text);
 
@@ -40,6 +41,12 @@ public interface WorkflowOutputBuffer extends Closeable {
 
   WorkflowOutputBuffer writeInstant(Instant instant);
 
+  WorkflowOutputBuffer writeMap(Map<String, Object> map);
+
+  WorkflowOutputBuffer writeCollection(Collection<Object> col);
+
+  WorkflowOutputBuffer writeObject(Object object);
+
   <T extends Enum<T>> WorkflowOutputBuffer writeEnum(T value);
 
   void close();