Skip to content

OnCompletion

OnCompletion wraps a processor and runs a callback after the wrapped processor finishes — either on success, on failure, or both. It implements the common cleanup/audit pattern.

Processors in this file

NewOnCompletion

YAML form:

steps:
  - on-completion:
      on-complete:
        - target: kafka://audit-success
      on-failure:
        - log: { message: "failed", level: ERROR }
        - target: kafka://audit-failure
      on-when: header("X-Audit") == "true"
      after-consumer: false
Field Type Default Description
on-complete list optional Steps to run when the wrapped processor succeeds AND ex.IsFailed() is false.
on-failure list optional Steps to run when the wrapped processor returns an error OR ex.IsFailed() is true.
on-when string optional Starlang predicate. If set, callbacks only run when this matches. Errors evaluating it are logged but do not affect the wrapped processor's return value.
after-consumer bool false Whether the callback should fire after the consumer ack rather than just after the route. (Currently stored; semantics depend on the consumer.)

Constructor: NewOnCompletion(processor engine.Processor) *OnCompletion. The Processor field holds the wrapped pipeline; assign OnComplete, OnFailure, OnWhen directly.

How it works:

  1. Runs the wrapped Processor. Captures the resulting error.
  2. If OnWhen is set, evaluates it. If false (or errored), skips callbacks and returns the original error from step 1.
  3. If step 1 succeeded and the exchange isn't marked failed, makes a copy of the exchange, sets HelixOnCompletionStatus = "success", and runs OnComplete on the copy.
  4. Otherwise (error or ex.IsFailed()), copies the exchange, sets HelixOnCompletionStatus = "failure", sets HelixExceptionCaught to the captured error, and runs OnFailure on the copy.
  5. Returns the original error from step 1 — callbacks do not transform the result.

Notes and gotchas:

  • Callbacks run on a copy of the exchange. Mutating headers/body in on-complete will not affect the upstream pipeline. This is so audit/cleanup logic can't accidentally rewrite the response.
  • A callback error is logged via slog.Error and otherwise ignored — the wrapped processor's status wins.
  • The on-when predicate is evaluated against the post-processing exchange state, so you can branch on the failure exception itself via property("HelixExceptionCaught").

Examples

workflows:
  - id: process-with-audit
    trigger: { uri: direct://orders }
    steps:
      - on-completion:
          on-complete:
            - set-body: { starlang: '{ "status": "ok", "id": header("X-Order-Id") }' }
            - target: kafka://order-success
          on-failure:
            - set-body: { starlang: '{ "status": "failed", "err": property("HelixExceptionCaught") }' }
            - target: kafka://order-failure
      - target: postgres://orders

Source

pkg/processor/on_completion.go

See also

  • error_handling — do-try for catch/finally; DeadLetterChannel for redelivery + DLQ.
  • eip_advanced — WireTapEIP for parallel side-channel observation.