Enable any type of collection cast on *ModelCollection

Signed-off-by: Ricardo Zanini <ricardozanini@gmail.com>

Author: ricardozanini

experimental/lambda/src/test/java/io/serverless/workflow/impl/executors/func/EventFilteringTest.java

@@ -38,7 +38,7 @@
 import io.serverlessworkflow.impl.events.EventPublisher;
 import java.net.URI;
 import java.nio.charset.StandardCharsets;
-import java.util.ArrayList;
+import java.util.List;
 import java.util.concurrent.CompletableFuture;
 import org.junit.jupiter.api.Test;
 
@@ -74,7 +74,6 @@ public void testIntelligentNewsletterApprovalPath() throws Exception {
     try (WorkflowApplication app =
         WorkflowApplication.builder().withListener(new TraceExecutionListener()).build()) {
 
-      // 1. Build the workflow using your exact DSL
       Workflow workflow =
           FuncWorkflowBuilder.workflow("intelligent-newsletter")
               .tasks(
@@ -86,11 +85,9 @@ public void testIntelligentNewsletterApprovalPath() throws Exception {
                                   consumed("org.acme.newsletter.review.done")
                                       .extensionByInstanceId("instanceid")))
                       .outputAs(
-                          (ArrayList events) -> {
+                          (List<CloudEventData> events) -> {
                             try {
-                              return MAPPER.readValue(
-                                  ((CloudEventData) events.iterator().next()).toBytes(),
-                                  HumanReview.class);
+                              return MAPPER.readValue(events.get(0).toBytes(), HumanReview.class);
                             } catch (Exception e) {
                               throw new RuntimeException("Failed to deserialize HumanReview", e);
                             }

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

@@ -17,10 +17,14 @@
 
 import io.serverlessworkflow.impl.WorkflowModel;
 import io.serverlessworkflow.impl.WorkflowModelCollection;
+import java.lang.reflect.Array;
 import java.util.ArrayList;
 import java.util.Collection;
+import java.util.HashSet;
 import java.util.Iterator;
+import java.util.List;
 import java.util.Optional;
+import java.util.Set;
 
 public class JavaModelCollection implements Collection<WorkflowModel>, WorkflowModelCollection {
 
@@ -138,9 +142,29 @@ public Class<?> objectClass() {
   }
 
   @Override
+  @SuppressWarnings({"rawtypes", "unchecked"})
   public <T> Optional<T> as(Class<T> clazz) {
-    return object.getClass().isAssignableFrom(clazz)
-        ? Optional.of(clazz.cast(object))
-        : Optional.empty();
+    if (object == null) return Optional.empty();
+
+    if (clazz.isInstance(object)) return Optional.of(clazz.cast(object));
+
+    if (clazz.isAssignableFrom(List.class)) return Optional.of(clazz.cast(new ArrayList<>(object)));
+    else if (clazz.isAssignableFrom(Set.class))
+      return Optional.of(clazz.cast(new HashSet<>(object)));
+
+    if (clazz.isArray()) {
+      Class<?> componentType = clazz.getComponentType();
+      if (!componentType.isPrimitive()) {
+        Object[] typedArray = (Object[]) Array.newInstance(componentType, 0);
+        return Optional.of(clazz.cast(object.toArray(typedArray)));
+      }
+
+      Object primitiveArray = Array.newInstance(componentType, object.size());
+      int i = 0;
+      for (Object item : object) Array.set(primitiveArray, i++, item);
+      return Optional.of(clazz.cast(primitiveArray));
+    }
+
+    return Optional.empty();
   }
 }

experimental/test/src/test/java/io/serverlessworkflow/fluent/test/FuncEventFilterTest.java

@@ -20,22 +20,28 @@
 import static io.serverlessworkflow.fluent.func.dsl.FuncDSL.toOne;
 import static org.assertj.core.api.Assertions.assertThat;
 
+import com.fasterxml.jackson.databind.JsonNode;
 import com.fasterxml.jackson.databind.node.ArrayNode;
 import io.serverlessworkflow.api.types.Workflow;
 import io.serverlessworkflow.fluent.func.FuncWorkflowBuilder;
 import io.serverlessworkflow.impl.WorkflowApplication;
 import io.serverlessworkflow.impl.WorkflowModel;
 import java.util.Collection;
+import java.util.List;
 import java.util.Map;
+import java.util.Set;
 import java.util.concurrent.CompletableFuture;
 import org.junit.jupiter.api.Test;
 
 /**
  * Tests for the Event Filter DSL specification. Verifies that the fluent builder correctly wires
- * the payload parsing and contextual lambdas into the final Workflow definitions.
+ * the payload parsing and contextual lambdas into the final Workflow definitions, and ensures the
+ * ModelCollection adapters seamlessly convert between types.
  */
 class FuncEventFilterTest {
 
+  public record Review(String author, String text, int rating) {}
+
   @Test
   void testListenToOneCollection() {
     runIt(
@@ -47,7 +53,7 @@ void testListenToOneCollection() {
   }
 
   @Test
-  void testListenToOneNode() {
+  void testListenToOneArrayNode() {
     runIt(
         FuncWorkflowBuilder.workflow("listenToOneReviewNode")
             .tasks(
@@ -56,6 +62,36 @@ void testListenToOneNode() {
             .build());
   }
 
+  @Test
+  void testListenToOneList() {
+    runIt(
+        FuncWorkflowBuilder.workflow("listenToOneReviewList")
+            .tasks(
+                listen("waitReview", toOne("org.acme.test.review"))
+                    .outputAs((List<JsonNode> list) -> list.get(0)))
+            .build());
+  }
+
+  @Test
+  void testListenToOneSet() {
+    runIt(
+        FuncWorkflowBuilder.workflow("listenToOneReviewSet")
+            .tasks(
+                listen("waitReview", toOne("org.acme.test.review"))
+                    .outputAs((Set<JsonNode> set) -> set.iterator().next()))
+            .build());
+  }
+
+  @Test
+  void testListenToOneArray() {
+    runIt(
+        FuncWorkflowBuilder.workflow("listenToOneReviewArray")
+            .tasks(
+                listen("waitReview", toOne("org.acme.test.review"))
+                    .outputAs((JsonNode[] array) -> array[0]))
+            .build());
+  }
+
   private Workflow reviewEmitter() {
     return FuncWorkflowBuilder.workflow("emitReview")
         .tasks(emitJson("draftReady", "org.acme.test.review", Review.class))
@@ -64,10 +100,14 @@ private Workflow reviewEmitter() {
 
   private void runIt(Workflow listen) {
     Review review = new Review("Torrente", "espectacular", 5);
+
     try (WorkflowApplication app = WorkflowApplication.builder().build()) {
+
       CompletableFuture<WorkflowModel> waiting =
           app.workflowDefinition(listen).instance(Map.of()).start();
+
       app.workflowDefinition(reviewEmitter()).instance(review).start().join();
+
       assertThat(waiting.join().as(Review.class).orElseThrow()).isEqualTo(review);
     }
   }

impl/model/src/main/java/io/serverlessworkflow/impl/model/jackson/JacksonModelCollection.java

@@ -20,9 +20,14 @@
 import io.serverlessworkflow.impl.WorkflowModel;
 import io.serverlessworkflow.impl.WorkflowModelCollection;
 import io.serverlessworkflow.impl.jackson.JsonUtils;
+import java.lang.reflect.Array;
+import java.util.ArrayList;
 import java.util.Collection;
+import java.util.HashSet;
 import java.util.Iterator;
+import java.util.List;
 import java.util.Optional;
+import java.util.Set;
 
 public class JacksonModelCollection implements WorkflowModelCollection {
 
@@ -38,12 +43,35 @@ public class JacksonModelCollection implements WorkflowModelCollection {
 
   @Override
   public <T> Optional<T> as(Class<T> clazz) {
-    if (clazz.equals(Collection.class)) {
-      return Optional.of(clazz.cast(this));
+    if (node == null) return Optional.empty();
+
+    if (clazz.isInstance(node)) return Optional.of(clazz.cast(node));
+
+    if (clazz.isInstance(this)) return Optional.of(clazz.cast(this));
+
+    List<JsonNode> elements = new ArrayList<>(node.size());
+    node.forEach(elements::add);
+
+    if (clazz.isAssignableFrom(List.class)) return Optional.of(clazz.cast(elements));
+    else if (clazz.isAssignableFrom(Set.class))
+      return Optional.of(clazz.cast(new HashSet<>(elements)));
+
+    if (clazz.isArray()) {
+      Class<?> componentType = clazz.getComponentType();
+
+      if (!componentType.isPrimitive()) {
+        Object[] typedArray = (Object[]) Array.newInstance(componentType, 0);
+        return Optional.of(clazz.cast(elements.toArray(typedArray)));
+      }
+
+      Object primitiveArray = Array.newInstance(componentType, elements.size());
+      int i = 0;
+      for (Object item : elements) Array.set(primitiveArray, i++, item);
+
+      return Optional.of(clazz.cast(primitiveArray));
     }
-    return clazz.isAssignableFrom(ArrayNode.class)
-        ? Optional.of(clazz.cast(node))
-        : Optional.empty();
+
+    return Optional.empty();
   }
 
   @Override