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]

Stages are lazy: nothing runs until a consumer (collect, reduce, count, find, ...) is applied, and each consumer re-runs the whole pipeline.

Types

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

A Stage wraps a Seq(a). In the compiled runtime this would be an actor Pid; here it holds the lazy fold function directly.

Functions

fnallall(stage, pred)#

True if all items satisfy pred; true for an empty stage.

Examples

march> Flow.all(Flow.range(0, 3), fn x -> x < 2)
false
march> Flow.all(Flow.range(0, 0), fn x -> x < 2)
true

fnanyany(stage, pred)#

True if any item satisfies pred; false for an empty stage.

Examples

march> Flow.any(Flow.range(0, 3), fn x -> x == 2)
true

fnbatchbatch(stage, n : {Int | _ > 0})#

Group items into lists of n elements (n must be positive); the last group holds the remainder and may be shorter.

Examples

march> Flow.collect(Flow.batch(Flow.range(0, 5), 2))
[[0, 1], [2, 3], [4]]

fncollectcollect(stage)#

Run the pipeline and collect all items into a List, in order.

Examples

march> 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]

fncountcount(stage)#

Count the total number of items produced.

Examples

march> Flow.count(Flow.range(0, 7))
7

fndropdrop(stage, n)#

Skip the first n items (nothing is skipped if n <= 0).

Examples

march> Flow.collect(Flow.drop(Flow.range(0, 5), 3))
[3, 4]

fneacheach(stage, f)#

Run f on each item for its side effects and return :ok. As with Seq.each, f must currently return an Atom such as :ok.

Flow.each(Flow.from_list(["a", "b"]), fn s -> do
  println(s)
  :ok
end)

fnfilterfilter(stage, pred)#

Keep only items where pred returns true.

Examples

march> Flow.collect(Flow.filter(Flow.range(0, 5), fn x -> x > 2))
[3, 4]

fnfindfind(stage, pred)#

Return the first item satisfying pred as Some, or None. The rest of the pipeline still runs; its items are ignored.

Examples

march> Flow.find(Flow.range(0, 10), fn x -> x * x > 20)
Some(5)
march> Flow.find(Flow.range(0, 3), fn x -> x > 5)
None

fnflat_mapflat_map(stage, f)#

Apply f to each item and flatten the results. f must return a Stage (e.g. Flow.from_list(...)), not a List.

Examples

march> Flow.collect(Flow.flat_map(Flow.from_list([1, 2]), fn x -> Flow.range(0, x)))
[0, 0, 1]

fnfrom_listfrom_list(xs)#

Create a stage from a list of items.

Examples

march> Flow.collect(Flow.from_list([1, 2, 3]))
[1, 2, 3]

fnfrom_seqfrom_seq(seq)#

Create a stage from an existing Seq.

Examples

march> Flow.collect(Flow.from_seq(Seq.from_list([1, 2])))
[1, 2]

fnintointo(stage, start, f)#

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

Examples

march> Flow.into(Flow.from_list(["a", "b"]), "", fn (acc, x) -> acc ++ x)
"ab"

fnmapmap(stage, f)#

Apply f to every item.

Examples

march> Flow.collect(Flow.map(Flow.from_list([1, 2, 3]), fn x -> x * 2))
[2, 4, 6]

fnrangerange(start, stop)#

Create a stage over the integer range [start, stop), counting up by 1. Empty when start >= stop.

Examples

march> Flow.collect(Flow.range(0, 5))
[0, 1, 2, 3, 4]
march> Flow.collect(Flow.range(5, 5))
[]

fnreducereduce(stage, start, f)#

Reduce all items to a single value using f(accumulator, item), starting from start. Returns start for an empty stage.

Examples

march> Flow.reduce(Flow.range(1, 5), 0, fn (acc, x) -> acc + x)
10

fnrunrun(stage)#

Run the pipeline for side effects; discard all items and return :ok.

Note: this passes a ()-returning callback to Seq.each, which the typechecker currently rejects (it expects an Atom), so use each with an :ok-returning callback instead.

fntaketake(stage, n)#

Keep at most the first n items (none if n <= 0). Upstream stages still run to completion; only the extra items are discarded.

Examples

march> Flow.collect(Flow.take(Flow.range(0, 10), 3))
[0, 1, 2]

fnunfoldunfold(seed, next)#

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

next is re-run from seed each time the stage is consumed. If it never returns None the stage is unbounded and no consumer terminates on it (not even after take).

Examples

march> Flow.collect(Flow.unfold(0, fn s -> if s >= 5 do None else Some((s, s + 1)) end))
[0, 1, 2, 3, 4]

fnwith_concurrencywith_concurrency(stage, _n) do stage end#

Parallelism hint: in the compiled runtime, spawn n worker actors for this stage. Currently a no-op everywhere: it returns stage unchanged.

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