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

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: ","

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

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

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

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¶

• multicast — RecipientList is a typed front for it.
• poll_enrich — Enrich for polling consumers.
• idempotent_redis — V2 idempotent consumer with Redis backing and context.
• aggregate — strategies used by Enrich.