Getting Started with EK9 · Chapter 12

Streams and Pipelines

Making Decisions and Repeating Work explained that EK9 has no break and no continue, the kind of jump Dijkstra argued against in 1968, and promised a better tool for the jobs they usually do: running a collection of values through a series of small steps, keeping some, changing others, and gathering a result. That tool is the stream.

A stream takes a collection, sends its values one at a time through a pipeline of stages joined by |, and ends by printing them or collecting them into a result. It reads left to right, like a sentence.

A first pipeline

This one takes the numbers one to six, keeps the even ones, doubles them, and prints the result:

#!ek9
defines module book.nums

  defines function
    isEven() as pure
      -> n as Integer
      <- rtn as Boolean: n rem 2 == 0

    doubleIt() as pure
      -> n as Integer
      <- rtn as Integer: n * 2

  defines program
    Numbers()
      stdout <- Stdout()
      for i in 1 ... 6 | filter by isEven | map with doubleIt > stdout

Run it and you get:

4
8
12

Read the last line as a pipeline of four stages:

• for i in 1 ... 6 starts the stream. 1 ... 6 is a range, and the for feeds its numbers into the pipeline one at a time. (A for followed by | is a stream source, not the counting loop you met earlier.)
• | filter by isEven lets a value through only if isEven says yes. 2, 4 and 6 pass; 1, 3 and 5 are dropped. (n rem 2 is the remainder after dividing n by 2.)
• | map with doubleIt replaces each value with a new one, here its double.
• > stdout ends the pipeline by writing what is left to the screen.

isEven and doubleIt are ordinary named functions, like those in Functions. filter and map use them.

A stream can also start with the word cat, which sends out the items of an existing collection one by one. Both cat and the | come from the Unix command line, where small tools are joined into a pipeline, as in cat file | grep error | sort. EK9's streams bring that idea — small pieces that each do one thing, joined end to end — inside the language. Douglas McIlroy devised it for Unix over fifty years ago.¹

The functions a pipeline expects

filter by needs a function that takes a value and answers yes or no: a test. map with needs a function that takes a value and returns a new value: a transformation. Your functions have exactly those shapes: isEven takes an Integer and returns a Boolean; doubleIt takes an Integer and returns an Integer.

EK9 has names for these shapes. A pure test — one value in, a Boolean out — is a Predicate. A pure transformation — one value in, one value out — is a Function. A Consumer takes one value and returns nothing, for a step that does something with each value. You do not write these names: the compiler works out from their shapes that isEven is a Predicate and doubleIt is a Function. Give filter a function of the wrong shape and it will not compile.²

The rule of thumb: filter wants a test, map wants a transformation, and any named function of the right shape will do.

The shopping basket again

