Streams and Pipelines in EK9

The Introduction touched on pipelining; read that section first. This page has the detail.

Concept

A pipeline streams zero or more Objects through a sequence of commands, as the Unix shell does and as chained methods (filter, map, reduce) do in other languages. Along the way a command can stop an Object going further, or transform it into a different type, enriching or simplifying the data it holds. Whatever emerges at the end is then used.

So a stream has three phases:

Catenate a source of Objects
The processing pipeline itself; with commands and functions
The collection or reduction of the Objects as they exit the pipeline

Starting the stream

There are two ways to start a stream:

cat (short for catenate) - for collections, iterators and other 'sources' of Objects. See cat flow control.
for - the 'range' for loop. See for range flow control.

Any type of Object can go through a pipeline, including function delegates. That matters when you want 'asynchronous' processing.

Pipeline processing

The pipe '|' joins the parts of the pipeline, as in Unix. The commands are:

filter/reject - filter keeps items matching a predicate, reject removes items matching a predicate.
map - transform an item into an item of a different type.
sort - order all items with a function and pass along the pipeline.
group - use a function to group items.
join - accept two items and join them returning a new item of the same type.
split - split a stream into groups using a predicate function. When the predicate returns true, the current group is saved and a new group begins (the triggering item starts the new group). Produces a stream of 'List of T'. Typically followed by 'flatten' to unpack groups, or 'map' to process each group independently.
uniq - ensure all items passing through are unique.
tee - take a side copy of the items that flow past it into another terminal node; with a later head, only the items that head demands.
head - limit the output to first 'n' items, and stop pulling items once it has them.
tail - limit the output to the last 'n' items.

EK9 adds:

skip - ignore the first 'n' items and then output all the rest.
flatten - if the item is some type of collection, this streams each item out from that collection.
call - if the item in the pipeline is a function - it is called.
async - if the item in the pipeline is a function it is called asynchronously. Results flow on in the order the calls complete, not the order the functions arrived.

Items are pulled on demand. Once a head has its 'n' items nothing more is pulled, so the stages before it run only for the items it needs: cat items | filter by expensive | head 1 stops at the first match, and head also ends a stream over a source that never ends.

The stop reaches back only as far as the nearest stage that must see every item before it can produce anything: sort, tail and join (and group and split) always read all of their input, and a head after them limits only what they pass on. So a stage function that prints or records something runs once per item actually pulled, and a tee records exactly the items that flow past it.

cat a, b is one stream: all of a flows through first, then b. Each source is pulled only when it is reached, so with cat a, b | head 2, when a has two items b is never read, and a source that never ends can follow a finite one.

async | head n gives the first 'n' results to complete, so which items they are (and their order) can change from run to run.

When nothing between the async and the head can drop an item (only map, call or tee), only 'n' calls are started.
When a stage between them can drop items, such as filter, more calls are started (up to a fixed window), because it is not known which of them will produce a wanted result. The results not needed are discarded.

A function called by async runs on another thread, so anything it captured and mutates (a field, a list, a counter) must be changed inside a MutexLock. Otherwise the compiler reports E08251, including for > target, >> target and | tee in target. Stdout/Stderr may be shared across workers; each line is written whole, but the order of lines from concurrent workers is not guaranteed.

Collection or Reduction

Zero or more Objects flow out of the end of the pipeline, and you decide what happens to them:

Direct them into a 'sink' of Objects (an output, or a collection) - use '>'. Into a List, Set, Dict, PriorityQueue or JSON array it replaces the collection's contents, and into a file output it clobbers the file, just as the shell's > replaces a file
Add them to what an existing collection or file already holds - use '>>', as the shell's >> appends to a file
Collect them into a new Collection or Object - using '| collect as'

#!ek9
defines module site.streamsAndPipelines.sink.replaces

  defines function

    entryFor()
      -> number as Integer
      <- rtn as DictEntry of (String, Integer): DictEntry($number, number)

  defines program

    SinkReplacesDemo()
      stdout <- Stdout()

      names <- ["p"]
      cat ["q", "r"] > names
      stdout.println(`after >: ${names}`)

      cat ["s"] >> names
      stdout.println(`after >>: ${names}`)

      letters <- Set() of String
      letters += "x"
      cat ["y", "z"] > letters
      stdout.println(`set after >: ${letters}`)

      scores <- Dict() of (String, Integer)
      scores += DictEntry("old", 0)
      for i in 1 ... 2 | map with entryFor > scores
      stdout.println(`dict after >: ${scores}`)

      nothing <- List() of String
      cat nothing > names
      stdout.println(`empty source: ${length names}`)

Download streamsAndPipelines_sink_replaces.ek9

The first '>' leaves names holding only "q" and "r" - the "p" it held is gone - and '>>' then adds "s". A Set and a Dict behave the same way, and a source with no items leaves the collection empty. It prints:

after >: ["q", "r"]
after >>: ["q", "r", "s"]
set after >: {"y", "z"}
dict after >: {1=1, 2=2}
empty source: 0

A '>' into a collection is all or nothing. The pipeline fills a new collection of the target's own type and, only when the pipeline has run to its end, that collection replaces what the target held. So every source reads the target as it was, even when the target is also a source (cat names, ["x"] > names reads names in full first); a pipeline that throws part-way leaves the target exactly as it was; and anything else that refers to the same collection sees the result. '>>' adds to the target as items arrive, as it always did.

#!ek9
defines module site.streamsAndPipelines.sink.all.or.nothing

  defines constant

    badNumber <- 3

  defines function

    checked()
      -> number as Integer
      <- rtn as String: String()
      if number == badNumber
        throw Exception("bad item")
      rtn: $number

  defines program

    SinkAllOrNothingDemo()
      stdout <- Stdout()

      names <- ["a", "b"]
      cat names, ["x"] > names
      stdout.println(`both sources read: ${names}`)

      kept <- ["old"]
      try
        for i in 1 ... 5 | map with checked > kept
      catch
        -> ex as Exception
        stdout.println(`failed part-way, still: ${kept}`)

      alias <- kept
      cat ["new"] > kept
      stdout.println(`an alias sees it: ${alias}`)

