Fix #1275 - Improve CloudEvent and CloudEventData handling

experimental/fluent/func/src/main/java/io/serverlessworkflow/fluent/func/FuncEventFilterPropertiesBuilder.java

@@ -20,7 +20,11 @@
 import io.serverlessworkflow.api.types.func.ContextPredicate;
 import io.serverlessworkflow.api.types.func.EventDataPredicate;
 import io.serverlessworkflow.api.types.func.FilterPredicate;
+import io.serverlessworkflow.api.types.func.TypedContextPredicate;
+import io.serverlessworkflow.api.types.func.TypedFilterPredicate;
+import io.serverlessworkflow.api.types.func.TypedPredicate;
 import io.serverlessworkflow.fluent.spec.AbstractEventPropertiesBuilder;
+import io.serverlessworkflow.impl.events.DefaultCloudEventPredicate;
 import java.util.function.Predicate;
 
 public class FuncEventFilterPropertiesBuilder
@@ -50,20 +54,23 @@ public FuncEventFilterPropertiesBuilder data(FilterPredicate<CloudEventData> pre
   }
 
   public FuncEventFilterPropertiesBuilder envelope(Predicate<CloudEvent> predicate) {
-    this.eventProperties.setData(
-        new EventDataPredicate().withPredicate(predicate, CloudEvent.class));
+    this.eventProperties.setAdditionalProperty(
+        DefaultCloudEventPredicate.ENVELOPE_PREDICATE,
+        new TypedPredicate<>(predicate, CloudEvent.class));
     return this;
   }
 
   public FuncEventFilterPropertiesBuilder envelope(ContextPredicate<CloudEvent> predicate) {
-    this.eventProperties.setData(
-        new EventDataPredicate().withPredicate(predicate, CloudEvent.class));
+    this.eventProperties.setAdditionalProperty(
+        DefaultCloudEventPredicate.ENVELOPE_PREDICATE,
+        new TypedContextPredicate<>(predicate, CloudEvent.class));
     return this;
   }
 
   public FuncEventFilterPropertiesBuilder envelope(FilterPredicate<CloudEvent> predicate) {
-    this.eventProperties.setData(
-        new EventDataPredicate().withPredicate(predicate, CloudEvent.class));
+    this.eventProperties.setAdditionalProperty(
+        DefaultCloudEventPredicate.ENVELOPE_PREDICATE,
+        new TypedFilterPredicate<>(predicate, CloudEvent.class));
     return this;
   }
 }

experimental/lambda/pom.xml

@@ -21,7 +21,7 @@
             <groupId>io.serverlessworkflow</groupId>
             <artifactId>serverlessworkflow-impl-core</artifactId>
         </dependency>
-         <dependency>
+        <dependency>
             <groupId>io.serverlessworkflow</groupId>
             <artifactId>serverlessworkflow-impl-jq</artifactId>
         </dependency>
@@ -50,6 +50,11 @@
             <artifactId>assertj-core</artifactId>
             <scope>test</scope>
         </dependency>
+        <dependency>
+            <groupId>org.awaitility</groupId>
+            <artifactId>awaitility</artifactId>
+            <scope>test</scope>
+        </dependency>
         <dependency>
             <groupId>ch.qos.logback</groupId>
             <artifactId>logback-classic</artifactId>

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

