Fix #1189 - Add correlate to the DSL and refactor listen/emit tasks

[ricardozanini]

Many thanks for submitting your Pull Request ❤️!
Closes #1189

What this PR does / why we need it:

Special notes for reviewers:

Additional information (if needed):

[Copilot]

Pull request overview

This PR adds support for event correlation filters in the fluent Java DSL (closing #1189) and refactors the listen/emit DSL APIs to better separate “consumed event filters” from “produced event properties”.

Changes:

• Introduces a new event filter spec base (AbstractEventFilterSpec) with correlate(...) support and refactors listen specs to accept filter consumers directly.
• Refactors emit DSL to use produced()/produced(type) (and adds convenience emit(String cloudEventType) / emit(name, cfg) overloads).
• Updates tests and experimental Func DSL to align naming (produced vs consumed) and builder structure.

Reviewed changes

Copilot reviewed 23 out of 23 changed files in this pull request and generated no comments.

Show a summary per file

File	Description
fluent/spec/src/test/java/io/serverlessworkflow/fluent/spec/dsl/TryCatchDslTest.java	Updates emit-task construction to use the new emit shortcuts.
fluent/spec/src/test/java/io/serverlessworkflow/fluent/spec/dsl/DSLTest.java	Updates listen→emit example to use produced(...) for emission.
fluent/spec/src/test/java/io/serverlessworkflow/fluent/spec/WorkflowBuilderTest.java	Refactors emit-event test to use emit(..., produced()...) and simplifies type imports.
fluent/spec/src/main/java/io/serverlessworkflow/fluent/spec/dsl/ListenSpec.java	Updates listen spec generics and wiring for the new filter-consumer strategy.
fluent/spec/src/main/java/io/serverlessworkflow/fluent/spec/dsl/ExprEventPropertiesSpec.java	Splits “event properties” steps from filter steps (jsonData now adds property steps).
fluent/spec/src/main/java/io/serverlessworkflow/fluent/spec/dsl/EventSpec.java	Removes the old EventSpec class (previously used for event properties config).
fluent/spec/src/main/java/io/serverlessworkflow/fluent/spec/dsl/EventPropertiesSpec.java	New shared base for setting CloudEvent properties (type/source/contentType/id/time).
fluent/spec/src/main/java/io/serverlessworkflow/fluent/spec/dsl/EventFilterSpec.java	Becomes a concrete filter spec built on AbstractEventFilterSpec.
fluent/spec/src/main/java/io/serverlessworkflow/fluent/spec/dsl/EmitSpec.java	Refactors emit spec to reuse ExprEventPropertiesSpec and property steps.
fluent/spec/src/main/java/io/serverlessworkflow/fluent/spec/dsl/DSL.java	Adds produced() helpers and new emit(...) overloads; updates event() to return EventFilterSpec.
fluent/spec/src/main/java/io/serverlessworkflow/fluent/spec/dsl/BaseListenSpec.java	Refactors listen strategy API to accept Consumer<EventFilterBuilder> filters directly.
fluent/spec/src/main/java/io/serverlessworkflow/fluent/spec/dsl/AbstractEventFilterSpec.java	New base that supports correlate(...) and applies properties via with(...).
fluent/spec/src/main/java/io/serverlessworkflow/fluent/spec/configurers/EventPropertiesConfigurer.java	Renames event properties configurer interface (was EventConfigurer).
experimental/fluent/func/src/test/java/io/serverlessworkflow/fluent/func/FuncDSLUniqueIdTest.java	Tightens imports and removes an unnecessary throws from a test method.
experimental/fluent/func/src/test/java/io/serverlessworkflow/fluent/func/FuncDSLTest.java	Updates Func DSL tests to use produced(...) and adjusted helpers.
experimental/fluent/func/src/main/java/io/serverlessworkflow/fluent/func/dsl/internal/CommonFuncOps.java	Renames internal helper methods from event→emit and removes predicate configurer helper.
experimental/fluent/func/src/main/java/io/serverlessworkflow/fluent/func/dsl/FuncEventFilterSpec.java	Refactors to AbstractEventFilterSpec and introduces dataMatches(Predicate) for listen-side filtering.
experimental/fluent/func/src/main/java/io/serverlessworkflow/fluent/func/dsl/FuncEmitSpec.java	Separates emission properties into EventPropertiesSpec and keeps function-based data helpers.
experimental/fluent/func/src/main/java/io/serverlessworkflow/fluent/func/dsl/FuncDSL.java	Introduces produced()/consumed() naming, updates toAll/toAny/toOne and emit helper plumbing.
experimental/fluent/func/src/main/java/io/serverlessworkflow/fluent/func/dsl/BaseFuncListenSpec.java	Adapts func listen base to the new BaseListenSpec signature (no withApplier).
experimental/fluent/func/src/main/java/io/serverlessworkflow/fluent/func/configurers/FuncEventPropertiesConfigurer.java	Renames func event properties configurer interface (was FuncEventConfigurer).
experimental/fluent/func/src/main/java/io/serverlessworkflow/fluent/func/FuncEventPropertiesBuilder.java	Adds predicate-based data(...) to support listen-side data matching.
experimental/fluent/func/src/main/java/io/serverlessworkflow/fluent/func/FuncEventFilterBuilder.java	Updates event properties builder type used by the filter builder.

Comments suppressed due to low confidence (6)

fluent/spec/src/main/java/io/serverlessworkflow/fluent/spec/dsl/DSL.java:650

• The Javadoc for the new emit/produced helpers includes stray trailing * characters inside the comment text (e.g., "...configurer. *" / "...emitted. *"). These render incorrectly and may fail strict Javadoc checks; remove the extra * so the sentences read normally.

  /**
   * Adds an {@code emit} task to the workflow's task sequence using a custom configurer. *
   *
   * <p>This method is typically used in conjunction with {@link #produced()} to fluently define the
   * properties of the event being emitted. *
   *

fluent/spec/src/main/java/io/serverlessworkflow/fluent/spec/dsl/DSL.java:211

• The Javadoc for event() still describes it as an event emission builder and directs users to pass it to emit(...), but event() now returns EventFilterSpec (a Consumer<EventFilterBuilder>) and cannot be used with emit(Consumer<EmitTaskBuilder>). Update this Javadoc to describe it as a listen/consumed event filter builder (including correlation), and point emit use-cases to produced() instead.

  /**
   * Start building an event emission specification.
   *
   * <p>Use methods on {@link EventFilterSpec} to define event type and payload, and pass it to
   * {@link #emit(Consumer)}.
   *
   * @return a new {@link EventFilterSpec}
   */
  public static EventFilterSpec event() {
    return new EventFilterSpec();

fluent/spec/src/main/java/io/serverlessworkflow/fluent/spec/dsl/AbstractEventFilterSpec.java:48

• No tests were added to validate the new correlate(...) DSL behavior. Given this PR’s goal (issue #1189), it would be good to add a unit test that builds a listen task with a correlated EventFilterSpec and asserts the resulting workflow model contains the expected EventFilter.correlate entries.

  public SELF correlate(String key, Consumer<ListenTaskBuilder.CorrelatePropertyBuilder> c) {
    filterSteps.add(f -> f.correlate(key, c));
    return self();
  }

fluent/spec/src/test/java/io/serverlessworkflow/fluent/spec/dsl/TryCatchDslTest.java:124

• Extra parentheses in emit(("org.acme.recover")) are redundant and reduce readability. Consider simplifying to emit("org.acme.recover").

                            .tasks(emit(("org.acme.recover")))

fluent/spec/src/test/java/io/serverlessworkflow/fluent/spec/dsl/TryCatchDslTest.java:177

• Extra parentheses in emit(("org.acme.retrying")) are redundant and reduce readability. Consider simplifying to emit("org.acme.retrying").

                            .tasks(emit(("org.acme.retrying")))

experimental/fluent/func/src/test/java/io/serverlessworkflow/fluent/func/FuncDSLTest.java:142

• The @DisplayName says emit(consumed(type, fn))..., but the test now uses emit(produced(...)). Update the display name to match the API being exercised to avoid confusion when reading test reports.

  @DisplayName("emit(consumed(type, fn)).when(...) -> still an EmitTask and builds")
  void emit_step_when_compiles_and_builds() {
    Workflow wf =
        FuncWorkflowBuilder.workflow("step-emit-when")
            .tasks(emit(produced("org.acme.sig").bytesDataUtf8()).when((Object ctx) -> true))

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.