How do reject and tee work in EK9 stream pipelines?

← Streams and Pipelines · Ref: Q980

reject is the inverse of filter — it removes matching items.
tee copies items to a side collection while continuing the pipeline.

  cat items
    | reject by isBlank
    | tee in backup
    | head 5
    > stdout

reject by predicate: removes items where predicate returns true (opposite of filter by).
tee in collection: copies each item into the collection AND continues the pipeline.

A tee is a lazy peek: it records only the items that flow past it. In the first example head 5 stops the pull, so backup holds the same (at most) 5 non-blank items that are printed, not every non-blank item in the list. Put the tee before a stage that must see everything (sort, tail, join), or drop the head, to capture them all.

Example

defines module qa.streams.rejecttee

  defines function

    isBelowThreshold() as pure
      -> measurement as Float
      <- rtn as Boolean?

      minimumReading <- 10.0
      rtn: measurement < minimumReading

  defines program

    RejectTeeDemo()
      stdout <- Stdout()

      readings <- [5.2, 15.8, 3.1, 22.4, 9.7, 18.3, 7.0]

      validReadings <- List() of Float

      stdout.println("Valid readings (above threshold):")
      cat readings
        | reject by isBelowThreshold
        | tee in validReadings
        > stdout

      stdout.println(`Saved ${validReadings.length()} valid readings`)

Common mistakes

E50001 — tee copies items into a collection that must already exist. 'savedReadings' is not declared anywhere, so E50001 is reported on that line. Declare it first (validReadings <- List() of Float) and name it in the tee. See ek9 -h E50001 for details.

Incorrect:

| tee in savedReadings

Correct:

| tee in validReadings

Coming from another language?

Java: no reject (use filter with negation). No tee (use peek with side-effect). EK9: reject is first-class, tee is explicit side-copy.

Keywords: side copy, reject, pipeline, stream, tee