Download streamsAndPipelines_sink_all_or_nothing.ek9

It prints:

both sources read: ["a", "b", "x"]
failed part-way, still: ["old"]
an alias sees it: ["new"]

A PriorityQueue and a JSON array are replaced in just the same way, and '>>' adds to them. A queue keeps what it was built with - its comparator and its size limit - when it is replaced, so the items that arrive are ordered, and trimmed, exactly as before. A JSON target follows its shape: an array (or a JSON that is not yet set) is replaced by the array the stream collects, while a JSON object keeps merging each streamed object into itself, as it always did - now all or nothing.

#!ek9
defines module site.streamsAndPipelines.sink.queue.and.json

  defines program

    SinkQueueAndJsonDemo()
      stdout <- Stdout()

      topTwo <- (PriorityQueue() of Integer).withSize(2)
      topTwo += 100
      cat [4, 8, 6] > topTwo
      stdout.println(`queue after >: ${topTwo.list()}`)

      cat [9] >> topTwo
      stdout.println(`queue after >>: ${topTwo.list()}`)

      numbers <- JSON().array()
      cat [JSON(1), JSON(2)] >> numbers
      cat [JSON(3)] > numbers
      stdout.println(`array after >: ${numbers}`)

      cat [JSON(4)] >> numbers
      stdout.println(`array after >>: ${numbers}`)

Download streamsAndPipelines_sink_queue_and_json.ek9

It prints:

queue after >: [8, 6]
queue after >>: [9, 8]
array after >: [3]
array after >>: [3,4]

'>>' adds as items arrive, so it must never add to the List that 'cat' is still reading: cat names >> names is an error (E08277), the same thing GNU cat says as "input file is output file", and so is a 'tee' into that List. Read a copy, or replace the List with '>'. Where the compiler cannot see that two names are one List (an alias, or one List passed as both source and target), the program raises an Exception saying the List changed while it was being read.

#!ek9
defines module site.streamsAndPipelines.sink.not.itself

  defines program

    SinkNotItselfDemo()
      stdout <- Stdout()

      //'cat numbers >> numbers' is E08277: '>>' would add to the List while 'cat' is still reading it.
      //Read a copy and append that.
      numbers <- [1, 2, 3]
      copy <- List() of Integer
      copy += numbers
      cat copy >> numbers
      stdout.println(`appended: ${numbers}`)

      //'>' fills a fresh List and swaps it in when the stream is finished, so reading what it held is fine.
      cat numbers | head 3 > numbers
      stdout.println(`replaced: ${numbers}`)

Download streamsAndPipelines_sink_not_itself.ek9

It prints:

appended: [1, 2, 3, 1, 2, 3]
replaced: [1, 2, 3]

A file output follows the shell too. TextFile(path).output() opens the file at once, but it empties the file only when something is first written: '>' clobbers the file, while '>>' keeps what the file holds and adds to it. To keep a header, write the header first (with println) and then use '>>' for the lines that follow it: a '>' after the header would empty the file again, header included.

#!ek9
defines module site.streamsAndPipelines.sink.files

  defines function

    linesOf()
      -> path as FileSystemPath
      <- rtn as List of String: List()
      cat TextFile($path).input() > rtn

  defines program

    SinkFilesDemo()
      stdout <- Stdout()
      tempDir <- FileSystemPath().withTemporaryDirectory()
      path <- tempDir + `ek9_site_sink_files_${$GUID()}.txt`
      created <- path.createFile()
      if created.isError()
        stdout.println("could not create the file")

      try
        -> output <- TextFile(path).output()
        output.println("header")
        cat ["line 1", "line 2"] >> output
      stdout.println(`header then >>: ${linesOf(path)}`)

      try
        -> output <- TextFile(path).output()
        cat ["line 3"] >> output
      stdout.println(`appended: ${linesOf(path)}`)

      try
        -> output <- TextFile(path).output()
        cat ["fresh"] > output
      stdout.println(`clobbered: ${linesOf(path)}`)

Download streamsAndPipelines_sink_files.ek9

It prints:

header then >>: ["header", "line 1", "line 2"]
appended: ["header", "line 1", "line 2", "line 3"]
clobbered: ["fresh"]

Give your own records and classes a '|' operator and they can be the terminal ('collector'/'sink') of a pipeline.

If you know the Unix/Linux shell, this syntax will be familiar.

Exceptions in pipelines

A pipeline is a single unit. When your code in a stage throws an Exception and does not catch it itself, the whole pipeline stops: no further item is pulled from the source, no later stage runs and nothing more reaches the terminal. The exception then goes to the nearest enclosing try, in the function that runs the pipeline or in whatever called it. If nothing catches it, the program ends with the exception's own exit code.

This is the same for every stage that runs your code: filter/reject, map, call, flatten, sort with a comparator, group and uniq with a key function, tee into your own sink, your own source iterator, collect into your own type and > into your own StringOutput. The runtime never catches the exception and carries on.

#!ek9
defines module site.streamsAndPipelines.exceptions

  defines function

    parsePrice()
      -> text as String
      <- rtn as Integer: Integer(text)
      if not rtn?
        throw Exception(`Not a price: ${text}`, 3)

    parsePriceOrZero()
      -> text as String
      <- rtn as Integer: 0
      try
        rtn: parsePrice(text)
      catch
        -> ex as Exception
        Stdout().println(`skipped: ${ex.reason()}`)

  defines program

    PipelineExceptionDemo()
      stdout <- Stdout()
      prices <- ["12", "oops", "30", "45"]

      try
        total <- cat prices | map with parsePrice | collect as List of Integer
        stdout.println(`unreachable: ${total}`)
      catch
        -> ex as Exception
        stdout.println(`pipeline stopped: ${ex.reason()}`)

      recovered <- cat prices | map with parsePriceOrZero | collect as List of Integer
      stdout.println(`recovered: ${recovered}`)

Download streamsAndPipelines_exceptions.ek9

