EIP processors (extra)¶

pkg/processor/eip_extra.go rounds out the EIP catalog with producer-driven enrichers, an in-memory idempotent repository, and richer variants of Delay, Loop, and Sample. These are the building blocks the reifier reaches for when you wire enrich:, poll-enrich:, sample: etc. against pre-resolved producers.

Processors in this file¶

NewEnrichProcessor¶

Sends a copy of the exchange to a pre-resolved producer (any engine.Processor) and merges the response into the original. Differs from Enrich in that it takes a producer directly, not a URI to resolve.

Constructor: NewEnrichProcessor(producer engine.Processor, strategy engine.AggregationStrategy) *EnrichProcessor.

NewPollEnrichProcessor¶

Like EnrichProcessor but creates a fresh exchange to send to the producer rather than copying the current one. Used for polling consumers (a queue/topic) where the input message is irrelevant. Supports an optional timeout.

Constructor: NewPollEnrichProcessor(producer, strategy, timeout) *PollEnrichProcessor.

Compare to PollEnrich, which resolves a URI each call and supports PollingEndpointResolver.

NewDelayProcessor¶

Pure delay with optional async mode.

Constructor: NewDelayProcessor(duration time.Duration, async bool) *DelayProcessor.

Synchronous mode honors ctx.Done() and returns ctx.Err() on early cancellation. Async mode discards the delay channel; useful only when you want to fire some side-effect on a timer but not wait. There is no downstream processor in this variant — for delay + run, see Delay.

NewLoopProcessor¶

Fixed-count loop with the option to run each iteration on a copy of the exchange (so iterations are independent) rather than the same exchange (so iterations accumulate state). Compare to Loop.

Constructor: NewLoopProcessor(count, processor, copyExchange bool) *LoopProcessor. Sets HelixLoopIndex and HelixLoopSize per iteration. Errors wrap the iteration index: loop processor: iteration N failed: ....

NewSampleProcessor¶

Probabilistic OR time-based sampler. Use Rate (a float 0.0-1.0) for random sampling, or Period for "at most one message per duration". If both are set, Period wins.

Constructor: NewSampleProcessor(rate float64, period time.Duration) *SampleProcessor.

Behavior:

• Time-based: stores the last-sent timestamp atomically. If now - last >= Period, updates the timestamp and lets the message through. Otherwise sets HelixSampleSkipped = true and returns nil.
• Rate-based: rolls rand.Float64(). If < Rate, passes. Otherwise sets HelixSampleSkipped = true.

Unlike Sample, this variant doesn't have an inner processor — it returns nil either way. Skipped messages are not surfaced as errors; downstream pipelines should check the property.

NewInMemoryIdempotentRepository¶

Thread-safe in-memory idempotent repository with optional size cap and TTL.

Constructor: NewInMemoryIdempotentRepository(maxSize int, ttl time.Duration) *InMemoryIdempotentRepository.

Implements the IdempotentRepository2 interface (see below). Add returns true if the key was new, false if a duplicate. Internally:

• Stores map[string]time.Time (insertion time).
• On Add at capacity, first calls evictExpiredLocked to remove all expired keys; if still at capacity, calls evictOldestLocked to remove a single oldest key (linear scan).
• TTL is checked on read, not on a background timer.

For Redis-backed distributed dedup see idempotent_redis.

IdempotentRepository2 interface¶

The non-context variant used by IdempotentConsumerProcessor:

type IdempotentRepository2 interface {
    Add(key string) bool   // returns true if key was new
    Contains(key string) bool
    Remove(key string)
    Clear()
}

Distinct from IdempotentRepository (in routing.go) — Add returns a bool here, and there's an extra Clear. Distinct also from DistributedIdempotentRepository (in idempotent_redis.go) — no context, no error returns.

NewIdempotentConsumerProcessor¶

Deduplicates using any IdempotentRepository2. On a duplicate, the inner processor is skipped; the duplicate is either silently dropped (SkipDuplicate: true) or marked via HelixDuplicateMessage = true (SkipDuplicate: false).

Constructor: NewIdempotentConsumerProcessor(repo, keyExpr, processor, skipDuplicate bool) *IdempotentConsumerProcessor.

Pitfalls¶

• DelayProcessor with Async: true returns immediately and runs no downstream processor — it's a fire-and-forget sleep. For "delay then process" use Delay in resilience.go.
• SampleProcessor time-based mode has at-most-once semantics per period but uses wall clock — clock skew or backwards jumps can let extra messages through.
• InMemoryIdempotentRepository eviction is linear in map size. For very large keysets, use the Redis backend.
• IdempotentConsumerProcessor with SkipDuplicate: true returns nil for duplicates — downstream observers can't tell the difference between "skipped duplicate" and "ran successfully".

Examples¶

Probabilistic sampling:

sampled := processor.NewSampleProcessor(0.05, 0) // 5% pass-through

Time-based sampling for log noise control:

oncePerMinute := processor.NewSampleProcessor(0, time.Minute)

In-memory dedup with TTL:

repo := processor.NewInMemoryIdempotentRepository(100_000, 24*time.Hour)
dedup := processor.NewIdempotentConsumerProcessor(
    repo,
    expression.NewStarlang(`header("X-Message-Id")`),
    downstream,
    true, // SkipDuplicate
)

Source¶

pkg/processor/eip_extra.go

See also¶

• eip — core EIPs.
• eip_advanced — ContentEnricher (callback-based instead of producer-based).
• poll_enrich — URI-resolving version with PollingEndpointResolver support.
• idempotent_redis — distributed dedup with TTL.
• resilience — Loop, Delay, Sample variants that include a downstream processor.