[Fix #1211] Support filters againts workflow data in listen task

fjtirado · fjtirado · commit 3f83e521f2f9 · 2026-03-10T14:50:05.000+01:00
Signed-off-by: fjtirado &lt;ftirados@redhat.com&gt;
diff --git a/impl/core/src/main/java/io/serverlessworkflow/impl/events/AbstractTypeConsumer.java b/impl/core/src/main/java/io/serverlessworkflow/impl/events/AbstractTypeConsumer.java
@@ -18,7 +18,9 @@
 import io.cloudevents.CloudEvent;
 import io.serverlessworkflow.api.types.EventFilter;
 import io.serverlessworkflow.api.types.EventProperties;
+import io.serverlessworkflow.impl.TaskContext;
 import io.serverlessworkflow.impl.WorkflowApplication;
+import io.serverlessworkflow.impl.WorkflowContext;
 import java.util.AbstractCollection;
 import java.util.Collection;
 import java.util.Iterator;
@@ -67,7 +69,7 @@ private static class CloudEventConsumer extends AbstractCollection<TypeEventRegi
     public void accept(CloudEvent ce) {
       logger.debug("Received cloud event {}", ce);
       for (TypeEventRegistration registration : registrations) {
-        if (registration.predicate().test(ce)) {
+        if (registration.predicate().test(ce, registration.workflow(), registration.task())) {
           registration.consumer().accept(ce);
         }
       }
@@ -94,14 +96,18 @@ public int size() {
     }
   }
 
+  @Override
   public TypeEventRegistration register(
-      TypeEventRegistrationBuilder builder, Consumer<CloudEvent> ce) {
+      TypeEventRegistrationBuilder builder,
+      Consumer<CloudEvent> ce,
+      WorkflowContext workflow,
+      TaskContext task) {
     if (builder.type() == null) {
       registerToAll(ce);
-      return new TypeEventRegistration(null, ce, null);
+      return new TypeEventRegistration(null, ce, null, workflow, task);
     } else {
       TypeEventRegistration registration =
-          new TypeEventRegistration(builder.type(), ce, builder.cePredicate());
+          new TypeEventRegistration(builder.type(), ce, builder.cePredicate(), workflow, task);
       registrations
           .computeIfAbsent(
               registration.type(),
diff --git a/impl/core/src/main/java/io/serverlessworkflow/impl/events/CloudEventAttrPredicate.java b/impl/core/src/main/java/io/serverlessworkflow/impl/events/CloudEventAttrPredicate.java
@@ -15,7 +15,10 @@
  */
 package io.serverlessworkflow.impl.events;
 
+import io.serverlessworkflow.impl.TaskContext;
+import io.serverlessworkflow.impl.WorkflowContext;
+
 @FunctionalInterface
 public interface CloudEventAttrPredicate<T> {
-  boolean test(T value);
+  boolean test(T value, WorkflowContext workflow, TaskContext task);
 }
diff --git a/impl/core/src/main/java/io/serverlessworkflow/impl/events/CloudEventPredicate.java b/impl/core/src/main/java/io/serverlessworkflow/impl/events/CloudEventPredicate.java
@@ -16,7 +16,9 @@
 package io.serverlessworkflow.impl.events;
 
 import io.cloudevents.CloudEvent;
+import io.serverlessworkflow.impl.TaskContext;
+import io.serverlessworkflow.impl.WorkflowContext;
 
 public interface CloudEventPredicate {
-  boolean test(CloudEvent event);
+  boolean test(CloudEvent event, WorkflowContext workflow, TaskContext task);
 }
diff --git a/impl/core/src/main/java/io/serverlessworkflow/impl/events/DefaultCloudEventPredicate.java b/impl/core/src/main/java/io/serverlessworkflow/impl/events/DefaultCloudEventPredicate.java
@@ -23,7 +23,9 @@
 import io.serverlessworkflow.api.types.EventSource;
 import io.serverlessworkflow.api.types.EventTime;
 import io.serverlessworkflow.api.types.UriTemplate;
+import io.serverlessworkflow.impl.TaskContext;
 import io.serverlessworkflow.impl.WorkflowApplication;
+import io.serverlessworkflow.impl.WorkflowContext;
 import io.serverlessworkflow.impl.WorkflowModelFactory;
 import io.serverlessworkflow.impl.WorkflowPredicate;
 import io.serverlessworkflow.impl.expressions.ExpressionDescriptor;
@@ -45,7 +47,7 @@ public class DefaultCloudEventPredicate implements CloudEventPredicate {
   private final CloudEventAttrPredicate<Map<String, Object>> additionalFilter;
 
   private static final <T> CloudEventAttrPredicate<T> isTrue() {
-    return x -> true;
+    return (x, w, t) -> true;
   }
 
   public DefaultCloudEventPredicate(EventProperties properties, WorkflowApplication app) {
@@ -72,12 +74,12 @@ private CloudEventAttrPredicate<Map<String, Object>> additionalFilter(
 
   private CloudEventAttrPredicate<CloudEventData> fromCloudEvent(
       WorkflowModelFactory workflowModelFactory, WorkflowPredicate filter) {
-    return d -> filter.test(null, null, workflowModelFactory.from(d));
+    return (d, w, t) -> filter.test(w, t, workflowModelFactory.from(d));
   }
 
   private CloudEventAttrPredicate<Map<String, Object>> fromMap(
       WorkflowModelFactory workflowModelFactory, WorkflowPredicate filter) {
-    return d -> filter.test(null, null, workflowModelFactory.from(d));
+    return (d, w, t) -> filter.test(w, t, workflowModelFactory.from(d));
   }
 
   private CloudEventAttrPredicate<CloudEventData> dataFilter(
@@ -98,9 +100,9 @@ private CloudEventAttrPredicate<OffsetDateTime> offsetTimeFilter(
         final WorkflowPredicate expr =
             app.expressionFactory()
                 .buildPredicate(ExpressionDescriptor.from(time.getRuntimeExpression()));
-        return s -> evalExpr(app.modelFactory(), expr, s);
+        return (s, w, t) -> evalExpr(app.modelFactory(), expr, s, w, t);
       } else if (time.getLiteralTime() != null) {
-        return s -> Objects.equals(s, CloudEventUtils.toOffset(time.getLiteralTime()));
+        return (s, w, t) -> Objects.equals(s, CloudEventUtils.toOffset(time.getLiteralTime()));
       }
     }
     return isTrue();
@@ -113,7 +115,7 @@ private CloudEventAttrPredicate<URI> dataSchemaFilter(
         final WorkflowPredicate expr =
             app.expressionFactory()
                 .buildPredicate(ExpressionDescriptor.from(dataSchema.getExpressionDataSchema()));
-        return s -> evalExpr(app.modelFactory(), expr, toString(s));
+        return (s, w, t) -> evalExpr(app.modelFactory(), expr, toString(s), w, t);
       } else if (dataSchema.getLiteralDataSchema() != null) {
         return templateFilter(dataSchema.getLiteralDataSchema());
       }
@@ -122,7 +124,7 @@ private CloudEventAttrPredicate<URI> dataSchemaFilter(
   }
 
   private CloudEventAttrPredicate<String> stringFilter(String str) {
-    return str == null ? isTrue() : x -> x.equals(str);
+    return str == null ? isTrue() : (x, w, t) -> str.equals(x);
   }
 
   private CloudEventAttrPredicate<URI> sourceFilter(EventSource source, WorkflowApplication app) {
@@ -131,7 +133,7 @@ private CloudEventAttrPredicate<URI> sourceFilter(EventSource source, WorkflowAp
         final WorkflowPredicate expr =
             app.expressionFactory()
                 .buildPredicate(ExpressionDescriptor.from(source.getRuntimeExpression()));
-        return s -> evalExpr(app.modelFactory(), expr, toString(s));
+        return (s, w, t) -> evalExpr(app.modelFactory(), expr, toString(s), w, t);
       } else if (source.getUriTemplate() != null) {
         return templateFilter(source.getUriTemplate());
       }
@@ -141,7 +143,7 @@ private CloudEventAttrPredicate<URI> sourceFilter(EventSource source, WorkflowAp
 
   private CloudEventAttrPredicate<URI> templateFilter(UriTemplate template) {
     if (template.getLiteralUri() != null) {
-      return u -> Objects.equals(u, template.getLiteralUri());
+      return (u, w, t) -> Objects.equals(u, template.getLiteralUri());
     }
     throw new UnsupportedOperationException("Template not supported here yet");
   }
@@ -151,25 +153,33 @@ private <T> String toString(T uri) {
   }
 
   private boolean evalExpr(
-      WorkflowModelFactory modelFactory, WorkflowPredicate expr, String value) {
-    return expr.test(null, null, modelFactory.from(value));
+      WorkflowModelFactory modelFactory,
+      WorkflowPredicate expr,
+      String value,
+      WorkflowContext workflow,
+      TaskContext task) {
+    return expr.test(workflow, task, modelFactory.from(value));
   }
 
   private boolean evalExpr(
-      WorkflowModelFactory modelFactory, WorkflowPredicate expr, OffsetDateTime value) {
-    return expr.test(null, null, modelFactory.from(value));
+      WorkflowModelFactory modelFactory,
+      WorkflowPredicate expr,
+      OffsetDateTime value,
+      WorkflowContext workflow,
+      TaskContext task) {
+    return expr.test(workflow, task, modelFactory.from(value));
   }
 
   @Override
-  public boolean test(CloudEvent event) {
-    return idFilter.test(event.getId())
-        && sourceFilter.test(event.getSource())
-        && subjectFilter.test(event.getSubject())
-        && contentTypeFilter.test(event.getDataContentType())
-        && typeFilter.test(event.getType())
-        && dataSchemaFilter.test(event.getDataSchema())
-        && timeFilter.test(event.getTime())
-        && dataFilter.test(event.getData())
-        && additionalFilter.test(CloudEventUtils.extensions(event));
+  public boolean test(CloudEvent event, WorkflowContext workflow, TaskContext task) {
+    return idFilter.test(event.getId(), workflow, task)
+        && sourceFilter.test(event.getSource(), workflow, task)
+        && subjectFilter.test(event.getSubject(), workflow, task)
+        && contentTypeFilter.test(event.getDataContentType(), workflow, task)
+        && typeFilter.test(event.getType(), workflow, task)
+        && dataSchemaFilter.test(event.getDataSchema(), workflow, task)
+        && timeFilter.test(event.getTime(), workflow, task)
+        && dataFilter.test(event.getData(), workflow, task)
+        && additionalFilter.test(CloudEventUtils.extensions(event), workflow, task);
   }
 }
diff --git a/impl/core/src/main/java/io/serverlessworkflow/impl/events/EventConsumer.java b/impl/core/src/main/java/io/serverlessworkflow/impl/events/EventConsumer.java
@@ -18,7 +18,9 @@
 import io.cloudevents.CloudEvent;
 import io.serverlessworkflow.api.types.EventFilter;
 import io.serverlessworkflow.impl.ServicePriority;
+import io.serverlessworkflow.impl.TaskContext;
 import io.serverlessworkflow.impl.WorkflowApplication;
+import io.serverlessworkflow.impl.WorkflowContext;
 import java.util.Collection;
 import java.util.function.Consumer;
 
@@ -29,7 +31,12 @@ public interface EventConsumer<T extends EventRegistration, V extends EventRegis
 
   Collection<V> listenToAll(WorkflowApplication workflowApplication);
 
-  T register(V builder, Consumer<CloudEvent> consumer);
+  default T register(V builder, Consumer<CloudEvent> consumer) {
+    return register(builder, consumer, null, null);
+  }
+
+  T register(
+      V builder, Consumer<CloudEvent> consumer, WorkflowContext workflow, TaskContext context);
 
   void unregister(T register);
 }
diff --git a/impl/core/src/main/java/io/serverlessworkflow/impl/events/EventRegistration.java b/impl/core/src/main/java/io/serverlessworkflow/impl/events/EventRegistration.java
@@ -15,4 +15,12 @@
  */
 package io.serverlessworkflow.impl.events;
 
-public interface EventRegistration {}
+import io.serverlessworkflow.impl.TaskContext;
+import io.serverlessworkflow.impl.WorkflowContext;
+
+public interface EventRegistration {
+
+  WorkflowContext workflow();
+
+  TaskContext task();
+}
diff --git a/impl/core/src/main/java/io/serverlessworkflow/impl/events/EventRegistrationInfo.java b/impl/core/src/main/java/io/serverlessworkflow/impl/events/EventRegistrationInfo.java
@@ -16,6 +16,8 @@
 package io.serverlessworkflow.impl.events;
 
 import io.cloudevents.CloudEvent;
+import io.serverlessworkflow.impl.TaskContext;
+import io.serverlessworkflow.impl.WorkflowContext;
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.concurrent.CompletableFuture;
@@ -27,11 +29,13 @@ public record EventRegistrationInfo(
   public static final <T> EventRegistrationInfo build(
       EventRegistrationBuilderCollection builderInfo,
       BiConsumer<CloudEvent, CompletableFuture<T>> consumer,
-      EventConsumer eventConsumer) {
-    Collection<EventRegistration> registrations = new ArrayList();
+      EventConsumer eventConsumer,
+      WorkflowContext workflow,
+      TaskContext task) {
+    Collection<EventRegistration> registrations = new ArrayList<>();
     CompletableFuture<T>[] futures =
         builderInfo.registrations().stream()
-            .map(reg -> toCompletable(reg, registrations, consumer, eventConsumer))
+            .map(reg -> toCompletable(reg, registrations, consumer, eventConsumer, workflow, task))
             .toArray(size -> new CompletableFuture[size]);
     return new EventRegistrationInfo(
         builderInfo.isAnd() ? CompletableFuture.allOf(futures) : CompletableFuture.anyOf(futures),
@@ -42,10 +46,13 @@ private static final <T> CompletableFuture<T> toCompletable(
       EventRegistrationBuilder regBuilder,
       Collection<EventRegistration> registrations,
       BiConsumer<CloudEvent, CompletableFuture<T>> ceConsumer,
-      EventConsumer eventConsumer) {
+      EventConsumer eventConsumer,
+      WorkflowContext workflow,
+      TaskContext task) {
     final CompletableFuture<T> future = new CompletableFuture<>();
     registrations.add(
-        eventConsumer.register(regBuilder, ce -> ceConsumer.accept((CloudEvent) ce, future)));
+        eventConsumer.register(
+            regBuilder, ce -> ceConsumer.accept((CloudEvent) ce, future), workflow, task));
     return future;
   }
 }
diff --git a/impl/core/src/main/java/io/serverlessworkflow/impl/events/TypeEventRegistration.java b/impl/core/src/main/java/io/serverlessworkflow/impl/events/TypeEventRegistration.java
@@ -16,8 +16,14 @@
 package io.serverlessworkflow.impl.events;
 
 import io.cloudevents.CloudEvent;
+import io.serverlessworkflow.impl.TaskContext;
+import io.serverlessworkflow.impl.WorkflowContext;
 import java.util.function.Consumer;
 
 public record TypeEventRegistration(
-    String type, Consumer<CloudEvent> consumer, CloudEventPredicate predicate)
+    String type,
+    Consumer<CloudEvent> consumer,
+    CloudEventPredicate predicate,
+    WorkflowContext workflow,
+    TaskContext task)
     implements EventRegistration {}
diff --git a/impl/core/src/main/java/io/serverlessworkflow/impl/executors/ListenExecutor.java b/impl/core/src/main/java/io/serverlessworkflow/impl/executors/ListenExecutor.java
@@ -121,12 +121,14 @@ public OrListenExecutor(ListenExecutorBuilder builder) {
 
     @Override
     protected <T> EventRegistrationInfo buildInfo(
-        BiConsumer<CloudEvent, CompletableFuture<T>> consumer) {
-      EventRegistrationInfo info = super.buildInfo(consumer);
+        BiConsumer<CloudEvent, CompletableFuture<T>> consumer,
+        WorkflowContext workflow,
+        TaskContext task) {
+      EventRegistrationInfo info = super.buildInfo(consumer, workflow, task);
       if (untilRegBuilders != null) {
         EventRegistrationInfo untilInfo =
             EventRegistrationInfo.build(
-                untilRegBuilders, (ce, f) -> f.complete(null), eventConsumer);
+                untilRegBuilders, (ce, f) -> f.complete(null), eventConsumer, workflow, task);
         untilInfo
             .completableFuture()
             .thenAccept(
@@ -171,7 +173,9 @@ protected CompletableFuture<WorkflowModel> internalExecute(
         buildInfo(
             (BiConsumer<CloudEvent, CompletableFuture<WorkflowModel>>)
                 ((ce, future) ->
-                    processCe(converter.apply(ce), output, workflow, taskContext, future)));
+                    processCe(converter.apply(ce), output, workflow, taskContext, future)),
+            workflow,
+            taskContext);
     return info.completableFuture()
         .thenApply(
             v -> {
@@ -181,9 +185,11 @@ protected CompletableFuture<WorkflowModel> internalExecute(
   }
 
   protected <T> EventRegistrationInfo buildInfo(
-      BiConsumer<CloudEvent, CompletableFuture<T>> consumer) {
+      BiConsumer<CloudEvent, CompletableFuture<T>> consumer,
+      WorkflowContext workflow,
+      TaskContext task) {
     return EventRegistrationInfo.build(
-        builderRegistrationInfo.registrations(), consumer, eventConsumer);
+        builderRegistrationInfo.registrations(), consumer, eventConsumer, workflow, task);
   }
 
   private void processCe(
diff --git a/impl/test/src/test/java/io/serverlessworkflow/impl/test/EventDefinitionTest.java b/impl/test/src/test/java/io/serverlessworkflow/impl/test/EventDefinitionTest.java
@@ -57,7 +57,7 @@ void testEventListened(String listen, String emit, JsonNode expectedResult, Obje
         appl.workflowDefinition(WorkflowReader.readWorkflowFromClasspath(listen));
     WorkflowDefinition emitDefinition =
         appl.workflowDefinition(WorkflowReader.readWorkflowFromClasspath(emit));
-    WorkflowInstance waitingInstance = listenDefinition.instance(Map.of());
+    WorkflowInstance waitingInstance = listenDefinition.instance(Map.of("threshold", 38));
     CompletableFuture<WorkflowModel> future = waitingInstance.start();
     assertThat(waitingInstance.status()).isEqualTo(WorkflowStatus.WAITING);
     emitDefinition.instance(emitInput).start().join();
diff --git a/impl/test/src/test/resources/workflows-samples/listen-to-any-filter.yaml b/impl/test/src/test/resources/workflows-samples/listen-to-any-filter.yaml
@@ -10,7 +10,7 @@ do:
           any:
           - with:
               type: com.fake-hospital.vitals.measurements.temperature
-              data: ${ .temperature > 38 }
+              data: ${ .temperature > $input.threshold }
           - with:
               type: com.fake-hospital.vitals.measurements.bpm
               data: ${ .bpm < 60 or .bpm > 100 }

Original file line number	Diff line number	Diff line change
`@@ -16,7 +16,9 @@`
`16`	`16`	`package io.serverlessworkflow.impl.events;`
`17`	`17`
`18`	`18`	`import io.cloudevents.CloudEvent;`
	`19`	`+import io.serverlessworkflow.impl.TaskContext;`
	`20`	`+import io.serverlessworkflow.impl.WorkflowContext;`
`19`	`21`
`20`	`22`	`public interface CloudEventPredicate {`
`21`		`- boolean test(CloudEvent event);`
	`23`	`+ boolean test(CloudEvent event, WorkflowContext workflow, TaskContext task);`
`22`	`24`	`}`