[Fix #1395] persistence helper classes for allcorrelation (#1396)

Signed-off-by: fjtirado <ftirados@redhat.com>

Author: fjtirado

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

@@ -444,7 +444,8 @@ public WorkflowApplication build() {
                 .orElseGet(() -> new DefaultCloudEventPredicateFactory());
       }
       if (allStrategyCorrelationInfoFactory == null) {
-        allStrategyCorrelationInfoFactory = definition -> new InMemoryAllStrategyCorrelationInfo();
+        allStrategyCorrelationInfoFactory =
+            definition -> InMemoryAllStrategyCorrelationInfo.instance();
       }
 
       if (defaultCatalogURI == null) {

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

@@ -16,6 +16,7 @@
 package io.serverlessworkflow.impl;
 
 import java.util.concurrent.CompletableFuture;
+import java.util.function.Supplier;
 
 public interface WorkflowInstance extends WorkflowInstanceData {
   CompletableFuture<WorkflowModel> start();
@@ -49,4 +50,6 @@ public interface WorkflowInstance extends WorkflowInstanceData {
   boolean cancel();
 
   boolean resume();
+
+  <T> T addMetadataIfAbsent(String key, Supplier<T> supplier);
 }

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

@@ -16,6 +16,7 @@
 package io.serverlessworkflow.impl;
 
 import java.time.Instant;
+import java.util.Optional;
 
 public interface WorkflowInstanceData {
   String id();
@@ -29,4 +30,6 @@ public interface WorkflowInstanceData {
   WorkflowStatus status();
 
   WorkflowModel context();
+
+  <T> Optional<T> findMetadata(String key, Class<T> objectClass);
 }

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

@@ -350,9 +350,16 @@ public void addCancelable(CompletableFuture<?> cancelable) {
     }
   }
 
-  public <T> T additionalObject(String key, Supplier<T> supplier) {
+  @Override
+  public <T> T addMetadataIfAbsent(String key, Supplier<T> supplier) {
     return (T) additionalObjects.computeIfAbsent(key, k -> supplier.get());
   }
 
+  @Override
+  public <T> Optional<T> findMetadata(String key, Class<T> objectClass) {
+    Object value = additionalObjects.get(key);
+    return objectClass.isInstance(value) ? Optional.of(objectClass.cast(value)) : Optional.empty();
+  }
+
   public void restoreContext(WorkflowContext workflow, TaskContext context) {}
 }

impl/core/src/main/java/io/serverlessworkflow/impl/marshaller/AbstractInputBuffer.java

@@ -15,7 +15,10 @@
  */
 package io.serverlessworkflow.impl.marshaller;
 
+import java.net.URI;
 import java.time.Instant;
+import java.time.OffsetDateTime;
+import java.time.ZoneOffset;
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.LinkedHashMap;
@@ -112,6 +115,12 @@ public Object readObject() {
       case CUSTOM:
         return readCustomObject();
 
+      case URI:
+        return URI.create(readString());
+
+      case OFFSET_DATE_TIME:
+        return OffsetDateTime.ofInstant(readInstant(), ZoneOffset.of(readString()));
+
       default:
         throw new IllegalStateException("Unsupported type " + type);
     }

impl/core/src/main/java/io/serverlessworkflow/impl/marshaller/AbstractOutputBuffer.java

@@ -16,7 +16,9 @@
 package io.serverlessworkflow.impl.marshaller;
 
 import io.serverlessworkflow.impl.WorkflowModel;
+import java.net.URI;
 import java.time.Instant;
+import java.time.OffsetDateTime;
 import java.util.Collection;
 import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
@@ -100,9 +102,16 @@ public WorkflowOutputBuffer writeObject(Object object) {
     } else if (object instanceof Instant value) {
       writeType(Type.INSTANT);
       writeInstant(value);
-    } else if (object instanceof byte[] bytes) {
+    } else if (object instanceof OffsetDateTime value) {
+      writeType(Type.OFFSET_DATE_TIME);
+      writeInstant(value.toInstant());
+      writeString(value.getOffset().toString());
+    } else if (object instanceof URI value) {
+      writeType(Type.URI);
+      writeString(value.toString());
+    } else if (object instanceof byte[] value) {
       writeType(Type.BYTES);
-      writeBytes(bytes);
+      writeBytes(value);
     } else {
       internalWriteObject(object);
     }

impl/core/src/main/java/io/serverlessworkflow/impl/marshaller/MarshallingUtils.java

@@ -15,14 +15,21 @@
  */
 package io.serverlessworkflow.impl.marshaller;
 
+import io.cloudevents.CloudEvent;
+import io.cloudevents.core.builder.CloudEventBuilder;
 import io.serverlessworkflow.impl.WorkflowModel;
 import java.io.ByteArrayInputStream;
 import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.io.UncheckedIOException;
 import java.lang.reflect.Modifier;
+import java.net.URI;
 import java.time.Instant;
+import java.time.OffsetDateTime;
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.List;
+import java.util.Set;
 import java.util.function.BiConsumer;
 import java.util.function.Function;
 import org.slf4j.Logger;
@@ -71,6 +78,58 @@ public static byte[] writeString(WorkflowBufferFactory factory, String value) {
     return writeValue(factory, value, (b, v) -> b.writeString(v));
   }
 
+  public static byte[] writeCloudEventExtensions(WorkflowBufferFactory factory, CloudEvent event) {
+    try (ByteArrayOutputStream bytesOut = new ByteArrayOutputStream();
+        WorkflowOutputBuffer out = factory.output(bytesOut)) {
+      writeCloudEventExtensions(out, event);
+      return bytesOut.toByteArray();
+    } catch (IOException e) {
+      throw new UncheckedIOException(e);
+    }
+  }
+
+  public static void writeCloudEventExtensions(WorkflowOutputBuffer out, CloudEvent event) {
+    Set<String> extensionNames = event.getExtensionNames();
+    out.writeInt(extensionNames.size());
+    for (String extensionName : extensionNames) {
+      out.writeString(extensionName);
+      out.writeObject(event.getExtension(extensionName));
+    }
+  }
+
+  public static CloudEventBuilder readCloudEventExtensions(
+      WorkflowBufferFactory factory, byte[] value, CloudEventBuilder builder) {
+    try (ByteArrayInputStream bytesInt = new ByteArrayInputStream(value);
+        WorkflowInputBuffer in = factory.input(bytesInt)) {
+      return readCloudEventExtenstions(in, value, builder);
+    } catch (IOException e) {
+      throw new UncheckedIOException(e);
+    }
+  }
+
+  public static CloudEventBuilder readCloudEventExtenstions(
+      WorkflowInputBuffer in, byte[] value, CloudEventBuilder builder) {
+    int size = in.readInt();
+    while (size-- > 0) {
+      String extensionName = in.readString();
+      Object extensionValue = in.readObject();
+      if (extensionValue instanceof Number extValue) {
+        builder.withExtension(extensionName, extValue);
+      } else if (extensionValue instanceof String extValue) {
+        builder.withExtension(extensionName, extValue);
+      } else if (extensionValue instanceof Boolean extValue) {
+        builder.withExtension(extensionName, extValue);
+      } else if (extensionValue instanceof byte[] extValue) {
+        builder.withExtension(extensionName, extValue);
+      } else if (extensionValue instanceof OffsetDateTime extValue) {
+        builder.withExtension(extensionName, extValue);
+      } else if (extensionValue instanceof URI extValue) {
+        builder.withExtension(extensionName, extValue);
+      }
+    }
+    return builder;
+  }
+
   public static String readString(WorkflowBufferFactory factory, byte[] value) {
     return readValue(factory, value, WorkflowInputBuffer::readString);
   }

impl/core/src/main/java/io/serverlessworkflow/impl/marshaller/Type.java

@@ -29,5 +29,7 @@ enum Type {
   MAP,
   COLLECTION,
   NULL,
-  CUSTOM
+  CUSTOM,
+  OFFSET_DATE_TIME,
+  URI
 }

impl/core/src/main/java/io/serverlessworkflow/impl/scheduler/AllStrategyCorrelationInfo.java

@@ -16,15 +16,22 @@
 package io.serverlessworkflow.impl.scheduler;
 
 import io.cloudevents.CloudEvent;
+import io.serverlessworkflow.impl.WorkflowInstance;
 import io.serverlessworkflow.impl.events.EventRegistrationBuilder;
 import java.util.Collection;
+import java.util.Map;
 import java.util.function.Consumer;
 
 public interface AllStrategyCorrelationInfo extends AutoCloseable {
-  void correlate(
-      EventRegistrationBuilder reg, CloudEvent event, Consumer<Collection<CloudEvent>> starter);
 
-  void register(EventRegistrationBuilder reg);
+  void init(
+      Collection<EventRegistrationBuilder> reg,
+      Consumer<Map<EventRegistrationBuilder, CloudEvent>> starter);
+
+  void correlate(EventRegistrationBuilder reg, CloudEvent event);
+
+  default void addMetadata(
+      WorkflowInstance instance, Map<EventRegistrationBuilder, CloudEvent> events) {}
 
   default void close() {}
 }

impl/core/src/main/java/io/serverlessworkflow/impl/scheduler/DefaultWorkflowScheduler.java

@@ -16,7 +16,7 @@
 package io.serverlessworkflow.impl.scheduler;
 
 import io.serverlessworkflow.impl.WorkflowDefinition;
-import io.serverlessworkflow.impl.WorkflowModel;
+import io.serverlessworkflow.impl.WorkflowInstance;
 import java.util.concurrent.Executors;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.ScheduledFuture;
@@ -83,10 +83,10 @@ protected CronResolverIntanceRunner(WorkflowDefinition definition) {
       }
 
       @Override
-      public void accept(WorkflowModel model) {
+      public void accept(WorkflowInstance instance) {
         if (!cancelled.get()) {
           scheduleNext();
-          super.accept(model);
+          super.accept(instance);
         }
       }
     }