Pipeline¶

The Pipeline processor implements the Pipes and Filters EIP. It chains a list of processors and runs them sequentially against the same exchange. Pipeline is the basic building block underneath every Helix workflow — the reifier compiles each workflows[*].steps list into a Pipeline.

Processors in this file¶

NewPipeline¶

YAML form: a workflow's steps: list is automatically wrapped in a Pipeline; there is no dedicated pipeline: keyword. Use step: (see step) for a named, traceable subgroup.

workflows:
  - id: order-pipeline
    trigger: { uri: direct://orders }
    steps:
      - set-header: { name: "X-Source", constant: "api" }
      - transform: { starlang: body() }
      - target: kafka://orders

Constructor: NewPipeline(processors ...engine.Processor) *Pipeline.

How it works:

For each child processor, in order:

1. Checks ctx.Err() — bails immediately if the context is cancelled.
2. Checks ex.IsFailed() — bails if an upstream step set an exception via ex.SetException.
3. Calls proc.Process(ctx, ex). Any error returned aborts the rest of the pipeline.

The exchange is mutated in place: each processor sees the previous one's effects.

Notes and gotchas:

• Pipeline is strict: any non-nil error stops the chain. Use do-try to catch and recover.
• A processor that calls ex.SetException(err) without returning the error itself will still cause the next iteration to bail at the IsFailed() check.
• Use step (StopAwarePipeline is similar) when you also want to honor HelixRouteStop set by stop.

Examples¶

The reifier always produces a Pipeline at the top of a workflow:

workflows:
  - id: example
    trigger: { uri: timer://tick?period=5s }
    steps:
      - log: { message: "tick" }
      - set-body: { constant: "hello" }
      - target: log://stdout

Source¶

pkg/processor/pipeline.go

See also¶

• step — a named, traceable subgroup of processors.
• eip_advanced — StopAwarePipeline variant that honors the Stop processor.
• error_handling — wrap the pipeline in do-try for try/catch/finally.