@@ -0,0 +1,161 @@
+/*
+ * 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.serverless.workflow.impl.executors.func;
+
+import static io.serverlessworkflow.fluent.func.dsl.FuncDSL.consume;
+import static io.serverlessworkflow.fluent.func.dsl.FuncDSL.consumed;
+import static io.serverlessworkflow.fluent.func.dsl.FuncDSL.emitJson;
+import static io.serverlessworkflow.fluent.func.dsl.FuncDSL.function;
+import static io.serverlessworkflow.fluent.func.dsl.FuncDSL.listen;
+import static io.serverlessworkflow.fluent.func.dsl.FuncDSL.switchWhenOrElse;
+import static io.serverlessworkflow.fluent.func.dsl.FuncDSL.to;
+import static org.awaitility.Awaitility.await;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import io.cloudevents.CloudEvent;
+import io.cloudevents.CloudEventData;
+import io.cloudevents.core.builder.CloudEventBuilder;
+import io.serverlessworkflow.api.types.Workflow;
+import io.serverlessworkflow.fluent.func.FuncWorkflowBuilder;
+import io.serverlessworkflow.impl.WorkflowApplication;
+import io.serverlessworkflow.impl.WorkflowDefinition;
+import io.serverlessworkflow.impl.WorkflowInstance;
+import io.serverlessworkflow.impl.WorkflowModel;
+import io.serverlessworkflow.impl.WorkflowStatus;
+import io.serverlessworkflow.impl.events.EventPublisher;
+import java.net.URI;
+import java.nio.charset.StandardCharsets;
+import java.time.Duration;
+import java.util.List;
+import java.util.concurrent.CompletableFuture;
+import org.junit.jupiter.api.Test;
+
+public class EventFilteringTest {
+
+  private static final ObjectMapper MAPPER = new ObjectMapper();
+
+  // --- Mock Service Methods (replacing Quarkus Agents) ---
+  public NewsletterDraft writeDraft(NewsletterRequest req) {
+    return new NewsletterDraft("Draft: " + req.topic(), "Initial body...");
+  }
+
+  public NewsletterDraft editDraft(HumanReview review) {
+    return new NewsletterDraft("Edited Draft", "Fixed based on: " + review.notes());
+  }
+
+  public void sendEmail(NewsletterDraft draft) {
+    // Simulates MailService.send
+  }
+
+  @Test
+  public void testIntelligentNewsletterApprovalPath() throws Exception {
+    try (WorkflowApplication app = WorkflowApplication.builder().build()) {
+
+      Workflow workflow =
+          FuncWorkflowBuilder.workflow("intelligent-newsletter")
+              .tasks(
+                  function("draftAgent", this::writeDraft).exportAsTaskOutput(),
+                  emitJson("draftReady", "org.acme.email.review.required", NewsletterDraft.class),
+                  listen(
+                          "waitHumanReview",
+                          to().one(
+                                  consumed("org.acme.newsletter.review.done")
+                                      .extensionByInstanceId("instanceid")))
+                      .outputAs(
+                          (List<CloudEventData> events) -> {
+                            try {
+                              return MAPPER.readValue(events.get(0).toBytes(), HumanReview.class);
+                            } catch (Exception e) {
+                              throw new RuntimeException("Failed to deserialize HumanReview", e);
+                            }
+                          }),
+                  switchWhenOrElse(
+                      h -> HumanReview.NEEDS_REVISION.equals(h.status()),
+                      "humanEditorAgent",
+                      "sendNewsletter",
+                      HumanReview.class),
+                  function("humanEditorAgent", this::editDraft)
+                      .exportAsTaskOutput()
+                      .then("draftReady"),
+                  consume("sendNewsletter", this::sendEmail)
+                      .inputFrom(
+                          (payload, wfc, tfc) ->
+                              payload instanceof HumanReview ? wfc.context() : payload))
+              .build();
+
+      WorkflowDefinition definition = app.workflowDefinition(workflow);
+      WorkflowInstance instance = definition.instance(new NewsletterRequest("Tech Stocks"));
+      CompletableFuture<WorkflowModel> future = instance.start();
+
+      // Wait for it to hit the listen state
+      await()
+          .atMost(Duration.ofSeconds(5))
+          .until(() -> instance.status() == WorkflowStatus.WAITING);
+
+      EventPublisher publisher = app.eventPublishers().iterator().next();
+
+      // --- THE NEGATIVE TEST: Fire an event with the WRONG instance id ---
+      CloudEvent maliciousEvent =
+          CloudEventBuilder.v1()
+              .withId("event-wrong")
+              .withSource(URI.create("test:/human-editor"))
+              .withType("org.acme.newsletter.review.done")
+              .withExtension("instanceid", "SOME-OTHER-ID-12345") // Does not match instance.id()
+              .withData(
+                  "application/json",
+                  "{\"status\":\"APPROVED\", \"notes\":\"Malicious approval\"}"
+                      .getBytes(StandardCharsets.UTF_8))
+              .build();
+
+      publisher.publish(maliciousEvent).toCompletableFuture().join();
+
+      await()
+          .pollDelay(Duration.ofMillis(250))
+          .atMost(Duration.ofMillis(500))
+          .until(() -> instance.status() == WorkflowStatus.WAITING && !future.isDone());
+
+      // --- THE POSITIVE TEST: Fire the CORRECT event ---
+      CloudEvent humanReviewEvent =
+          CloudEventBuilder.v1()
+              .withId("event-123")
+              .withSource(URI.create("test:/human-editor"))
+              .withType("org.acme.newsletter.review.done")
+              .withExtension("instanceid", instance.id()) // Matches exactly
+              .withData(
+                  "application/json",
+                  "{\"status\":\"APPROVED\", \"notes\":\"Looks good\"}"
+                      .getBytes(StandardCharsets.UTF_8))
+              .build();
+
+      publisher.publish(humanReviewEvent).toCompletableFuture().join();
+
+      // Assert successful completion
+      await()
+          .atMost(Duration.ofSeconds(5))
+          .until(() -> instance.status() == WorkflowStatus.COMPLETED);
+    }
+  }
+
+  // --- Mock Domain Models ---
+  public record NewsletterRequest(String topic) {}
+
+  public record NewsletterDraft(String title, String body) {}
+
+  public record HumanReview(String status, String notes) {
+    public static final String NEEDS_REVISION = "NEEDS_REVISION";
+    public static final String APPROVED = "APPROVED";
+  }
+}

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

@@ -15,11 +15,13 @@
  */
 package io.serverlessworkflow.impl.model.func;
 
