How do call and async work in EK9 stream pipelines?

← Streams and Pipelines · Ref: Q813

EK9 stream pipelines support call (sequential) and async (concurrent) operators to execute function delegates flowing through the pipeline.

CALL AND ASYNC

The call and async operators execute functions that are items in the stream:

  cat [getSteve, getLimb] | call > collector

The stream must contain functions. call executes each sequentially; async executes them concurrently on separate threads.

FUNCTION REQUIREMENTS

Functions used with call/async must:
1. Be actual function delegates (not integers, strings, or class instances)
2. Take no parameters (zero-argument suppliers)
3. Return a value (the return becomes the next stream element)

COMMON MISTAKES

- Streaming non-function values: cat [1, 2] | call triggers E04040 TYPE_MUST_BE_FUNCTION
- Using void functions: cat [noReturn] | call triggers E07490 FUNCTION_MUST_RETURN_VALUE
- Using functions with parameters: cat [needsArg] | call triggers E06310 REQUIRE_NO_ARGUMENTS
- Passing a function TO call: cat items | call myFunc triggers E07880 FUNCTION_OR_DELEGATE_NOT_REQUIRED

DYNAMIC FUNCTIONS

If a function needs captured data, use a dynamic function:

  supplier <- () is AbstractFn as function (rtn: capturedValue)
  cat [supplier] | call > result

See Q812 for stream head/tail/skip errors. See Q126 for complete stream reference.

Example

defines module qa.streams.call.async.rules

  defines function

    getGreeting()
      <- rtn <- "Hello"

    getWorld()
      <- rtn <- "World"

  defines program

    StreamFunctionDemo()
      stdout <- Stdout()

      // These zero-arg functions return String values.
      // They are suitable for use with call/async in streams:
      //   cat [getGreeting, getWorld] | call > stdout

      greeting <- getGreeting()
      stdout.println(greeting)

      world <- getWorld()
      stdout.println(world)

Coming from another language?

Java: No built-in stream call/async. Use map(Supplier::get) or CompletableFuture.supplyAsync(). Python: map(func, iterable) for sequential, asyncio.gather() for parallel. Rust: .map(|f| f()) in iterators. Go: goroutines with channels. EK9: call and async are native pipeline operators with compile-time function signature validation.

Keywords: parallel, delegate, concurrent, call, execute, stream, supplier, pipeline, function, async