The basket from Collections had a list of items, a price list, and some things the shop does not stock. There it was totalled with a for loop and an if. A pipeline says the same thing more directly. The work gets a name, basketTotal, so that it can be tested:

    priceOf() as pure
      -> item as String
      <- rtn <- Money()
      prices <- {"Apples": 0.35#GBP, "Bread": 1.10#GBP, "Milk": 0.85#GBP}
      rtn: prices.getOrDefault(item, Money())

    isAvailable() as pure
      -> price as Money
      <- rtn as Boolean: price?

    basketTotal() as pure
      -> items as List of String
      <- rtn <- Money()
      rtn: cat items | map with priceOf | filter by isAvailable | collect as Money

The pipeline in basketTotal reads like the numbers one:

• cat items streams the shopping list, name by name.
• | map with priceOf turns each name into its price, an unset Money if the shop does not stock it (priceOf uses the unset Money() default from Collections).
• | filter by isAvailable keeps only the prices that are set, dropping the items with no price.
• | collect as Money ends the pipeline by adding up what is left into a single Money total.

collect as Money turns a stream back into one value, as > stdout sent a stream to the screen. Give basketTotal a basket with two unstocked items and it still returns the right total for what the shop can sell.

Small pieces you can trust

A pipeline is built from small, separate functions, and small, separate functions are easy to test. Prove each piece on its own, and you can trust the whole.

So the pieces are tested as in Testing as You Go, with @Test programs in a dev/ directory. One useful pattern is a table of cases — inputs paired with the answers you expect — which lets one test check many examples:

    @Test
    PriceOfTest()
      known <- {"Apples": 0.35#GBP, "Bread": 1.10#GBP, "Milk": 0.85#GBP}
      for entry in known
        assert priceOf(entry.key()) == entry.value()

    @Test
    MissingPriceTest()
      assert not priceOf("Kiwi")?

    @Test
    IsAvailableTest()
      assert isAvailable(priceOf("Bread"))
      assert not isAvailable(priceOf("Mango"))

    @Test
    BasketTotalTest()
      expected <- 2.30#GBP
      assert basketTotal(["Apples", "Bread", "Kiwi", "Milk"]) == expected

Run them with ek9 -t:

ℹ️ Found 4 tests:
  4 assert (unit tests with assertions)

Executing 4 tests...

✅ PASS book.shop.tests::BasketTotalTest [Assert] (6ms)
✅ PASS book.shop.tests::IsAvailableTest [Assert] (5ms)
✅ PASS book.shop.tests::MissingPriceTest [Assert] (5ms)
✅ PASS book.shop.tests::PriceOfTest [Assert] (5ms)

Summary: 4 passed, 0 failed (4 total)
Types: 4 assert
Duration: 23ms

PriceOfTest checks a table of names against their prices in one go. MissingPriceTest shows that an unstocked item comes back unset, not free. IsAvailableTest shows that the filter's test behaves. And BasketTotalTest checks the whole pipeline: Kiwi is not stocked, so a basket of four items totals only what the shop sells. Each stage is checked, and then the way they combine is checked, so the pipeline is something you have shown to work.

Why this replaces break and continue

Think about the number example as a loop written with jumps: a counter, a loop, an if that skips the odd numbers with a continue, perhaps a break to stop early, and somewhere to gather the results. Each of those parts is a known source of bugs: the off-by-one count, the continue that skips one line too many, the break in the wrong loop.

The pipeline has none of them. There is no counter to miscount and no early exit to misplace. "Skip the ones I do not want" is a filter. "Stop after the first few" is a head stage. "Do this to each" is a map. Each is a named stage with one job, and the whole reads in the order it runs. EK9 removed break and continue and put in their place something that does the same work with fewer ways to go wrong, and that you can test piece by piece.

There is more than this chapter shows

This chapter covers the everyday pipeline: start with cat or a for, then filter and map, and finish with collect or >. EK9's streams go further. There are stages to sort values, drop duplicates with uniq, take or skip a few (head, tail, skip), flatten a stream of lists into a plain stream of values, and group, join or split items. A pipeline can be teed, in the Unix sense: a copy of the flow branches off to be logged or counted while the main stream carries on. And collect is the simplest ending; there are richer ways to accumulate a result as the values go by. sort returns in Using Operators; the rest are for the second volume. The stages in this chapter already handle a great deal of real work.

Try it yourself

1. Change the numbers pipeline to keep the odd numbers instead. Write an isOdd, or think about how else you could do it, and which reads more clearly.
2. Add a | head 2 stage after the map in the numbers pipeline, and see how the output changes.
3. Add "Cheese" to a basket you pass to basketTotal, with no price for it, and add a test showing the total is unchanged. That test stays green as long as the pipeline keeps ignoring what the shop does not sell.

Recap

A stream sends a collection through a pipeline of stages joined by |. You start it with a for over a range or with cat, then filter by a test and map with a transformation, and finish by writing to output with > or gathering a result with collect as. The stages run ordinary named functions, which fit by shape — a Predicate for filter, a Function for map — and EK9 works out the fit. A pipeline of small functions is one you can test: check each piece, check how they combine, and trust the whole. Streams are what replace break, continue and the loop with a counter.

The next chapter, Two Ways to Build: Data, Functions and Objects, starts designing with types of your own.

In the EK9 reference: Design note: why EK9 has no break, continue or return, Design note: pipelines over loops, Streams and Pipelines in EK9