Skip to content

PollEnrich

PollEnrich implements the polling variant of the Content Enricher EIP. Unlike Enrich, which sends the current exchange to a producer, PollEnrich actively polls an endpoint with an empty exchange — useful for pulling enrichment data from queues, topics, or any consumer-style endpoint.

Processors in this file

NewPollEnrich

YAML form:

steps:
  - poll-enrich:
      uri: file:///var/in?fileName=enrich.json
      timeout: 5s
      strategy: list
Field Type Default Description
uri string required URI to poll. Resolved via the route's EndpointResolver.
timeout duration 0 (no timeout) Maximum wait. Parsed as a Go duration (500ms, 5s, 1m).
strategy string none (replace body) Built-in aggregation strategy used to merge the polled exchange into the current one. One of list, latest, first, useOriginal, map.

Constructor: NewPollEnrich(uri string, timeout time.Duration, resolver EndpointResolver, strategy engine.AggregationStrategy) *PollEnrich.

How it works:

  1. Applies the configured timeout to a derived context.
  2. If the resolver implements PollingEndpointResolver (Helix's polling-consumer hook), calls PollEndpoint(ctx, uri). For consumer-style endpoints (e.g. sftp://, ftp://) this runs a one-shot poll that downloads a single message; for producer-style endpoints it falls back to a producer call (sends an empty exchange and uses the result).
  3. If the resolver is not a PollingEndpointResolver, falls back directly to that producer call.
  4. If polling returns no exchange within the timeout, returns nil — the current exchange is left untouched.
  5. If Strategy is set, calls Strategy.Aggregate(current, polled) and writes the merged In() back to the current exchange. Otherwise replaces the body with the polled body.

Notes and gotchas:

  • Empty/timed-out polls are not errors. Downstream steps must defend against the original body if your route depends on enrichment.
  • For request-reply enrichment (where the producer needs the current message as context), use enrich instead.
  • PollEnrich differs from PollEnrichProcessor: the former resolves a URI string each call; the latter holds a pre-resolved producer.

Examples

workflows:
  - id: enrich-with-rates
    trigger: { uri: timer://tick?period=10s }
    steps:
      - target: http://localhost:8080/orders
      - poll-enrich:
          uri: redis://rates?op=get&key=fx:latest
          timeout: 2s
          strategy: list
      - target: kafka://orders-with-fx

Source

pkg/processor/poll_enrich.go

See also

  • routing — Enrich for the request-reply variant.
  • aggregate — full list of aggregation strategies.
  • eip_extra — PollEnrichProcessor for pre-resolved producers.