EIP processors (core)¶

pkg/processor/eip.go is the core grab-bag of Enterprise Integration Patterns implementations that don't fit elsewhere: Saga (Long-Running Transaction), Claim Check, Sort, Bean reflection, the original WireTap (renamed WireTapEIP lives in eip_advanced), and the legacy SchemaValidator. The "advanced" and "extra" files extend these with richer variants.

Processors in this file¶

NewSaga¶

Implements the Saga / Long-Running Transaction pattern: execute steps sequentially, and if any step fails, run compensators (rollback actions) for the already-completed steps in reverse order.

steps:
  - saga:
      steps:
        - target: http://reserve-inventory
        - target: http://charge-card
        - target: http://schedule-shipment
      compensate:
        - target: http://release-inventory
        - target: http://refund-card
        # no compensator for shipment because it runs last

Constructor: NewSaga(steps ...SagaStep) *Saga. A SagaStep has Name, Processor, and Compensator.

Behavior:

1. Iterates forward steps. Sets HelixSagaCurrentStep = step.Name before each.
2. On failure: runs compensators for already-completed steps in reverse order. Compensator errors are intentionally swallowed (each gets a chance to run).
3. Sets HelixSagaCompleted to true on success, false on failure.
4. Returns saga step "<name>" (index N) failed: <err> on failure; the original error is wrapped.

Claim Check — NewInMemoryClaimCheckStore, NewClaimCheckIn, NewClaimCheckOut¶

The Claim Check pattern lets you strip large payloads from a message, replace them with a key, and re-attach them later — useful for keeping in-flight messages light.

steps:
  - claim-check:
      operation: push    # also: pop, get
      key: header("X-Order-Id")

ClaimCheckStore interface:

type ClaimCheckStore interface {
    Store(ctx context.Context, key string, data any) error
    Retrieve(ctx context.Context, key string) (any, error)
    Remove(ctx context.Context, key string) error
}

InMemoryClaimCheckStore is a thread-safe map[string]any implementation. For a key removal-on-get version, see ClaimCheckGet.

Constructors:

• NewInMemoryClaimCheckStore() *InMemoryClaimCheckStore
• NewClaimCheckIn(store ClaimCheckStore, keyExpr engine.Expression) *ClaimCheckIn
• NewClaimCheckOut(store ClaimCheckStore, keyExpr engine.Expression) *ClaimCheckOut

NewSort¶

Collects a slice body, sorts via a comparator, and forwards to a downstream processor. Implements the data-flow analog of sort in EIP terms.

steps:
  - sort:
      starlang: body()    # the slice to sort

Constructor: NewSort(processor engine.Processor, comparator func(a, b *exchange.Exchange) bool) *Sort. The default comparator sorts by body-as-string ascending.

Implementation note: each slice element is wrapped in a temporary Exchange so comparators can use the full exchange API (e.g. compare by header). Stable sort.

NewBean¶

Invokes a method on a Go object retrieved from a service registry via reflection.

steps:
  - bean:
      ref: orderService
      method: validate

Constructor: NewBean(registry engine.Registry, beanName, method string) *Bean.

Argument binding (in order, by parameter type):

Return value handling:

• Last return that's an error is checked; non-nil causes the processor to fail.
• First non-error return is written to the body.

Panics inside the method are caught and surfaced as bean: method "X" on "Y" panicked: ....

WireTapEIP (placeholder in this file)¶

The full implementation lives in eip_advanced. This file documents the simple WireTap in multicast.go — note the name collision: WireTap (multicast) is fire-and-forget; WireTapEIP (eip_advanced) integrates with a main pipeline and supports deep copy.

NewSchemaValidator¶

A JSON-Schema-lite validator. Supports type, required, min, max, pattern, nested properties, and strict mode. Compiled regexes are cached (bounded LRU at 1000 entries).

schema := map[string]any{
    "type": "object",
    "required": []any{"name", "age"},
    "properties": map[string]any{
        "name": map[string]any{"type": "string", "pattern": "^[A-Z]"},
        "age":  map[string]any{"type": "number", "min": 0, "max": 150},
    },
}
v := processor.NewSchemaValidator(schema, true)

Validation walks the schema tree recursively. Multiple errors are aggregated into a *SchemaValidationErrors. Each error has Field (dot-path) and Message.

Supported types: string, number/integer, boolean/bool, array, object. Unknown type strings are ignored.

If the body is a JSON-encoded string, the validator tries json.Unmarshal first so you can validate raw HTTP bodies without an explicit unmarshal step.

This is not a full JSON Schema implementation — no $ref, no oneOf, no additionalProperties (closest equivalent is StrictMode). For full Draft 7+ compliance use an external library.

Pitfalls¶

• Saga compensators are best-effort. If a compensator panics or fails, sibling compensators still run — but the failure is swallowed. For audit, log inside each compensator.
• Claim Check store is process-local by default. Use a Redis-backed ClaimCheckStore (implement the interface yourself) for multi-instance setups.
• Bean reflection is slow compared to native processors. Use it for compatibility/glue, not hot paths.
• Sort materializes the entire slice in memory, with one temporary Exchange per element. Fine for thousands of items; bad for millions.

Examples¶

Long-running transaction:

workflows:
  - id: place-order
    trigger: { uri: http://localhost:8080/orders }
    steps:
      - saga:
          steps:
            - target: http://inventory/reserve
            - target: http://payments/charge
            - target: http://logistics/schedule
          compensate:
            - target: http://inventory/release
            - target: http://payments/refund

Claim check across a slow queue:

workflows:
  - id: in
    trigger: { uri: http://localhost:8080/upload }
    steps:
      - claim-check: { operation: push, key: header("X-Doc-Id") }
      - target: kafka://doc-jobs

  - id: worker
    trigger: { uri: kafka://doc-jobs?group=workers }
    steps:
      - claim-check: { operation: pop, key: body() }
      - target: http://parser

Source¶

pkg/processor/eip.go

See also¶

• eip_advanced — ClaimCheckGet, WireTapEIP, Normalizer, MessageRouter, ProcessManager, ScatterGather, MessageBridge, ChannelAdapter, Detour, LogEIP, Threads, Rollback, Stop, ServiceCall, StopAwarePipeline.
• eip_extra — EnrichProcessor, PollEnrichProcessor, DelayProcessor, LoopProcessor, SampleProcessor, InMemoryIdempotentRepository, IdempotentConsumerProcessor.
• error_handling — try/catch, DLQ, redelivery.
• resilience — retry, circuit breaker, throttle, timeout, load balancer.