[Fix #1266] Marshaling improvements

experimental/model/pom.xml

@@ -8,9 +8,21 @@
   <artifactId>serverlessworkflow-experimental-model</artifactId>
   <name>Serverless Workflow :: Experimental :: Model</name>
   <dependencies>
-   <dependency>
-            <groupId>io.serverlessworkflow</groupId>
-            <artifactId>serverlessworkflow-impl-core</artifactId>
-        </dependency>
+    <dependency>
+        <groupId>io.serverlessworkflow</groupId>
+        <artifactId>serverlessworkflow-impl-core</artifactId>
+    </dependency>
+    <dependency>
+        <groupId>org.junit.jupiter</groupId>
+        <artifactId>junit-jupiter-engine</artifactId>
+    </dependency>
+    <dependency>
+        <groupId>org.assertj</groupId>
+        <artifactId>assertj-core</artifactId>
+    </dependency>
+    <dependency>
+        <groupId>ch.qos.logback</groupId>
+        <artifactId>logback-classic</artifactId>
+    </dependency>
   </dependencies>
 </project>

experimental/model/src/main/java/io/serverlessworkflow/impl/model/func/JavaModel.java

@@ -22,6 +22,7 @@
 import java.util.Collections;
 import java.util.Map;
 import java.util.Map.Entry;
+import java.util.Objects;
 import java.util.Optional;
 import java.util.stream.Collectors;
 
@@ -104,4 +105,18 @@ protected <T> Optional<T> convert(Class<T> clazz) {
         ? Optional.of(clazz.cast(object))
         : Optional.empty();
   }
+
+  @Override
+  public int hashCode() {
+    return Objects.hash(object);
+  }
+
+  @Override
+  public boolean equals(Object obj) {
+    if (this == obj) return true;
+    if (obj == null) return false;
+    if (getClass() != obj.getClass()) return false;
+    JavaModel other = (JavaModel) obj;
+    return Objects.equals(object, other.object);
+  }
 }

experimental/model/src/main/java/io/serverlessworkflow/impl/model/func/JavaModelMarshaller.java

