Flow

Flow module: backpressure-aware pipeline abstraction.

A Flow pipeline is a chain of stages: producers → transformers → consumers. Demand flows backward (consumers pull from producers), which prevents fast producers from overwhelming slow consumers (backpressure).

DESIGN: In the interpreter, Flow is implemented as a functional pipeline over lazy Seq(a) values (fold-based, church-encoded). The Stage(a) type wraps a Seq(a), and all transformations delegate to the Seq module.

In the compiled runtime, each Stage would be backed by an actor. The consumer sends a Demand(n) message upstream; each stage forwards demand to its source and buffers at most n items before sending them downstream. This is the GenStage/Flow model from Elixir, adapted for March actors.

with_concurrency/2 is a no-op in the interpreter but signals that the stage may execute with n concurrent workers in the compiled runtime.

See stdlib/docs/flow.md for design rationale, usage examples, and a comparison with Elixir's GenStage.

Quick start:

Flow.from_list([1, 2, 3, 4, 5])
  |> Flow.map(fn x -> x * 2)
  |> Flow.filter(fn x -> x > 4)
  |> Flow.collect
-- [6, 8, 10]

Types

typeStageStage(a) = Stage(Seq(a))#

Functions

fnallall(stage, pred)#

True if all items satisfy pred.

fnanyany(stage, pred)#

True if any item satisfies pred.

fnbatchbatch(stage, n)#

Group items into lists of at most n elements.

fncollectcollect(stage)#

Collect all items into a List.

fncountcount(stage)#

Count the total number of items produced.

fndropdrop(stage, n)#

Skip the first n items.

fneacheach(stage, f)#

Run f on each item for its side effects.

fnfilterfilter(stage, pred)#

Keep only items where pred returns true.

fnfindfind(stage, pred)#

Return the first item satisfying pred, or None.

fnflat_mapflat_map(stage, f)#

Apply f to each item (f returns a list) and flatten the results.

fnfrom_listfrom_list(xs)#

Create a stage from a list of items.

fnfrom_seqfrom_seq(seq)#

Create a stage from an existing Seq.

fnintointo(stage, start, f)#

Consume with f(acc, item) -> acc (alias for reduce).

fnmapmap(stage, f)#

Apply f to every item.

fnrangerange(start, stop)#

Create a stage over the integer range [start, stop).

fnreducereduce(stage, start, f)#

Reduce all items to a single value using f(accumulator, item) -> accumulator.

fnrunrun(stage)#

Run the pipeline for side effects; discard all items.

fntaketake(stage, n)#

Keep at most n items.

fnunfoldunfold(seed, next)#

Create a stage by unfolding a seed value. next : seed -> Option((item, new_seed)) Produces items until next returns None.

Example — integers from 0 to 4:
    Flow.unfold(0, fn s -> if s >= 5 do None else Some((s, s + 1))) end

fnwith_concurrencywith_concurrency(stage, _n) do stage end#

Parallelism hint: in the compiled runtime, spawn n worker actors for this stage. No-op in the interpreter (single-threaded execution).

    Flow.from_list(urls)
      |> Flow.map(fn url -> Http.get(url))
      |> Flow.with_concurrency(8)
      |> Flow.collect