EIP processors (advanced)¶

pkg/processor/eip_advanced.go adds richer or more configurable variants of the patterns in eip.go: callback-driven enrichment and filtering, claim-check retrieval, data-driven normalizer and message router, stateful process manager, scatter-gather, messaging bridge, channel adapter, detour, structured logging, worker pool, rollback, stop, service call, and a stop-aware pipeline.

Most of these are Go-first APIs; some are exposed via the DSL (step, stop, rollback, threads, etc.).

Processors in this file¶

NewContentEnricher¶

Enriches an exchange by calling a user-supplied Go callback that returns extra data. Compare to EnrichProcessor which delegates to a producer/endpoint.

Constructor: NewContentEnricher(enrichFunc func(ctx, ex) (any, error), strategy engine.AggregationStrategy) *ContentEnricher.

NewContentFilter¶

The inverse of ContentEnricher — strips data from the body via a user callback. Use for PII removal, payload trimming, etc.

Constructor: NewContentFilter(filterFunc func(ctx, body any) (any, error)) *ContentFilter.

NewClaimCheckGet¶

A configurable variant of ClaimCheckOut: retrieve a stored payload by key, optionally remove the store entry, and optionally set the result as a header (HelixClaimCheckData) instead of replacing the body.

Constructor: NewClaimCheckGet(store, keyExpr, remove bool) *ClaimCheckGet. Set Merge directly.

Normalizer¶

Data-driven variant of Normalize — rules can be added dynamically and there's an optional fallback.

n := processor.NewNormalizer(fallbackProc)
n.AddRule(predicateXML, xmlNormalizer)
n.AddRule(predicateJSON, jsonNormalizer)

Sets HelixNormalizerMatched to the index of the matched rule, or -1 if none matched.

Constructor: NewNormalizer(fallback engine.Processor) *Normalizer then AddRule(predicate, processor) for each rule.

MessageRouter¶

Data-driven variant of Choice. Named rules, optional fallback, and an AllMatch mode that runs every matching rule rather than just the first.

r := processor.NewMessageRouter(fallbackProc)
r.AddRule("vip", isVIP, vipPipeline)
r.AddRule("eu", isEU, euPipeline)
r.AllMatch = true   // run every match

Sets HelixRouterRuleName to the matched rule (or "fallback").

ProcessManager¶

Stateful step-by-step routing — each step's transition function decides what runs next.

pm := processor.NewProcessManager("review")
pm.AddStep(&processor.ProcessStep{
    Name: "review",
    Processor: reviewProc,
    Transition: func(ctx, ex) (string, error) {
        if approved(ex) { return "ship", nil }
        return "reject", nil
    },
})
pm.AddStep(&processor.ProcessStep{Name: "ship", Processor: shipProc /* Transition: nil → end */ })
pm.AddStep(&processor.ProcessStep{Name: "reject", Processor: rejectProc})

Sets HelixProcessManagerStep, HelixProcessManagerStepCount, HelixProcessManagerComplete. A nil transition ends the process normally.

ScatterGather¶

Combined Multicast + Aggregate. Sends copies in parallel, aggregates responses with a strategy, and supports a single gather-phase timeout.

Sets HelixScatterGatherSize (recipient count) and HelixScatterGatherIndex on each copy.

Constructor: NewScatterGather(strategy engine.AggregationStrategy, timeout time.Duration, processors ...engine.Processor) *ScatterGather.

MessageBridge¶

Implements the Messaging Bridge EIP: in-processor → transformer → out-processor, with named source/target identifiers for tracing.

Constructor: NewMessageBridge(source, target string, inProc, transformer, outProc engine.Processor) *MessageBridge.

ChannelAdapter¶

Adapts a non-messaging data source (file pull, REST poll, DB query) into the exchange. The SourceFunc produces a value; the adapter writes it to the body and runs the downstream processor.

Constructor: NewChannelAdapter(sourceName string, sourceFunc, processor) *ChannelAdapter.

Detour¶

Conditionally routes through an extra processor — useful for debug/audit injection without rewriting the main pipeline.

Sets HelixDetourEnabled to indicate activation.

LogEIP¶

Structured logging with optional inclusion of body/headers/properties, MDC (Mapped Diagnostic Context) keys, and custom log function.

Sets HelixLogTimestamp = time.Now() before logging.

Threads¶

Worker-pool wrapper. Submits each call to a bounded semaphore pool; with WaitForSlot: true callers block until a slot is free, otherwise they fail fast with pool exhausted.

steps:
  - threads:
      pool-size: 8
      steps:
        - target: http://cpu-bound

Sets HelixThreadsPoolSize; sets HelixThreadsRejected = true if a non-waiting call was refused. Constructor: NewThreads(processor engine.Processor, poolSize int) *Threads.

Rollback¶

Marks the exchange for rollback. Sets HelixRollbackOnly = true and HelixRollbackMessage = <msg>. By default also calls ex.SetException(err) and returns the error so the surrounding pipeline aborts; MarkOnly: true skips setting the exception (useful for cooperative rollback from within a try-block).

steps:
  - rollback:
      message: "insufficient balance"

Constructor: NewRollback(message string) *Rollback.

Stop¶

Halts routing for the current exchange. Sets HelixRouteStop = true and returns nil. The surrounding pipeline does not automatically honor this — wrap in StopAwarePipeline (or step, which the reifier wraps similarly in some contexts).

steps:
  - choice:
      when:
        - starlang: header("X-Drop") == "1"
          steps:
            - stop: {}
  - target: kafka://outbound

Constructor: NewStop() *Stop.

ServiceCall¶

Looks up a processor in the engine.Registry by service name and invokes it. Optionally runs a ServiceFilter to mutate the exchange first.

Sets HelixServiceCallServiceName and HelixServiceCallServiceFound.

Constructor: NewServiceCall(registry engine.Registry, serviceName string) *ServiceCall.

StopAwarePipeline¶

A Pipeline that checks HelixRouteStop between processors and exits early when set. Use anywhere Stop should terminate the chain.

Constructor: NewStopAwarePipeline(processors ...engine.Processor) *StopAwarePipeline.

Pitfalls¶

• MessageRouter.AllMatch = true runs every matching rule — if rules are order-dependent, list them carefully.
• ScatterGather aggregation does not depend on completion order; results are merged in receipt order from a channel, which is unordered. Use a strategy that's commutative, or post-process to re-order by HelixScatterGatherIndex.
• Threads with WaitForSlot: false returns an error on rejection — callers must handle this or wrap in do-try.
• Stop only works if the surrounding pipeline is StopAwarePipeline; a plain Pipeline ignores HelixRouteStop.

Examples¶

Stop with awareness in a normal Pipeline (use step:):

workflows:
  - id: filtered-pipeline
    trigger: { uri: direct://events }
    steps:
      - step:
          id: filter-and-publish
          steps:
            - choice:
                when:
                  - starlang: body().drop == true
                    steps:
                      - stop: {}
            - target: kafka://events

Worker pool:

workflows:
  - id: bounded-concurrency
    trigger: { uri: kafka://heavy }
    steps:
      - threads:
          pool-size: 16
          steps:
            - target: http://cpu-bound

Source¶

pkg/processor/eip_advanced.go

See also¶

• eip — base EIPs (Saga, ClaimCheck, Sort, Bean, SchemaValidator).
• eip_extra — EnrichProcessor, PollEnrichProcessor, DelayProcessor, LoopProcessor, SampleProcessor, in-memory idempotent repo.
• multicast — Multicast and WireTap; ScatterGather here combines the patterns.
• normalize — first-match-wins variant of Normalizer.
• choice — first-match-wins variant of MessageRouter.