The first pipeline stops at "oops"; "30" and "45" are never parsed. To skip a bad item and keep going, catch inside the function the stage calls, as parsePriceOrZero does: the second pipeline gives [12, 0, 30, 45]. It prints:

pipeline stopped: Not a price: oops
skipped: Not a price: oops
recovered: [12, 0, 30, 45]

async: when a task throws while others are still running, the first failure observed propagates, when the pipeline next takes a result. Before it does, the other tasks are cancelled and waited for, so no task outlives the pipeline and none is still running when your catch runs.
Resources: a pipeline inside a try with resources behaves like any other statement in it. The exception goes to the catch, then finally runs, and then the resources are closed in reverse declaration order. In EK9 a resource stays open for the catch and finally blocks, so they can still use it (see Exceptions/Error handling).

Examples

The examples below each have an explanation after them. Other pages show join, call and async; the duration/date example shows '>>', and Integer pipeline shows 'reducing' or 'accumulating'.

To keep only the best N of the Objects that reach the end of a pipeline, by some weighting, use a priority queue with a finite size.

The Preamble and setup

The examples use a library of published 'Books' and their 'Authors', and extract information about the books with pipelines. The first few are trivial; the later ones build up a significant pipeline. The setup source is shown once here and used by every example.

defines constant
  MINIMUM_BOOKS <- 3
  SENIOR_AGE <- 50

defines type

  AuthorId as Integer constrain
    > 0

  Age as Integer constrain
    > 0

  Name as String constrain as
    matches /^[a-zA-Z -]+$/

  BookTitle as String constrain as
    matches /^[a-zA-Z0-9 -+]+$/

  <?-
    Simple enumeration for controlling filtering
  -?>
  BookFilterSelection
    SkipTwo
    JustFirst
    JustLast

defines class

  <?-
    Models the author of books.
  -?>
  Author
    id as AuthorId?
    age as Age?
    firstname as Name?
    surname as Name?

    Author() as pure
      ->
        id as AuthorId
        age as Age
        firstname as Name
        surname as Name
      require id? and age? and firstname? and surname?
      this.id:=? AuthorId(id)
      this.age:=? Age(age)
      this.firstname:=? Name(firstname)
      this.surname:=? Name(surname)

    Author() as pure
      -> author as Author
      this(id: author.id(), age: author.age(), firstname: author.firstname(), surname: author.surname())

    <?-
      Stop creation of empty Author.
    -?>
    default private Author() as pure

    id() as pure
      <- rtn as AuthorId: id

    age() as pure
      <- rtn as Age: age

    firstname() as pure
      <- rtn as Name: firstname

    surname() as pure
      <- rtn as Name: surname

    operator == as pure
      -> author as Author
      <- rtn as Boolean: id == author.id()

    operator #? as pure
      <- rtn as Integer: #?id

    operator $ as pure
      <- rtn as String: `Author: ${firstname} ${surname}, Age: ${age}`

    override operator ? as pure
      <- rtn as Boolean: age? and firstname? and surname?

    operator :=:
      -> author as Author
      id :=: author.id()
      age :=: author.age()
      firstname :=: author.firstname()
      surname :=: author.surname()

  Book
    title as BookTitle?
    author as Author?
    published as Date?

    Book() as pure
      ->
        title as BookTitle
        author as Author
        published as Date
      require title? and author? and published?
      this.title:=? BookTitle(title)
      this.author:=? Author(author)
      this.published:=? Date(published)

    <?-
      Stop creation of empty Book.
    -?>
    default private Book() as pure

    author() as pure
      <- rtn as Author: author

    published() as pure
      <- rtn as Date: published

    operator $ as pure
      <- rtn as String: `Title: ${title}, ${author}, Published: ${published}`

    operator #? as pure
      <- rtn as Integer: #? $this

    operator #^ as pure
      <- rtn as String: $this

    operator <=> as pure
      -> book as Book
      <- rtn as Integer: published() <=> book.published()

    override operator ? as pure
      <- rtn as Boolean: title? and author? and published?

  <?-
    Model the concept of a library, just an in-memory map for now
  -?>
  Library
    books as List of Book?

    Library() as pure
      -> withBooks as List of Book
      books :=? withBooks

    Library() as pure
      this([
        Book(BookTitle("Java"), Author(id: AuthorId(1), age: Age(50), firstname: Name("John"), surname: Name("Doe")), 1998-01-01),
        Book(BookTitle("C++"), Author(id: AuthorId(1), age: Age(42), firstname: Name("John"), surname: Name("Doe")), 1990-01-07),
        Book(BookTitle("Scala"), Author(id: AuthorId(1), age: Age(67), firstname: Name("John"), surname: Name("Doe")), 2015-03-02),
        Book(BookTitle("Python"), Author(id: AuthorId(1), age: Age(62), firstname: Name("John"), surname: Name("Doe")), 2010-12-02),
        Book(BookTitle("HTML"), Author(id: AuthorId(2), age: Age(58), firstname: Name("Mark"), surname: Name("Pickford")), 2008-07-02),
        Book(BookTitle("CSS"), Author(id: AuthorId(4), age: Age(51), firstname: Name("Mark"), surname: Name("Keely")), 2008-04-02),
        Book(BookTitle("ADA"), Author(id: AuthorId(5), age: Age(44), firstname: Name("Ada"), surname: Name("Lovelace")), 1988-01-02),
        Book(BookTitle("Dart"), Author(id: AuthorId(6), age: Age(47), firstname: Name("Peter"), surname: Name("Dove")), 2020-01-02),
        Book(BookTitle("C#"), Author(id: AuthorId(7), age: Age(60), firstname: Name("William"), surname: Name("Fence")), 2012-10-02),
        Book(BookTitle("Javascript"), Author(id: AuthorId(3), age: Age(52), firstname: Name("James"), surname: Name("Pickford")), 2008-03-02),
        Book(BookTitle("C"), Author(id: AuthorId(1), age: Age(42), firstname: Name("John"), surname: Name("Doe")), 1990-01-07),
        Book(BookTitle("C++"), Author(id: AuthorId(7), age: Age(38), firstname: Name("William"), surname: Name("Fence")), 1990-04-02),
        Book(BookTitle("C"), Author(id: AuthorId(7), age: Age(38), firstname: Name("William"), surname: Name("Fence")), 1990-04-14),
        Book(BookTitle("Haskell"), Author(id: AuthorId(7), age: Age(30), firstname: Name("William"), surname: Name("Fence")), 1982-04-14),
        Book(BookTitle("Lisp"), Author(id: AuthorId(7), age: Age(25), firstname: Name("William"), surname: Name("Fence")), 1977-09-24)
        ])

    <?-
      While you would never really 'iterate' through all the books in a library - this
      just enables the demonstration of streaming.
      Clearly you would normally have multiple maps/indexes etc to locate books via a search
      mechanism.
    -?>
    iterator() as pure
      <- rtn as Iterator of Book: books.iterator()

    operator + as pure
      -> book as Book
      <- rtn as Library: Library(books + book)

    override operator ? as pure
      <- rtn as Boolean: books?

