How do I use functions in stream pipelines?

← Getting Started · Ref: Q59

Dynamic functions with variable capture are essential for EK9 stream pipelines. They provide context to pipeline stages like 'map with' and 'filter by' without global state or side effects.

BASIC PIPELINE WITH NAMED FUNCTIONS

  cat items | map with transformer | filter by checker | collect as List of String

Each stage uses a function reference. 'map with' applies a transformation function. 'filter by' uses a predicate function.

CAPTURING CONTEXT FOR PIPELINE STAGES

The real power comes from dynamic functions that capture variables for use in pipeline stages:

  names <- Dict() of (Integer, String)
  names += DictEntry(1, "Alice")
  names += DictEntry(2, "Bob")

  idToName <- (names) is idMapper as function
    result: names.getOrDefault(id, "Unknown")

  resolved <- cat [1, 2, 3]
    | map with idToName
    | collect as List of String

The dynamic function 'idToName' captures the 'names' Dict and uses it inside the pipeline. Each pipeline element gets the captured context without global state.

STATEFUL PIPELINE FUNCTIONS

Dynamic functions can hold state across pipeline iterations — useful for accumulation, counting, or multi-element processing:

  count <- 0
  counter <- (count) is someAccumulator as function
    count++
    result:=? count

This is explicitly stated as a design feature: dynamic functions provide real power when building stream pipelines where you want to retain state as part of the pipeline process rather than depending on a single reduce at the end.

INLINE FUNCTIONS IN PIPELINES

Use inline syntax directly in the pipeline:

  results <- cat keys
    | map with (names) is idMapper (result: names.getOrDefault(id, "Unknown"))
    | collect as List of String

This is compact but still nominally typed — the compiler verifies the function matches the pipeline stage requirements.

FLATTEN FOR OPTIONAL RESULTS

When a pipeline stage returns Optional values, use 'flatten' to extract set values and discard unset ones:

  results <- cat keys
    | map with lookupFunction
    | flatten
    | collect as List of String

This eliminates null checking in pipeline code.

In Java, streams use lambdas but capture is hidden and by reference. In Python, generators and comprehensions have no type safety. In Rust, iterators with closures require complex lifetime annotations. EK9 pipelines with dynamic function captures are type-safe, explicit, and maintain quality enforcement.

See Q51 for abstract functions used as pipeline stage types. See Q52 for dynamic functions. See Q53 for variable capture. See Q54 for Consumer patterns in pipeline callbacks. See Q45 for List operations. See Q46 for Dict operations. See Q89 for stream pipeline basics. See Q122 for collect as and custom aggregators. See Q235 for complete stream operations reference.

Example

defines module qa.functionpipeline

  defines function

    nameMapper() as abstract
      -> identifier as Integer
      <- result as String?

  defines program

    PipelineDemo()
      stdout <- Stdout()

      // Set up context data
      names <- Dict() of (Integer, String)
      names += DictEntry(1, "Alice")
      names += DictEntry(2, "Bob")
      names += DictEntry(3, "Charlie")

      // Dynamic function captures Dict for pipeline use
      lookup <- (names) is nameMapper as function
        result: names.getOrDefault(identifier, "Unknown")

      // Pipeline with captured context
      ids <- [1, 2, 4, 3]
      results <- cat ids
        | map with lookup
        | collect as List of String

      for name in results
        stdout.println(`resolved: ${name}`)

      // Direct function call — same function works outside pipelines
      stdout.println(`direct: ${lookup(2)}`)

Common mistakes

E07880 — Using 'call' instead of 'map with' for a function that takes a parameter and returns a value triggers E07830 — stream pipeline type mismatch. Use 'map with' for transformation functions. See ek9 -h E07830 for details.

Incorrect:

| call lookup

Correct:

| map with lookup

E50060 — EK9 Dict does not have a get() method. Use getOrDefault(key, default) for safe access. Triggers E50060 — method not resolved. See ek9 -h Dict for the full API.

Incorrect:

names.get(identifier, "Unknown")

Correct:

names.getOrDefault(identifier, "Unknown")

Coming from another language?

Java: Stream API with lambdas, hidden capture by reference, map/filter/reduce pattern, no explicit capture list, lambda quality not enforced, flatMap for Optional unwrapping. Python: generators, list comprehensions, map/filter built-ins, no type safety, closures capture by reference. JavaScript: Array.map/filter/reduce with arrow functions, no type safety, promise chains, hidden capture. Rust: Iterator trait with map/filter/collect, closures with complex lifetime annotations, move semantics for capture, turbofish syntax for type annotation. Go: no built-in stream pipelines, manual loops, channels for pipeline patterns, verbose. Kotlin: sequences with lambda chains, structural typing, inline functions for performance, no explicit capture. Swift: lazy sequences with closures, structural typing, no explicit capture list in pipeline context. EK9: stream pipelines with cat/map/filter/collect, dynamic functions with EXPLICIT by-value capture provide pipeline context, stateful functions for accumulation, inline syntax for compact stages, flatten for Optional unwrapping, type-safe and quality-enforced.

Keywords: closure, beginner, start, stream, dynamic, context, anonymous, filter, stateful, intro, pipeline, first, function, capture, map, flatten, migrate, collect, cat, stage