Skip to content

Filter

The Filter processor implements the Message Filter EIP. It evaluates a predicate against the incoming exchange; if the predicate matches, the wrapped processor runs. If the predicate is false, the filter is a no-op and the exchange passes through unchanged.

Processors in this file

NewFilter

YAML form:

steps:
  - filter:
      starlang: <predicate expression>
      steps:
        - log: { message: "matched" }
Field Type Default Description
starlang string required Starlang predicate. Truthy values cause the nested steps to run; falsy values short-circuit the filter.
steps list required Sub-processors to execute when the predicate matches.

Constructor: NewFilter(predicate engine.Predicate, processor engine.Processor) *Filter.

How it works:

  1. Evaluates Predicate.Matches(ctx, ex).
  2. If true, delegates to the inner Processor.
  3. If false, returns nil immediately — the exchange continues down the parent pipeline unchanged.

Notes and gotchas:

  • The filter does not drop the exchange from the parent pipeline when the predicate is false. It simply skips its own nested steps. Use stop (see eip_advanced) to terminate routing entirely.
  • A predicate error (Starlang compile or runtime failure) propagates as a processing error.
  • The default expression engine is Starlang; see the Starlang guide for predicate syntax (e.g. body().status == "ok").

Examples

workflows:
  - id: orders-filter
    trigger:
      uri: direct://orders
    steps:
      - filter:
          starlang: body().total > 100
          steps:
            - log: { message: "large order" }
            - target: kafka://high-value-orders

Source

pkg/processor/filter.go

See also