@@ -0,0 +1,43 @@
+/*
+ * 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.model.func;
+
+import io.serverlessworkflow.impl.marshaller.CustomObjectMarshaller;
+import io.serverlessworkflow.impl.marshaller.WorkflowInputBuffer;
+import io.serverlessworkflow.impl.marshaller.WorkflowOutputBuffer;
+
+public class JavaModelMarshaller implements CustomObjectMarshaller<JavaModel> {
+
+  @Override
+  public void write(WorkflowOutputBuffer buffer, JavaModel object) {
+    buffer.writeObject(object.asJavaObject());
+  }
+
+  @Override
+  public JavaModel read(WorkflowInputBuffer buffer, Class<? extends JavaModel> clazz) {
+    return new JavaModel(buffer.readObject());
+  }
+
+  @Override
+  public Class<JavaModel> getObjectClass() {
+    return JavaModel.class;
+  }
+
+  @Override
+  public int priority() {
+    return Integer.MAX_VALUE;
+  }
+}

experimental/model/src/main/java/io/serverlessworkflow/impl/model/func/SerializableObjectMarshaller.java

@@ -0,0 +1,63 @@
+/*
+ * 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.model.func;
+
+import io.serverlessworkflow.impl.marshaller.CustomObjectMarshaller;
+import io.serverlessworkflow.impl.marshaller.WorkflowInputBuffer;
+import io.serverlessworkflow.impl.marshaller.WorkflowOutputBuffer;
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.io.ObjectInputStream;
+import java.io.ObjectOutputStream;
+import java.io.Serializable;
+import java.io.UncheckedIOException;
+
+public class SerializableObjectMarshaller implements CustomObjectMarshaller<Serializable> {
+
+  @Override
+  public void write(WorkflowOutputBuffer buffer, Serializable object) {
+    try (ByteArrayOutputStream bytesOut = new ByteArrayOutputStream();
+        ObjectOutputStream out = new ObjectOutputStream(bytesOut)) {
+      out.writeObject(object);
+      buffer.writeBytes(bytesOut.toByteArray());
+    } catch (IOException io) {
+      throw new UncheckedIOException(io);
+    }
+  }
+
+  @Override
+  public Serializable read(WorkflowInputBuffer buffer, Class<? extends Serializable> objectClass) {
+    try (ByteArrayInputStream bytesIn = new ByteArrayInputStream(buffer.readBytes());
+        ObjectInputStream in = new ObjectInputStream(bytesIn)) {
+      return objectClass.cast(in.readObject());
+    } catch (IOException io) {
+      throw new UncheckedIOException(io);
+    } catch (ClassNotFoundException ex) {
+      throw new IllegalStateException(ex);
+    }
+  }
+
+  @Override
+  public Class<Serializable> getObjectClass() {
+    return Serializable.class;
+  }
+
+  @Override
+  public int priority() {
+    return Integer.MAX_VALUE;
+  }
+}

experimental/model/src/main/resources/META-INF/services/io.serverlessworkflow.impl.marshaller.CustomObjectMarshaller

@@ -0,0 +1,2 @@
+io.serverlessworkflow.impl.model.func.JavaModelMarshaller
+io.serverlessworkflow.impl.model.func.SerializableObjectMarshaller

experimental/model/src/test/java/io/serverlessworkflow/impl/model/func/Address.java

@@ -0,0 +1,20 @@
+/*
+ * 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.model.func;
+
+import java.io.Serializable;
+
+record Address(String street, int number) implements Serializable {}

experimental/model/src/test/java/io/serverlessworkflow/impl/model/func/JavaModelSerializationTest.java

@@ -0,0 +1,48 @@
+/*
+ * 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.model.func;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import io.serverlessworkflow.impl.marshaller.DefaultBufferFactory;
+import io.serverlessworkflow.impl.marshaller.WorkflowBufferFactory;
+import io.serverlessworkflow.impl.marshaller.WorkflowInputBuffer;
+import io.serverlessworkflow.impl.marshaller.WorkflowOutputBuffer;
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import org.junit.jupiter.api.Test;
+
+class JavaModelSerializationTest {
+
+  @Test
+  void testSerializableJavaModel() throws IOException {
+    testMarshallUnMarshall(
+        new JavaModel(new Person("Pepe Gotera", 32, new Address("Rue del Percebe", 13))));
+  }
+
+  private void testMarshallUnMarshall(Object object) {
+    WorkflowBufferFactory factory = DefaultBufferFactory.factory();
+    ByteArrayOutputStream output = new ByteArrayOutputStream();
+    try (WorkflowOutputBuffer writer = factory.output(output)) {
+      writer.writeObject(object);
+    }
+    ByteArrayInputStream input = new ByteArrayInputStream(output.toByteArray());
+    try (WorkflowInputBuffer reader = factory.input(input)) {
+      assertThat(reader.readObject()).isEqualTo(object);
+    }
+  }
+}

experimental/model/src/test/java/io/serverlessworkflow/impl/model/func/Person.java

@@ -0,0 +1,20 @@
+/*
+ * 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.model.func;
+
+import java.io.Serializable;
+
+record Person(String name, int age, Address address) implements Serializable {}

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

@@ -131,10 +131,7 @@ protected Class<?> loadClass(String className) throws ClassNotFoundException {
 
   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));
+    return MarshallingUtils.getCustomMarshaller(customMarshallers, objectClass)
+        .read(this, objectClass);
   }
 }

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

@@ -73,7 +73,7 @@ public WorkflowOutputBuffer writeObject(Object object) {
       writeLong(number);
     } else if (object instanceof Byte number) {
       writeType(Type.BYTE);
-      writeLong(number);
+      writeByte(number);
     } else if (object instanceof Float number) {
       writeType(Type.FLOAT);
       writeFloat(number);
@@ -109,14 +109,11 @@ protected void writeClass(Class<?> objectClass) {
     writeString(objectClass.getCanonicalName());
   }
 
+  @SuppressWarnings({"rawtypes", "unchecked"})
   protected void writeCustomObject(Object object) {
     CustomObjectMarshaller marshaller =
-        customMarshallers.stream()
-            .filter(m -> m.getObjectClass().isAssignableFrom(object.getClass()))
-            .findFirst()
-            .orElseThrow(
-                () -> new IllegalArgumentException("Unsupported type " + object.getClass()));
-    writeClass(marshaller.getObjectClass());
+        MarshallingUtils.getCustomMarshaller(customMarshallers, object.getClass());
+    writeClass(object.getClass());
     marshaller.write(this, marshaller.getObjectClass().cast(object));
   }
 

impl/persistence/api/src/main/java/io/serverlessworkflow/impl/marshaller/CustomObjectMarshaller.java → impl/core/src/main/java/io/serverlessworkflow/impl/marshaller/CustomObjectMarshaller.java

@@ -20,7 +20,7 @@
 public interface CustomObjectMarshaller<T> extends ServicePriority {
   void write(WorkflowOutputBuffer buffer, T object);
 
-  T read(WorkflowInputBuffer buffer);
+  T read(WorkflowInputBuffer buffer, Class<? extends T> clazz);
 
   Class<T> getObjectClass();
 }

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

@@ -29,6 +29,7 @@ private static class DefaultBufferFactoryHolder {
         new DefaultBufferFactory(
             ServiceLoader.load(CustomObjectMarshaller.class).stream()
                 .map(ServiceLoader.Provider::get)
+                .sorted()
                 .toList());
   }
 

impl/persistence/api/src/main/java/io/serverlessworkflow/impl/marshaller/MarshallingUtils.java → impl/core/src/main/java/io/serverlessworkflow/impl/marshaller/MarshallingUtils.java

@@ -19,6 +19,7 @@
 import java.io.ByteArrayInputStream;
 import java.io.ByteArrayOutputStream;
 import java.time.Instant;
+import java.util.Collection;
 import java.util.function.BiConsumer;
 import java.util.function.Function;
 
@@ -95,4 +96,32 @@ private static <T> T readValue(
       return valueConsumer.apply(buffer);
     }
   }
+
+  /**
+   * Retrieve more proper marshaler for the given class. Collection is assumed to be already sorted
+   * by priority. No matter which priority is given, if the object class is equal to the marshaler
+   * class, it should have precedence over an object class which is assignable to the marshaler
+   * class.
+   *
+   * @param marshallers Priority Sorted collection of marshalers available on classpath
+   * @param clazz The class of the object being marshaled
+   * @return The most suitable marshaler for that object class
+   * @throws IllegalArgumentException if no marshaler is found for that object class
+   */
+  @SuppressWarnings({"rawtypes", "unchecked"})
+  public static CustomObjectMarshaller getCustomMarshaller(
+      Collection<CustomObjectMarshaller> marshallers, Class clazz) {
+    CustomObjectMarshaller assignable = null;
+    for (CustomObjectMarshaller marshaller : marshallers) {
+      if (marshaller.getObjectClass().equals(clazz)) {
+        return marshaller;
+      } else if (marshaller.getObjectClass().isAssignableFrom(clazz) && assignable == null) {
+        assignable = marshaller;
+      }
+    }
+    if (assignable == null) {
+      throw new IllegalArgumentException("Cannot find proper marshaler for class " + clazz);
+    }
+    return assignable;
+  }
 }

impl/core/src/test/java/io/serverlessworkflow/impl/marshaller/Employee.java

@@ -0,0 +1,18 @@
+/*
+ * 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;
+
+class Employee extends Person {}