Multicast and WireTap¶

pkg/processor/multicast.go defines the broadcast/fan-out primitives: Multicast EIP sends a copy of an exchange to N processors and optionally aggregates their responses; Wire Tap EIP forks an asynchronous side-channel copy while the original exchange continues unchanged.

Processors in this file¶

NewMulticast¶

Sends a copy of the exchange to every processor in the list. If Parallel is true the processors run concurrently; otherwise sequentially. Without an aggregation strategy, multicast is fire-and-forget (responses are dropped). With a strategy, the merged result replaces the original body.

YAML form:

steps:
  - multicast:
      uris:
        - kafka://orders
        - log://orders-audit
        - http://localhost:8081/replica
      parallel: true
      strategy: list

Constructor: NewMulticast(parallel bool, processors ...engine.Processor) *Multicast or NewMulticastWithStrategy(parallel bool, strategy, processors...).

How it works:

1. Per processor: copy the exchange, set HelixMulticastIndex, run the processor with the copy.
2. Sequential mode runs them in declared order; parallel mode spawns one goroutine per processor and waits with a sync.WaitGroup.
3. If Strategy is set, results are merged in deterministic index order (so output is reproducible even in parallel mode). useOriginal is a fast-path that skips aggregation entirely.
4. Any processor error becomes a multicast[i] error: ... and the multicast returns that error after waiting for all parallel siblings. In sequential mode, errors fail-fast.

Notes and gotchas:

• Order: with parallel: true and an aggregation strategy, the aggregation is done in index order — not completion order — so the resulting list/map is stable.
• A processor that does its own publishing (e.g. target: kafka://x) typically returns no useful body; pair it with useOriginal strategy to keep the source body.
• Multicast does not roll back successful siblings if a later sibling fails. Use Saga for compensation.

NewWireTap¶

Sends a copy of the exchange to a side processor in a goroutine and returns immediately. Errors and panics inside the tap are logged but never propagated.

steps:
  - wiretap: kafka://audit-log
  - target: postgres://orders

Constructor: NewWireTap(processor engine.Processor) *WireTap.

How it works:

1. ex.Copy() — produces a shallow copy of the exchange.
2. Launches a goroutine that processes the copy. A defer recover() wraps the call so a panic in the tap is logged but does not kill the parent.
3. Returns nil immediately — the original exchange continues in the parent pipeline.

Pitfalls:

• ex.Copy() is shallow. If you mutate a shared body slice/map inside the tap, you risk a data race with the parent. Use WireTapEIP with deepCopyEnabled: true for full isolation.
• Tap errors are not observable in the parent route. If you need backpressure or error feedback, use multicast with a strategy instead.

Examples¶

Broadcast with response collection:

workflows:
  - id: query-all-shards
    trigger: { uri: http://localhost:8080/q }
    steps:
      - multicast:
          parallel: true
          strategy: list
          uris:
            - http://shard-1/q
            - http://shard-2/q
            - http://shard-3/q
      - transform:
          starlang: |
            %sl 2.0
            output application/json
            input payload
            ---
            { "results": payload }

Async audit log:

workflows:
  - id: audited-publish
    trigger: { uri: direct://orders }
    steps:
      - wiretap: kafka://audit
      - target: kafka://orders

Source¶

pkg/processor/multicast.go

See also¶

• aggregate — full list of built-in AggregationStrategy implementations.
• routing — RecipientList (dynamic URIs), RoutingSlip (sequential), DynamicRouter (iterative).
• eip_advanced — WireTapEIP with deep-copy and main-pipeline integration.
• eip_advanced — ScatterGather for multicast + gather with a single timeout.