Split¶

Split implements the Splitter EIP: it evaluates an expression that yields a slice (or a channel, for streaming mode), then invokes an inner processor for each element. Per-element results can be aggregated back into the original exchange with any of the built-in aggregation strategies.

Processors in this file¶

`NewSplit`¶

YAML form:

steps:
  - split:
      starlang: body().items
      parallel: false
      streaming: false
      stop-on-exception: false
      share-unit-of-work: false
      aggregation-strategy: list
      steps:
        - target: kafka://order-line

Field	Type	Default	Description
`starlang`	string	required	Starlang expression that returns a slice (`[]any`, `[]string`, `[]map[string]any`) or — when `streaming: true` — a `chan any`.
`parallel`	bool	`false`	Spawn one goroutine per element. Order is preserved in the aggregated output but inner steps run concurrently.
`streaming`	bool	`false`	Treat the expression result as a `chan any` and process items as they arrive. Mutually exclusive with `parallel`.
`stop-on-exception`	bool	`false`	In sequential mode: abort on the first error. In parallel mode: cancel sibling goroutines via a derived `context.WithCancel`.
`share-unit-of-work`	bool	`false`	Pass the parent exchange's `UnitOfWork` to each split exchange so commits/rollbacks coordinate.
`aggregation-strategy`	string	none	Built-in strategy name (`list`, `latest`, `first`, `useOriginal`, `map`). If unset, per-element results are discarded.
`steps`	list	required	Inner pipeline applied to each element.

Constructor: NewSplit(expression engine.Expression, processor engine.Processor) *Split. Set the bool fields and Strategy directly on the returned struct.

How it works:

Evaluates Expression. If streaming: true and the result is a channel, switches to processStreaming (otherwise falls through to slice mode).
Calls toSlice to coerce the result. Supports []any, []string, []map[string]any. Anything else returns an error.
For each element, calls createSplitExchange: a fresh exchange with the element as body, parent headers copied, and the properties:
HelixSplitIndex (0-based)
HelixSplitSize (total count; -1 in streaming mode)
HelixSplitComplete (true on the last element)
Runs Processor.Process(ctx, splitEx). If Strategy is set, folds the result into a running aggregate.
After all elements are processed, if Strategy is set, writes the aggregated body back to the original exchange.

Modes¶

Sequential (parallel: false, streaming: false): straight for loop. Errors propagate immediately; stop-on-exception is implicit because we already abort on first error.
Parallel: one goroutine per element, results collected in an indexed slice and merged in index order at the end. With stop-on-exception, a single error cancels a derived context which other workers honor.
Streaming: blocks on the result channel. Items are processed as they arrive; size is unknown so HelixSplitSize is -1 and HelixSplitComplete is always false. Closing the channel completes the split.

Pitfalls¶

The expression result must be a slice or channel type listed above. A bare string like "a,b,c" is not auto-split — use Starlang split(s, ",") first.
In parallel mode, an error in one element waits for all in-flight siblings to finish before returning (unless stop-on-exception is set, which triggers cooperative cancel).
Streaming mode + stop-on-exception skips remaining items but does not drain the channel; senders may block. Have producers respect ctx.Done().
Inner processors mutate the split exchange. Aggregating with the list strategy preserves their bodies; without a strategy, mutations are lost.

Examples¶

Sequential split with list aggregation:

workflows:
  - id: explode-line-items
    trigger: { uri: direct://orders }
    steps:
      - split:
          starlang: body().lineItems
          aggregation-strategy: list
          steps:
            - transform:
                starlang: |
                  %sl 2.0
                  output application/json
                  input payload
                  ---
                  { "sku": payload.sku, "qty": payload.qty, "amount": payload.price * payload.qty }
      - target: kafka://line-item-events

Parallel split with map aggregation, stop on first error:

workflows:
  - id: query-shards-then-merge
    trigger: { uri: http://localhost:8080/q }
    steps:
      - set-body:
          starlang: |
            %sl 2.0
            output application/json
            ---
            ["shard-1", "shard-2", "shard-3"]
      - split:
          starlang: body()
          parallel: true
          stop-on-exception: true
          aggregation-strategy: map
          steps:
            - target: http://internal-router/{HelixSplitIndex}

Source¶

pkg/processor/split.go