How do I compose multiple stream operations into a reusable pipeline in EK9?

Build complex pipelines by chaining operations with |. Each operation receives items from the previous stage.

SIMPLE:

  cat items | filter by isValid > stdout

MULTI-STAGE:

  cat items
    | filter by isValid
    | sort by compareByName
    | map with formatForDisplay
    | head 10
    > stdout

WITH SIDE CAPTURE (tee):

  cat items
    | filter by isValid
    | tee in validBackup
    | sort by compareByName
    | head 10
    > stdout

The tee operation copies items into a collection while continuing the pipeline. This lets you capture intermediate results without breaking the flow.

Reusable parts come from the functions you pass to each operation — isValid, compareByName, formatForDisplay are all standalone pure functions that can be tested independently.

Example

defines module qa.streams.pipelinecomposition

  defines record

    Task
      taskName as String: String()
      priority as Integer: 0
      completed as Boolean: false

      Task()
        ->
          taskName as String
          priority as Integer
          completed as Boolean
        this.taskName :=: taskName
        this.priority :=: priority
        this.completed :=: completed

      default operator

  defines function

    isIncomplete() as pure
      -> task as Task
      <- rtn as Boolean: not task.completed

    compareByPriority() as pure
      ->
        left as Task
        right as Task
      <- rtn as Integer: left.priority <=> right.priority

  defines program

    PipelineCompositionDemo()
      stdout <- Stdout()

      tasks <- [
        Task("Write tests", 2, false),
        Task("Fix bug", 1, true),
        Task("Review PR", 1, false),
        Task("Deploy", 3, false),
        Task("Update docs", 2, true)
      ]

      //Multi-stage: filter incomplete, sort by priority, take top 2
      stdout.println("Top priority incomplete tasks:")
      cat tasks
        | filter by isIncomplete
        | sort by compareByPriority
        | head 2
        > stdout

Other ways to ask this

Can I break a complex pipeline into smaller reusable parts?
How do I build a multi-stage stream pipeline in EK9?
Show me a complex stream pipeline with multiple operations

Coming from another language?

EK9 pipeline composition uses named pure functions at each stage. Each function is independently testable and reusable across different pipelines.

Keywords: reusable, compose, multi-stage, tee, stream, pipeline