Sum the prices of all products over 20 using a stream pipeline.

Filter, map to price, then join with an addition function:

  total <- cat products | filter by isExpensive | map with getPrice | join with addFloats | collect as Float

The join operation reduces a stream to a single value via a binary function. Each pipeline stage uses a named pure function. See Q1139 for join basics. See Q235 for stream operations reference.

Example

defines module qa.streams.sumaggregation

  defines class

    Product
      name as String?
      price as Float?

      default private Product() as pure

      Product() as pure
        ->
          name as String
          price as Float
        this.name :=? name
        this.price :=? price

      name() as pure
        <- rtn as String: String(name)

      price() as pure
        <- rtn as Float: Float(price)

      operator $ as pure
        <- rtn as String: `${name} \$${price}`

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

      override operator ? as pure
        <- rtn as Boolean: name? and price?

  defines function

    isExpensive() as pure
      -> product as Product
      <- rtn as Boolean?
      expensiveThreshold <- 20.0
      rtn: product.price() > expensiveThreshold

    getPrice() as pure
      -> product as Product
      <- rtn as Float: product.price()

    addFloats() as pure
      ->
        a as Float
        b as Float
      <- rtn as Float: a + b

  defines program

    StreamSumDemo()
      stdout <- Stdout()

      products <- [
        Product("Widget", 15.0),
        Product("Gadget", 45.0),
        Product("Gizmo", 30.0),
        Product("Trinket", 5.0),
        Product("Device", 60.0)
        ]

      // Filter to expensive products, extract price, sum
      total <- cat products | filter by isExpensive | map with getPrice | join with addFloats | collect as Float
      stdout.println(`Total of expensive items: ${total}`)

      // Count expensive items by collecting as list and measuring length
      expensiveList <- cat products | filter by isExpensive | collect as List of Product
      stdout.println(`Number of expensive items: ${length expensiveList}`)

Common mistakes

E50060 — EK9 uses cat/pipe stream syntax, not Java-style .stream() calls. Reduction uses 'join with' and a binary function. See ek9 -h E50060 for details.

Incorrect:

total <- products.stream().filter(p -> p.price() > 20).sum()

Correct:

total <- cat products | filter by isExpensive | map with getPrice | join with addFloats | collect as Float

E07520 — A function used with 'filter by' must return Boolean; making isExpensive return Float triggers E07520 (must return a Boolean). See ek9 -h E07520 for details.

Incorrect:

      <- rtn as Float?
      rtn: product.price()

Correct:

      <- rtn as Boolean?
      expensiveThreshold <- 20.0
      rtn: product.price() > expensiveThreshold

Other ways to ask this

Write code to filter expensive products and sum their prices in a stream
I have a list of products and need the total price of those costing more than 20
Given a product list, compute the sum of prices above a threshold using a pipeline
In Java I'd use stream().filter().mapToDouble().sum(). Write the EK9 equivalent

Coming from another language?

Java: stream().filter(p -> p.price > 20).mapToDouble(Product::getPrice).sum(). Python: sum(p.price for p in products if p.price > 20). Rust: iter().filter().map().sum(). EK9: cat | filter by | map with | join with | collect as.

Keywords: map, total, reduce, join, price, filter, stream, pipeline, aggregate, sum