Full source: streamsAndPipelines_library.ek9 - Download streamsAndPipelines_library.ek9

The Model

The model is a few types (strong typing of simple values) and classes for Books, Authors and the Library. Because pipelines are the functional part of EK9, the model uses pure for immutability, and the compiler enforces it.

Strictly, the accessor methods should return clones/copies of the properties, so that nothing outside could alter the Objects provided or returned.

The Programs

The first program outputs all the books to 'Standard Out'.

Program 1

defines program

  <?-
    Just send all the books in the library out to stdout.
    Because Library has an iterator method 'cat' will use it to get each book.
  -?>
  JustCatBooks()
    stdout <- Stdout()
    library <- Library()

    cat library > stdout

Full source: streamsAndPipelines_library.ek9 - Download streamsAndPipelines_library.ek9

It prints:

Title: Java, Author: John Doe, Age: 50, Published: 1998-01-01
Title: C++, Author: John Doe, Age: 42, Published: 1990-01-07
Title: Scala, Author: John Doe, Age: 67, Published: 2015-03-02
Title: Python, Author: John Doe, Age: 62, Published: 2010-12-02
Title: HTML, Author: Mark Pickford, Age: 58, Published: 2008-07-02
Title: CSS, Author: Mark Keely, Age: 51, Published: 2008-04-02
Title: ADA, Author: Ada Lovelace, Age: 44, Published: 1988-01-02
Title: Dart, Author: Peter Dove, Age: 47, Published: 2020-01-02
Title: C#, Author: William Fence, Age: 60, Published: 2012-10-02
Title: Javascript, Author: James Pickford, Age: 52, Published: 2008-03-02
Title: C, Author: John Doe, Age: 42, Published: 1990-01-07
Title: C++, Author: William Fence, Age: 38, Published: 1990-04-02
Title: C, Author: William Fence, Age: 38, Published: 1990-04-14
Title: Haskell, Author: William Fence, Age: 30, Published: 1982-04-14
Title: Lisp, Author: William Fence, Age: 25, Published: 1977-09-24

No processing has taken place yet. Because the Book class provides the '#^' (promote) operator, a Book can be sent straight to stdout: it is converted (promoted) to a String.

Program 2

This program sorts the Books by Author ID and outputs them all to 'Standard Out'.

sort is the first stream pipeline command here. Its single parameter is a function: a named function, a call to a higher order function that returns one, or a dynamic function. Here it is a simple pure function.

A library of short pure functions, with a set of records, traits and classes, gives you reuse, and lets you write unit tests and examples for each part.

defines function

  comparingAuthor() as pure
    ->
      book1 as Book
      book2 as Book
    <-
      rtn as Integer: book1.author().id() <=> book2.author().id()

defines program

  <?-
    Now sort the books before outputting.
    This uses a specific comparator.
  -?>
  SortBooksByAuthor()
    stdout <- Stdout()
    library <- Library()

    cat library | sort by comparingAuthor > stdout

Full source: streamsAndPipelines_library.ek9 - Download streamsAndPipelines_library.ek9

It prints, in Author ID order:

Title: Java, Author: John Doe, Age: 50, Published: 1998-01-01
Title: C++, Author: John Doe, Age: 42, Published: 1990-01-07
Title: Scala, Author: John Doe, Age: 67, Published: 2015-03-02
Title: Python, Author: John Doe, Age: 62, Published: 2010-12-02
Title: C, Author: John Doe, Age: 42, Published: 1990-01-07
Title: HTML, Author: Mark Pickford, Age: 58, Published: 2008-07-02
Title: Javascript, Author: James Pickford, Age: 52, Published: 2008-03-02
Title: CSS, Author: Mark Keely, Age: 51, Published: 2008-04-02
Title: ADA, Author: Ada Lovelace, Age: 44, Published: 1988-01-02
Title: Dart, Author: Peter Dove, Age: 47, Published: 2020-01-02
Title: C#, Author: William Fence, Age: 60, Published: 2012-10-02
Title: C++, Author: William Fence, Age: 38, Published: 1990-04-02
Title: C, Author: William Fence, Age: 38, Published: 1990-04-14
Title: Haskell, Author: William Fence, Age: 30, Published: 1982-04-14
Title: Lisp, Author: William Fence, Age: 25, Published: 1977-09-24

Program 3

John Doe has two books published on the same date (1990-01-07). This program groups the books by the same author published on the same date. group joins only adjacent items whose keys are equal, so the books are first sorted by author and then by date ('comparingAuthorThenDate'); each group is a List of Book, output on its own line.

defines function

  dateBookPublished() as pure
    -> book as Book
    <- rtn as Date: book.published()

  comparingAuthorThenDate() as pure
    ->
      book1 as Book
      book2 as Book
    <-
      rtn as Integer: comparingAuthor(book1, book2)
    if rtn == 0
      rtn: book1 <=> book2

