[Fix #1395] AllStrategyCorrelation persistence

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);
         }
       }
     }

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

@@ -26,38 +26,50 @@
 
 public class InMemoryAllStrategyCorrelationInfo implements AllStrategyCorrelationInfo {
 
+  private static class InMemoryAllStrategyCorrelationInfoHolder {
+    private static InMemoryAllStrategyCorrelationInfo INSTANCE =
+        new InMemoryAllStrategyCorrelationInfo();
+  }
+
+  public static AllStrategyCorrelationInfo instance() {
+    return InMemoryAllStrategyCorrelationInfoHolder.INSTANCE;
+  }
+
+  private InMemoryAllStrategyCorrelationInfo() {}
+
   private Map<EventRegistrationBuilder, List<CloudEvent>> correlatedEvents;
+  private Consumer<Map<EventRegistrationBuilder, CloudEvent>> starter;
 
   @Override
-  public void correlate(
-      EventRegistrationBuilder reg, CloudEvent event, Consumer<Collection<CloudEvent>> starter) {
-    Collection<CloudEvent> collection = new ArrayList<>();
+  public void correlate(EventRegistrationBuilder reg, CloudEvent event) {
+    Map<EventRegistrationBuilder, CloudEvent> result = new HashMap<>();
     // to minimize the critical section, conversion is done later, here we are
     // performing just collection, if any
     synchronized (correlatedEvents) {
       correlatedEvents.get(reg).add(event);
-      Collection<List<CloudEvent>> events = correlatedEvents.values();
-      if (satisfyCondition(events)) {
-        for (List<CloudEvent> values : events) {
-          collection.add(values.remove(0));
+      if (satisfyCondition(correlatedEvents)) {
+        for (java.util.Map.Entry<EventRegistrationBuilder, List<CloudEvent>> values :
+            correlatedEvents.entrySet()) {
+          result.put(values.getKey(), values.getValue().remove(0));
         }
       }
     }
-    if (!collection.isEmpty()) {
-      starter.accept(collection);
+    if (!result.isEmpty()) {
+      starter.accept(result);
     }
   }
 
   @Override
-  public void register(EventRegistrationBuilder reg) {
-    if (correlatedEvents == null) {
-      correlatedEvents = new HashMap<>();
-    }
-    correlatedEvents.put(reg, new ArrayList<CloudEvent>());
+  public void init(
+      Collection<EventRegistrationBuilder> regs,
+      Consumer<Map<EventRegistrationBuilder, CloudEvent>> starter) {
+    correlatedEvents = new HashMap<>();
+    this.starter = starter;
+    regs.forEach(reg -> correlatedEvents.put(reg, new ArrayList<CloudEvent>()));
   }
 
-  private boolean satisfyCondition(Collection<List<CloudEvent>> events) {
-    for (List<CloudEvent> values : events) {
+  private boolean satisfyCondition(Map<EventRegistrationBuilder, List<CloudEvent>> events) {
+    for (List<CloudEvent> values : events.values()) {
       if (values.isEmpty()) {
         return false;
       }

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

@@ -19,13 +19,16 @@
 
 import io.cloudevents.CloudEvent;
 import io.serverlessworkflow.impl.WorkflowDefinition;
+import io.serverlessworkflow.impl.WorkflowInstance;
 import io.serverlessworkflow.impl.WorkflowModel;
 import io.serverlessworkflow.impl.WorkflowModelCollection;
 import io.serverlessworkflow.impl.events.EventConsumer;
 import io.serverlessworkflow.impl.events.EventRegistration;
+import io.serverlessworkflow.impl.events.EventRegistrationBuilder;
 import io.serverlessworkflow.impl.events.EventRegistrationBuilderInfo;
 import java.util.ArrayList;
 import java.util.Collection;
+import java.util.Map;
 import java.util.function.Function;
 
 public class ScheduledEventConsumer implements AutoCloseable {
@@ -53,19 +56,15 @@ public ScheduledEventConsumer(
         && builderInfo.registrations().registrations().size() > 1) {
       this.allStrategyCorrelationInfo =
           definition.application().allStrategyCorrelationInfoFactory().apply(definition);
-      builderInfo
-          .registrations()
-          .registrations()
-          .forEach(
-              reg -> {
-                allStrategyCorrelationInfo.register(reg);
-                registrations.add(
-                    eventConsumer.register(
-                        reg,
-                        ce ->
-                            allStrategyCorrelationInfo.correlate(
-                                reg, (CloudEvent) ce, this::start)));
-              });
+      Collection<EventRegistrationBuilder> registrationBuilders =
+          builderInfo.registrations().registrations();
+      allStrategyCorrelationInfo.init(registrationBuilders, this::start);
+      registrationBuilders.forEach(
+          reg -> {
+            registrations.add(
+                eventConsumer.register(
+                    reg, ce -> allStrategyCorrelationInfo.correlate(reg, (CloudEvent) ce)));
+          });
     } else {
       builderInfo
           .registrations()
@@ -78,13 +77,15 @@ public ScheduledEventConsumer(
   protected void start(CloudEvent ce) {
     WorkflowModelCollection model = definition.application().modelFactory().createCollection();
     model.add(converter.apply(ce));
-    instanceRunner.accept(model);
+    instanceRunner.accept(definition.instance(model));
   }
 
-  protected void start(Collection<CloudEvent> ces) {
+  protected void start(Map<EventRegistrationBuilder, CloudEvent> ces) {
     WorkflowModelCollection model = definition.application().modelFactory().createCollection();
-    ces.forEach(ce -> model.add(converter.apply(ce)));
-    instanceRunner.accept(model);
+    ces.values().forEach(ce -> model.add(converter.apply(ce)));
+    WorkflowInstance instance = definition.instance(model);
+    allStrategyCorrelationInfo.addMetadata(instance, ces);
+    instanceRunner.accept(instance);
   }
 
   public void close() {