Skip to content

Routing processors

pkg/processor/routing.go collects the URI- and key-driven routing EIPs: Recipient List, Routing Slip, Dynamic Router, Content Enricher, and Idempotent Consumer. Each computes one or more URIs from message data and delegates to the route's EndpointResolver.

Processors in this file

EndpointResolver interface

The seam between these processors and the rest of the engine:

type EndpointResolver interface {
    ResolveProcessor(uri string) (engine.Processor, error)
}

Implemented by engine.HelixContext; injected into every processor here.

NewRecipientList

Evaluates an expression that returns a list of URIs (or a delimited string) and broadcasts the exchange to all of them via the standard Multicast machinery.

steps:
  - recipient-list:
      starlang: split(header("X-Targets"), ",")
      delimiter: ","
      parallel: true
Field Type Default Description
starlang string required Expression returning URI(s). Result is normalized via toStringSlice(result, delimiter).
delimiter string "," Used when the expression result is a single string to split into multiple URIs.
parallel bool false Forward to multicast in parallel.

Constructor: NewRecipientList(expr engine.Expression, delimiter string, resolver EndpointResolver) *RecipientList.

Internally, RecipientList constructs a Multicast{Processors: ..., Parallel: r.Parallel} and delegates — so all behavior of multicast applies (per-target copy, indexed properties, etc.). It does not support an AggregationStrategy directly; if you need one, use Multicast explicitly.

NewRoutingSlip

Sends the exchange through a sequence of URIs computed from an expression. Each URI receives the (already-mutated) exchange from the previous step — order matters.

steps:
  - routing-slip:
      starlang: header("X-Slip")  # e.g. "validate,enrich,publish"
      delimiter: ","
Field Type Default Description
starlang string required Expression returning the slip.
delimiter string "," Splitter when the result is a single string.

Constructor: NewRoutingSlip(expr engine.Expression, delimiter string, resolver EndpointResolver) *RoutingSlip.

Errors short-circuit the slip — a failing step stops the rest. Use do-try around it for resilience.

NewDynamicRouter

Iteratively asks an expression "where next?" and sends the exchange there, repeating until the expression returns an empty string. Useful for state-machine-style routing where the next hop depends on the result of the current hop.

steps:
  - dynamic-router:
      starlang: nextStep(body())   # returns "" to terminate
Field Type Default Description
starlang string required Expression returning the next URI. Empty string terminates.
MaxIterations int (Go-only) 100 Safety cap. The processor errors after this many hops.

Constructor: NewDynamicRouter(expr engine.Expression, resolver EndpointResolver) *DynamicRouter.

On each hop, sets HelixDynamicRouterSlipIndex so the expression can branch on iteration count.

NewEnrich

Sends a copy of the current exchange to a single URI (request-reply pattern). The response is merged back into the original via an AggregationStrategy, or replaces the body if no strategy is given. Compare to PollEnrich which polls an endpoint with an empty exchange.

steps:
  - enrich:
      uri: http://crm/customer
      strategy: list
Field Type Default Description
uri string required Target URI.
strategy string none Built-in aggregation strategy. If unset, the response replaces the body.

Constructor: NewEnrich(uri string, resolver EndpointResolver, strategy engine.AggregationStrategy) *Enrich.

NewIdempotentConsumer (V1)

Deduplicates messages by expression-derived key. The V1 version uses the simpler in-memory IdempotentRepository interface; see idempotent_redis for the V2 distributed variant.

steps:
  - idempotent-consumer:
      starlang: header("X-Message-Id")
      repository: memory
      steps:
        - target: postgres://orders
Field Type Default Description
starlang string required Expression yielding the dedup key.
repository string memory memory (V1, this file) or redis (V2, idempotent_redis).
steps list required Inner pipeline for new (non-duplicate) keys.

Constructor: NewIdempotentConsumer(expr engine.Expression, repo IdempotentRepository, processor engine.Processor) *IdempotentConsumer.

On a duplicate, sets HelixDuplicateMessage = true and returns nil — the message is silently dropped.

IdempotentRepository interface

V1 contract:

type IdempotentRepository interface {
    AddIfAbsent(key string) bool   // atomic check-and-set
    Contains(key string) bool
    Add(key string)
    Remove(key string)
}

Note that AddIfAbsent is atomic — V1 avoids the V2 race between Contains and Add. The trade-off is no context.Context and no error returns.

NewMemoryIdempotentRepository

In-memory implementation of IdempotentRepository backed by map[string]struct{} + mutex. Unbounded — for TTL/cap support use InMemoryIdempotentRepository.

Pitfalls

  • RecipientList and RoutingSlip both stringify expression results with fmt.Sprintf("%v", item). A []any of mixed types produces stringified URIs — which may or may not be what you want.
  • DynamicRouter checks result.(string). A non-string return (e.g. nil, int) terminates routing — which is convenient for "no next step" but means types must be string-only.
  • Enrich uses the same exchange copy semantics as multicast — mutations to headers in the enrich URI affect a copy, not the original. Use the aggregation strategy to merge.

Examples

Recipient list — broadcast to dynamically-named topics:

workflows:
  - id: tenant-fanout
    trigger: { uri: direct://events }
    steps:
      - set-header:
          name: X-Targets
          starlang: '"kafka://tenant-" ++ header("X-Tenant") ++ ",kafka://global-audit"'
      - recipient-list:
          starlang: header("X-Targets")
          delimiter: ","
          parallel: true

Routing slip — pipeline-as-data:

workflows:
  - id: slip
    trigger: { uri: direct://in }
    steps:
      - routing-slip:
          starlang: '"direct://validate,direct://enrich,direct://publish"'

Source

pkg/processor/routing.go

See also