defines program

  <?-
    Example of grouping books by the same author that were published on the same date.
    'group' only joins ADJACENT books whose keys are equal, so the books are first sorted by
    author and then by published date; each run of equal dates then becomes one group.
    The groups are output in the order they were created, each as a List of Book on its own line.
  -?>
  GroupBooksByPublishedDate()
    stdout <- Stdout()
    library <- Library()
    cat library | sort by comparingAuthorThenDate | group by dateBookPublished > stdout

Full source: streamsAndPipelines_library.ek9 - Download streamsAndPipelines_library.ek9

It prints:

[Title: C++, Author: John Doe, Age: 42, Published: 1990-01-07, Title: C, Author: John Doe, Age: 42, Published: 1990-01-07]
[Title: Java, Author: John Doe, Age: 50, Published: 1998-01-01]
[Title: Python, Author: John Doe, Age: 62, Published: 2010-12-02]
[Title: Scala, Author: John Doe, Age: 67, Published: 2015-03-02]
[Title: HTML, Author: Mark Pickford, Age: 58, Published: 2008-07-02]
[Title: Javascript, Author: James Pickford, Age: 52, Published: 2008-03-02]
[Title: CSS, Author: Mark Keely, Age: 51, Published: 2008-04-02]
[Title: ADA, Author: Ada Lovelace, Age: 44, Published: 1988-01-02]
[Title: Dart, Author: Peter Dove, Age: 47, Published: 2020-01-02]
[Title: Lisp, Author: William Fence, Age: 25, Published: 1977-09-24]
[Title: Haskell, Author: William Fence, Age: 30, Published: 1982-04-14]
[Title: C++, Author: William Fence, Age: 38, Published: 1990-04-02]
[Title: C, Author: William Fence, Age: 38, Published: 1990-04-14]
[Title: C#, Author: William Fence, Age: 60, Published: 2012-10-02]

John Doe's two books of 1990-01-07 are one group; every other book is a group of its own. Sorted only by author, those two books would not be adjacent, and group would not join them.

Program 4

This program orders each author's books by published date, keeping the books in author order.

defines function

  compareDatePublished() as pure
    ->
      book1 as Book
      book2 as Book
    <-
      rtn as Integer: book1.published() <=> book2.published()

  orderOnPublishedDate() as pure
    -> books as List of Book
    <- rtn as List of Book: cat books | sort by compareDatePublished | collect as List of Book

defines program

  <?-
    This example outputs the books in author order, but also for each of those authors sorts their
    books by published date.
  -?>
  ProcessByAuthor()
    stdout <- Stdout()
    library <- Library()

    //This is the nearest ek9 gets to a lambda - it's a bit 'wordy'.
    //In effect we are creating a function delegate that is a 'Function' that accepts a 'Book' and
    //returns an 'AuthorId' type. It is pure in nature - it does not mutate the book in anyway.
    //The arguments of the 'Function' inferred from its definition and so can just be used in the parenthesis.

    authorId <- () is Function of (Book, AuthorId) as pure function (r:=? t.author().id())

    cat library
      | sort by comparingAuthor
      | group by authorId
      | map by orderOnPublishedDate
      | flatten
      > stdout

Full source: streamsAndPipelines_library.ek9 - Download streamsAndPipelines_library.ek9

The list is ordered by Author ID first, then grouped into Lists of Books, one per author.

Next, map uses the function orderOnPublishedDate. The Object in the pipeline is now a 'List of Book', not a 'Book', so orderOnPublishedDate accepts a 'List of Book'. It orders that list with a pipeline of its own and returns a new, ordered list, using the terminal command collect as.

To get a stream of 'Book' again rather than of 'List of Book', flatten (thank you origin?) outputs the Books of each List in sequence. flatten works with Optional and Result too.

With pipelines (and chained methods in other languages) it can be hard to know the type at any point, because processing can change it. In EK9 the strong and explicit typing of functions and methods tells you:

sort never alters the type; so whatever the sort function uses for parameters is the pipeline type.
split/group always produce a 'List' of the current pipeline type.
map is where types are transformed: look at the input type and the output type of its function.
flatten takes something like a 'List of Book' and outputs 'Book' as the pipeline type (each Book in that List).

Summary

Each function has a single purpose, can be reused and can be tested in isolation, and the pipeline itself stays readable. It takes more thought to design than an imperative loop with state variables.

It prints, in author order and, for each author, in publication date order:

Title: C++, Author: John Doe, Age: 42, Published: 1990-01-07
Title: C, Author: John Doe, Age: 42, Published: 1990-01-07
Title: Java, Author: John Doe, Age: 50, Published: 1998-01-01
Title: Python, Author: John Doe, Age: 62, Published: 2010-12-02
Title: Scala, Author: John Doe, Age: 67, Published: 2015-03-02
Title: HTML, Author: Mark Pickford, Age: 58, Published: 2008-07-02
Title: Javascript, Author: James Pickford, Age: 52, Published: 2008-03-02
Title: CSS, Author: Mark Keely, Age: 51, Published: 2008-04-02
Title: ADA, Author: Ada Lovelace, Age: 44, Published: 1988-01-02
Title: Dart, Author: Peter Dove, Age: 47, Published: 2020-01-02
Title: Lisp, Author: William Fence, Age: 25, Published: 1977-09-24
Title: Haskell, Author: William Fence, Age: 30, Published: 1982-04-14
Title: C++, Author: William Fence, Age: 38, Published: 1990-04-02
Title: C, Author: William Fence, Age: 38, Published: 1990-04-14
Title: C#, Author: William Fence, Age: 60, Published: 2012-10-02

Program 5

This program limits the output to authors with three or more books.

defines function

  sufficientBooks() as pure
    -> books as List of Book
    <- rtn as Boolean: length books >= MINIMUM_BOOKS

defines program
  <?-
    This example shows two things, firstly it inlines the dynamic function (not keen on this but it can be done).
    Then it filters the lists that do not have sufficient books (in this case three or more).
    Only then does it order those books on their published date.
  -?>
  ProcessByAuthorWithThreeOrMoreBooks()
    stdout <- Stdout()
    library <- Library()

    cat library
      | sort by comparingAuthor
      | group by () is Function of (Book, AuthorId) as pure function (r:=? t.author().id())
      | filter with sufficientBooks
      | map by orderOnPublishedDate
      | flatten
      > stdout

Full source: streamsAndPipelines_library.ek9 - Download streamsAndPipelines_library.ek9

The pipeline is now long, so it uses the vertical layout, which is easier to read (if longer on the page).

The filter command lets through only a 'List of Book' holding three or more books.

Why EK9 does it this way. You could argue that 'sufficient number of books' is only an 'if statement' or a 'lambda'. EK9's aim is to capture business logic: a function called 'sufficientBooks' makes it explicit.

It prints:

Title: C++, Author: John Doe, Age: 42, Published: 1990-01-07
Title: C, Author: John Doe, Age: 42, Published: 1990-01-07
Title: Java, Author: John Doe, Age: 50, Published: 1998-01-01
Title: Python, Author: John Doe, Age: 62, Published: 2010-12-02
Title: Scala, Author: John Doe, Age: 67, Published: 2015-03-02
Title: Lisp, Author: William Fence, Age: 25, Published: 1977-09-24
Title: Haskell, Author: William Fence, Age: 30, Published: 1982-04-14
Title: C++, Author: William Fence, Age: 38, Published: 1990-04-02
Title: C, Author: William Fence, Age: 38, Published: 1990-04-14
Title: C#, Author: William Fence, Age: 60, Published: 2012-10-02

Program 6

This program also omits the first two books of each author with three or more books.

defines program

  <?-
    Similar to the other examples, but this one just skips the first two books by
    each author (if they have 3 or more books).
    Note, there is a sort of common pattern to these, if an intermediate operation needs
    a function, you can either define a basic function or just create a dynamic function.
  -?>
  ProcessByAuthorWithThreeOrMoreBooksIgnoreFirstTwo()
    stdout <- Stdout()
    library <- Library()

    authorId <- () is Function of (Book, AuthorId) as pure function (r:=? t.author().id())
    excludingFirstTwoBooks <- () is Function of(List of Book, List of Book) as pure function
      r:=? cat t | skip 2 | collect as List of Book

    cat library
      | sort by comparingAuthor
      | group by authorId
      | filter with sufficientBooks
      | map by orderOnPublishedDate
      | map by excludingFirstTwoBooks
      | flatten
      > stdout

Full source: streamsAndPipelines_library.ek9 - Download streamsAndPipelines_library.ek9

It prints:

Title: Java, Author: John Doe, Age: 50, Published: 1998-01-01
Title: Python, Author: John Doe, Age: 62, Published: 2010-12-02
Title: Scala, Author: John Doe, Age: 67, Published: 2015-03-02
Title: C++, Author: William Fence, Age: 38, Published: 1990-04-02
Title: C, Author: William Fence, Age: 38, Published: 1990-04-14
Title: C#, Author: William Fence, Age: 60, Published: 2012-10-02

The dynamic function excludingFirstTwoBooks skips books per author. It could have been a standard function, or a dynamic function returned by a higher order function. Programs 7 and 8 follow the same theme. Program 7 uses a higher order function to select the 'skip two' filter; Program 8 also shows only the first and only the last book of each author with three or more books.

Reuse

When you need different views, summaries or filters of a set of data, a set of reusable functions like these gives you the reuse. An imperative loop with loop variables and intermediate collections tends not to be reusable, because the logic is nested in the loops and their conditions.

For a single requirement, such as the latest book by author ID=1, a loop is tempting. But software that lives long gets more queries, and a pipeline makes it obvious how to extend and reuse what you have written. A short contrast comes later, and Design note: pipelines over loops makes the argument in full.

Program 7

This program uses an enumeration and a higher order function that returns a function based on the enumeration value.

defines function

  bookFilter() as pure abstract
    -> books as List of Book
    <- filtered as List of Book?

  <?-
    Example of a higher order function and dynamic functions.
    There are other ways to solve the mapping of an enumerations requirement to a function.
    This uses a switch/given expression in a pure context
  -?>
  suitableBookFilter() as pure
    -> selection as BookFilterSelection
    <- rtn as bookFilter: given selection
      <- theFilter as bookFilter?
      when BookFilterSelection.SkipTwo
        theFilter:=? () is bookFilter as pure function (filtered:=? cat books | skip 2 | collect as List of Book)
      when BookFilterSelection.JustFirst
        theFilter:=? () is bookFilter as pure function (filtered:=? cat books | head 1 | collect as List of Book)
      when BookFilterSelection.JustLast
        theFilter:=? () is bookFilter as pure function (filtered:=? cat books | tail 1 | collect as List of Book)
      default
        theFilter:=? () is bookFilter as pure function (filtered:=? books)

defines program

  <?-
    Example of using a higher order function to create the book filter.
  -?>
  ProcessByAuthorUsingHigherOrderFunction()
    stdout <- Stdout()
    library <- Library()
    authorId <- () is Function of (Book, AuthorId) as pure function (r:=? t.author().id())

    stdout.println("Omit first two books where author has three or more books")
    cat library
      | sort by comparingAuthor
      | group by authorId
      | filter with sufficientBooks
      | map by orderOnPublishedDate
      | map with suitableBookFilter(BookFilterSelection.SkipTwo)
      | flatten
      > stdout

Full source: streamsAndPipelines_library.ek9 - Download streamsAndPipelines_library.ek9

The higher order function 'switches' on the enumeration to return the right dynamic function. The pipeline code itself could be refactored the same way, into a function.

Program 8

This program is that refactoring: it shows functional composition and the value of strong typing of functions and higher order functions.

defines function

  filterBooksToOutput()
    ->
      library as Library
      filterSelection as BookFilterSelection
      output as StringOutput

    authorId <- () is Function of (Book, AuthorId) as pure function (r:=? t.author().id())

    cat library
      | sort by comparingAuthor
      | group by authorId
      | filter with sufficientBooks
      | map by orderOnPublishedDate
      | map with suitableBookFilter(filterSelection)
      | flatten
      > output

defines program

  <?-
    Next logical step is to extract the above and put it into a function called 'filterBooksToOutput'
    Then it can just be called with appropriate parameters.
  -?>
  ProcessByAuthorUsingParameterisedFunction()
    stdout <- Stdout()
    library <- Library()

    stdout.println("Omit first two books where author has three or more books")
    filterBooksToOutput(library, BookFilterSelection.SkipTwo, stdout)

    stdout.println("First book where author has three or more books")
    filterBooksToOutput(library, BookFilterSelection.JustFirst, stdout)

    stdout.println("Last book where author has three or more books")
    filterBooksToOutput(library, BookFilterSelection.JustLast, stdout)

Full source: streamsAndPipelines_library.ek9 - Download streamsAndPipelines_library.ek9

Simple (enumerated) types and the switch expression let you structure a range of solutions from small reusable functions. With the polymorphism of traits, classes and functions, you can change which filter functions are used and where the results are output.

Abstract functions, simple types and records with a select number of operators make refactoring and functional composition easier.

It prints each selection in turn; the first is the same as Program 7:

Omit first two books where author has three or more books
Title: Java, Author: John Doe, Age: 50, Published: 1998-01-01
Title: Python, Author: John Doe, Age: 62, Published: 2010-12-02
Title: Scala, Author: John Doe, Age: 67, Published: 2015-03-02
Title: C++, Author: William Fence, Age: 38, Published: 1990-04-02
Title: C, Author: William Fence, Age: 38, Published: 1990-04-14
Title: C#, Author: William Fence, Age: 60, Published: 2012-10-02
First book where author has three or more books
Title: C++, Author: John Doe, Age: 42, Published: 1990-01-07
Title: Lisp, Author: William Fence, Age: 25, Published: 1977-09-24
Last book where author has three or more books
Title: Scala, Author: John Doe, Age: 67, Published: 2015-03-02
Title: C#, Author: William Fence, Age: 60, Published: 2012-10-02

Program 9

This program tees parts of the pipeline off and saves them for later. It also removes items duplicated by a key (here the publication date).

defines function

  bookSigningEvent() as pure
    -> book as Book
    <- rtn as String: `Date: ${book.published()}`

defines program

  UniquePublishingDatesFromAuthorWithThreeOrMoreBooks()
    stdout <- Stdout()
    library <- Library()
    authorId <- () is Function of (Book, AuthorId) as pure function (r:=? t.author().id())

    authorsBooks as List of Book: List()

    booksByEachAuthor as List of List of Book: List()

    stdout.println("Unique signing events on day of publication by authors with three or more books.")

    cat library
      | sort by comparingAuthor
      | group by authorId
      | tee in booksByEachAuthor
      | filter by sufficientBooks
      | map by orderOnPublishedDate
      | flatten
      | tee in authorsBooks
      | uniq by dateBookPublished
      | sort by compareDatePublished
      | map with bookSigningEvent
      > stdout

    stdout.println("Books From author with three or more books")
    cat authorsBooks > stdout

    stdout.println(`There are ${length booksByEachAuthor} authors in total with any number of books`)

Full source: streamsAndPipelines_library.ek9 - Download streamsAndPipelines_library.ek9

It prints:

Unique signing events on day of publication by authors with three or more books.
Date: 1977-09-24
Date: 1982-04-14
Date: 1990-01-07
Date: 1990-04-02
Date: 1990-04-14
Date: 1998-01-01
Date: 2010-12-02
Date: 2012-10-02
Date: 2015-03-02
Books From author with three or more books
Title: C++, Author: John Doe, Age: 42, Published: 1990-01-07
Title: C, Author: John Doe, Age: 42, Published: 1990-01-07
Title: Java, Author: John Doe, Age: 50, Published: 1998-01-01
Title: Python, Author: John Doe, Age: 62, Published: 2010-12-02
Title: Scala, Author: John Doe, Age: 67, Published: 2015-03-02
Title: Lisp, Author: William Fence, Age: 25, Published: 1977-09-24
Title: Haskell, Author: William Fence, Age: 30, Published: 1982-04-14
Title: C++, Author: William Fence, Age: 38, Published: 1990-04-02
Title: C, Author: William Fence, Age: 38, Published: 1990-04-14
Title: C#, Author: William Fence, Age: 60, Published: 2012-10-02
There are 7 authors in total with any number of books

Each tee stores the contents streaming past it in a separate collection. uniq limits the output by a key: the function dateBookPublished makes the publication date the uniqueness key.

Program 10 (The Final Example)

This program uses only authors aged 50 or over (when they wrote the book). It includes each author once, sorts the authors by surname (then first name when surnames match) and outputs their names.

defines function

  bookAuthor() as pure
    -> book as Book
    <- author as Author: book.author()

  acceptableAuthorAge() as pure
    -> author as Author
    <- rtn as Boolean: author.age() >= Age(50)

  authorName() as pure
    -> author as Author
    <- rtn as String: `${author.firstname()} ${author.surname()}`

  comparingAuthorName() as pure
    ->
      author1 as Author
      author2 as Author
    <-
      rtn as Integer: author1.surname() <=> author2.surname()
    if rtn == 0
      rtn: author1.firstname() <=> author2.firstname()

defines program

  LibraryExample()
    stdout <- Stdout()
    library <- Library()

    cat library
      | map with bookAuthor
      | filter by acceptableAuthorAge
      | uniq
      | sort by comparingAuthorName
      | map with authorName
      > stdout

Full source: streamsAndPipelines_library.ek9 - Download streamsAndPipelines_library.ek9

It prints:

John Doe
William Fence
Mark Keely
James Pickford
Mark Pickford

Program 11 (Split)

split divides a stream into groups at a boundary predicate. Unlike group (which groups consecutive items with the same key), split starts a new group each time the predicate returns true, and the triggering item becomes the first item in the new group.

Here the library is sorted by publication date, then split with 'isModernBook'. Every book published on or after 2000-01-01 starts a new group, so the pre-2000 books are one group and each modern book is a group of its own. head 1 then reduces each group to its first book.

defines function

  isModernBook() as pure
    -> book as Book
    <- rtn as Boolean: book.published() >= 2000-01-01

  firstBookOnly() as pure
    -> books as List of Book
    <- rtn as List of Book: cat books | head 1 | collect as List of Book

defines program

  <?-
    Split books by publication date: every modern book (published 2000 or later)
    starts a new group, then take just the first book from each group.
  -?>
  SplitBooksByEra()
    stdout <- Stdout()
    library <- Library()

    stdout.println("First book from each group, a new group starting at each modern book")

    cat library
      | sort by compareDatePublished
      | split with isModernBook
      | map by firstBookOnly
      | flatten
      > stdout

  <?-
    To get exactly two eras, group by era instead: once sorted by date the
    pre-2000 books and the modern books are each one run of equal keys.
  -?>
  GroupBooksByEra()
    stdout <- Stdout()
    library <- Library()

    stdout.println("First book from each era (pre-2000 and 2000+)")

    cat library
      | sort by compareDatePublished
      | group by isModernBook
      | map by firstBookOnly
      | flatten
      > stdout

Full source: streamsAndPipelines_library.ek9 - Download streamsAndPipelines_library.ek9

It prints:

First book from each group, a new group starting at each modern book
Title: Lisp, Author: William Fence, Age: 25, Published: 1977-09-24
Title: Javascript, Author: James Pickford, Age: 52, Published: 2008-03-02
Title: CSS, Author: Mark Keely, Age: 51, Published: 2008-04-02
Title: HTML, Author: Mark Pickford, Age: 58, Published: 2008-07-02
Title: Python, Author: John Doe, Age: 62, Published: 2010-12-02
Title: C#, Author: William Fence, Age: 60, Published: 2012-10-02
Title: Scala, Author: John Doe, Age: 67, Published: 2015-03-02
Title: Dart, Author: Peter Dove, Age: 47, Published: 2020-01-02

To get exactly two eras, 'GroupBooksByEra' (in the same source) uses group by isModernBook instead: once the books are sorted by date, the pre-2000 books and the modern books are each one run of equal keys. It prints:

First book from each era (pre-2000 and 2000+)
Title: Lisp, Author: William Fence, Age: 25, Published: 1977-09-24
Title: Javascript, Author: James Pickford, Age: 52, Published: 2008-03-02

The difference between split and group is the function they take:

group uses a key function (T) → K and groups consecutive items with the same key.
split uses a boundary predicate (T) → Boolean that fires at arbitrary points in the stream. Use it for boundary detection: section headers in text, delimiters in tokenisation, or era transitions as here.

Like group, split produces a stream of List of T. Use flatten to unpack the groups back into individual items, or map to process each group independently (as above). If the boundary marker itself is unwanted, a map function can trim it from each group — the data is preserved, and you decide what to keep.

Summary

Pipelines are the most functional part of EK9. With functions, traits, records, classes and collections, they remove the need for deep, layered loops with stateful variables.

They also tolerate missing data: if a book's published date had been unset, there would still have been no null pointer exception.

Developers from an imperative or Object-Oriented background can take time to get used to this. Many OO languages now have 'chained'/'fluent' APIs, so EK9's clean, simple and composable syntax should be easier to understand and adopt. The syntax follows the Unix/Linux shell 'pipe' because the concept is so similar.

Why EK9 does it this way. The same functionality takes less code in an imperative style, and EK9 supports that style. But small, rounded, documented components are more reliable, reusable and easier to understand over the long term. The design pattern is a standard one, so it is clear from the page what is being done; nested loops take longer to 'grok', and each time similar functionality is needed they are written slightly differently, which leads to confusion and defects. The design note has the full argument.

For contrast, here is a partial procedural/imperative version, without the author name sorting.

defines program
  ImperativeLibraryExample()
    stdout <- Stdout()
    library <- Library()

    uniqueAuthors as Dict of (AuthorId, Author): Dict()

    for book in library
      author <- book.author()
      if author.age() >= SENIOR_AGE
        if uniqueAuthors not contains author.id()
          uniqueAuthors += DictEntry(author.id(), author)

    for author in uniqueAuthors.values()
      stdout.println(`${author.firstname()} ${author.surname()}`)

Full source: streamsAndPipelines_library.ek9 - Download streamsAndPipelines_library.ek9

The uniq step is done with a Dictionary/Map; a List would also work. It prints (the author names are not sorted):

John Doe
Mark Pickford
Mark Keely
William Fence
James Pickford

The code is shorter (without sorting the author names), and for one-off processing it may be the better approach. Once you need to make alterations here and there, the pipeline is easier to change.

Pipeline Complexity

Pipelines have a cognitive cost: at each stage the reader must understand what flows through and how it changes. So each stage adds weighted complexity to the containing function's complexity score (see Code Quality).

Type-preserving operations (filter, reject, sort, head, tail, skip, uniq) cost +1 each: the element type stays the same, so the reader's mental model carries forward.
call and async also cost +1 each.
Type-changing operations (map, group, flatten, split, join, tee) cost +2 each: the element type changes (or a side-channel is introduced), so the reader must work out again what flows downstream.
The source (cat) and the terminal (>, collect as) add +1 each.

Aim for 3–5 stages. At 6–7 stages, consider whether intermediate variables would read better. The weights count towards the function's complexity limit (E11010), so a function with long pipelines and other logic is pushed to split into focused helpers.

Why EK9 does it this way. EK9 applies the same structural quality rules to pipelines as to nested if statements and deeply recursive logic, so that declarative pipelines stay readable, not only correct.

Conclusion

Pipelines give you a standard, flexible set of tools to process collections of Objects clearly. The functional approach reduces the number of methods needed on classes and gives more reuse of code. It takes more thought, and sometimes more code, than an imperative approach, but when you review it, how it works is clear from the page. With EK9 you have the choice.

Next Steps

Advanced class methods is next. It is not about pipelines; it follows on from Classes.