[Fix #1280] Support both casting to JsonNode and Collection

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

Author: fjtirado

experimental/pom.xml

@@ -49,5 +49,6 @@
     <module>lambda-fluent</module>
     <module>fluent</module>
     <module>model</module>
+    <module>test</module>
   </modules>
 </project>

experimental/test/pom.xml

@@ -0,0 +1,43 @@
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
+  <modelVersion>4.0.0</modelVersion>
+  <parent>
+    <groupId>io.serverlessworkflow</groupId>
+    <artifactId>serverlessworkflow-experimental</artifactId>
+    <version>8.0.0-SNAPSHOT</version>
+  </parent>
+  <artifactId>serverlessworkflow-experimental-test</artifactId>
+  <name>Serverless Workflow :: Experimental :: Test</name>
+  <dependencies>
+      <dependency>
+            <groupId>org.junit.jupiter</groupId>
+            <artifactId>junit-jupiter-engine</artifactId>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.mockito</groupId>
+            <artifactId>mockito-core</artifactId>
+             <scope>test</scope>    
+        </dependency>
+         <dependency>
+            <groupId>org.assertj</groupId>
+            <artifactId>assertj-core</artifactId>
+              <scope>test</scope>
+        </dependency>
+        <dependency>
+          <groupId>io.serverlessworkflow</groupId>
+          <artifactId>serverlessworkflow-experimental-fluent-func</artifactId>
+          <scope>test</scope>
+        </dependency>
+          <dependency>
+          <groupId>io.serverlessworkflow</groupId>
+           <artifactId>serverlessworkflow-experimental-lambda</artifactId>
+          <scope>test</scope>
+        </dependency>
+      <dependency>
+          <groupId>io.serverlessworkflow</groupId>
+          <artifactId>serverlessworkflow-impl-model</artifactId>
+          <version>${project.version}</version>
+          <scope>test</scope>
+        </dependency>
+  </dependencies>
+</project>

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

@@ -0,0 +1,74 @@
+/*
+ * 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.fluent.test;
+
+import static io.serverlessworkflow.fluent.func.dsl.FuncDSL.emitJson;
+import static io.serverlessworkflow.fluent.func.dsl.FuncDSL.listen;
+import static io.serverlessworkflow.fluent.func.dsl.FuncDSL.toOne;
+import static org.assertj.core.api.Assertions.assertThat;
+
+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.Map;
+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.
+ */
+class FuncEventFilterTest {
+
+  @Test
+  void testListenToOneCollection() {
+    runIt(
+        FuncWorkflowBuilder.workflow("listenToOneReviewCol")
+            .tasks(
+                listen("waitReview", toOne("org.acme.test.review"))
+                    .outputAs((Collection<?> node) -> node.iterator().next()))
+            .build());
+  }
+
+  @Test
+  void testListenToOneNode() {
+    runIt(
+        FuncWorkflowBuilder.workflow("listenToOneReviewNode")
+            .tasks(
+                listen("waitReview", toOne("org.acme.test.review"))
+                    .outputAs((ArrayNode node) -> node.get(0)))
+            .build());
+  }
+
+  private Workflow reviewEmitter() {
+    return FuncWorkflowBuilder.workflow("emitReview")
+        .tasks(emitJson("draftReady", "org.acme.test.review", Review.class))
+        .build();
+  }
+
+  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);
+    }
+  }
+}

experimental/test/src/test/java/io/serverlessworkflow/fluent/test/Review.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.fluent.test;
+
+record Review(String title, String description, int rating) {}

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

@@ -38,6 +38,9 @@ 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));
+    }
     return clazz.isAssignableFrom(ArrayNode.class)
         ? Optional.of(clazz.cast(node))
         : Optional.empty();