+import io.serverlessworkflow.impl.CollectionConversionUtils;
 import io.serverlessworkflow.impl.WorkflowModel;
 import io.serverlessworkflow.impl.WorkflowModelCollection;
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.Iterator;
+import java.util.List;
 import java.util.Optional;
 
 public class JavaModelCollection implements Collection<WorkflowModel>, WorkflowModelCollection {
@@ -78,14 +80,22 @@ public Iterator<WorkflowModel> iterator() {
     return new ModelIterator(object.iterator());
   }
 
+  private List<WorkflowModel> toModelList() {
+    List<WorkflowModel> models = new ArrayList<>(object.size());
+    for (Object obj : object)
+      models.add(obj instanceof WorkflowModel value ? value : nextItem(obj));
+
+    return models;
+  }
+
   @Override
   public Object[] toArray() {
-    throw new UnsupportedOperationException("toArray is not supported yet");
+    return toModelList().toArray();
   }
 
   @Override
   public <T> T[] toArray(T[] a) {
-    throw new UnsupportedOperationException("toArray is not supported yet");
+    return toModelList().toArray(a);
   }
 
   @Override
@@ -139,8 +149,11 @@ public Class<?> objectClass() {
 
   @Override
   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(this)) return Optional.of(clazz.cast(this));
+    if (clazz.isInstance(object)) return Optional.of(clazz.cast(object));
+
+    return CollectionConversionUtils.as(object, clazz);
   }
 }

experimental/pom.xml

@@ -15,7 +15,7 @@
             <artifactId>serverlessworkflow-impl-core</artifactId>
             <version>${project.version}</version>
         </dependency>
-         <dependency>
+        <dependency>
             <groupId>io.serverlessworkflow</groupId>
             <artifactId>serverlessworkflow-impl-jq</artifactId>
             <version>${project.version}</version>

experimental/test/pom.xml

@@ -1,43 +1,54 @@
-<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>
+<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>    
+            <scope>test</scope>
         </dependency>
-         <dependency>
+        <dependency>
             <groupId>org.assertj</groupId>
             <artifactId>assertj-core</artifactId>
-              <scope>test</scope>
+            <scope>test</scope>
         </dependency>
         <dependency>
-          <groupId>io.serverlessworkflow</groupId>
-          <artifactId>serverlessworkflow-experimental-fluent-func</artifactId>
-          <scope>test</scope>
+            <groupId>org.awaitility</groupId>
+            <artifactId>awaitility</artifactId>
+            <scope>test</scope>
         </dependency>
-          <dependency>
-          <groupId>io.serverlessworkflow</groupId>
-           <artifactId>serverlessworkflow-experimental-lambda</artifactId>
-          <scope>test</scope>
+        <dependency>
+            <groupId>ch.qos.logback</groupId>
+            <artifactId>logback-classic</artifactId>
+            <scope>test</scope>
         </dependency>
-      <dependency>
-          <groupId>io.serverlessworkflow</groupId>
-          <artifactId>serverlessworkflow-impl-model</artifactId>
-          <version>${project.version}</version>
-          <scope>test</scope>
+        <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>
+    </dependencies>
 </project>