datastream/fold

Pure terminal reductions over Stream(a).

This module is the fold half of the terminal layer: every function here drives a stream to completion (or until an early-exit decision) and returns a plain value. Side-effecting terminals live in sink.

Building combinators alone has no observable effect — calling one of these functions is what actually pulls elements from the source. Pipelines re-run the source on every terminal call; there is no implicit caching.

Values

all

</>

pub fn all(
  over stream: datastream.Stream(a),
  satisfying predicate: fn(a) -> Bool,
) -> Bool

Return True only when predicate holds for every element.

Returns True on an empty stream and short-circuits on the first False, so all terminates on infinite sources whose first failure is reachable.

any

</>

pub fn any(
  over stream: datastream.Stream(a),
  satisfying predicate: fn(a) -> Bool,
) -> Bool

Return True as soon as predicate holds for some element.

Returns False on an empty stream and short-circuits on the first True, so any terminates on infinite sources whose first match is reachable.

collect_result

</>

pub fn collect_result(
  stream: datastream.Stream(Result(a, e)),
) -> Result(List(a), e)

Walk a Stream(Result(a, e)), returning Ok(values) if every element is Ok, or short-circuiting with the first Error(e).

Order is preserved: Ok values land in the returned list in source order. Pulling stops the moment the first Error is seen, so this is safe to use on infinite streams whose first failure is reachable.

count

</>

pub fn count(stream: datastream.Stream(a)) -> Int

Count the number of elements the stream produces.

Drives the stream to completion without retaining elements. Returns 0 on an empty stream.

drain

</>

pub fn drain(stream: datastream.Stream(a)) -> Nil

Drive the stream to completion and discard every element.

Useful when the pipeline is built purely for traversal (for example to walk a tap-instrumented chain) and the caller does not want to import sink for a no-op consumer.

find

</>

pub fn find(
  over stream: datastream.Stream(a),
  satisfying predicate: fn(a) -> Bool,
) -> option.Option(a)

Return the first element satisfying predicate, or None.

Stops pulling upstream the moment a match is observed.

first

</>

pub fn first(stream: datastream.Stream(a)) -> option.Option(a)

Return the first element produced by the stream, if any.

Pulls at most one element from upstream, closes the unconsumed continuation, and stops. This makes first safe on infinite resource-backed streams. Returns None if the stream is empty.

fold

</>

pub fn fold(
  over stream: datastream.Stream(a),
  from initial: b,
  with step: fn(b, a) -> b,
) -> b

Left-fold over the stream, seeded with initial.

Returns initial unchanged on an empty stream. The accumulator is updated by step(acc, element) in source order.

last

</>

pub fn last(stream: datastream.Stream(a)) -> option.Option(a)

Return the last element produced by the stream, if any.

Drives the stream to completion. Returns None on an empty stream. Does not terminate on an infinite stream.

partition_map

</>

pub fn partition_map(
  over stream: datastream.Stream(a),
  with split: fn(a) -> Result(b, c),
) -> #(List(b), List(c))

Route each element through split and accumulate the two outputs separately, both in source order.

Ok(left) lands in the first list, Error(right) in the second. Drives the stream to completion.

partition_result

</>

pub fn partition_result(
  stream: datastream.Stream(Result(a, e)),
) -> #(List(a), List(e))

Split a Stream(Result(a, e)) into #(oks, errors), both in source order.

Drives the stream to completion: this is a full-traversal reducer, not a short-circuiting one. Use collect_result when you want to stop on the first failure.

product_int

</>

pub fn product_int(stream: datastream.Stream(Int)) -> Int

Multiply a stream of Ints. Returns 1 on the empty stream.

reduce

</>

pub fn reduce(
  over stream: datastream.Stream(a),
  with step: fn(a, a) -> a,
) -> option.Option(a)

Left-fold without an explicit seed: the head element seeds the fold.

Returns None on an empty stream, Some(only) on a one-element stream, and Some(fold_of_tail_seeded_with_head) otherwise.

sum_float

</>

pub fn sum_float(stream: datastream.Stream(Float)) -> Float

Sum a stream of Floats. Returns 0.0 on the empty stream.

sum_int

</>

pub fn sum_int(stream: datastream.Stream(Int)) -> Int

Sum a stream of Ints. Returns 0 on the empty stream.

to_bit_array

</>

pub fn to_bit_array(
  stream: datastream.Stream(BitArray),
) -> BitArray

Materialise a Stream(BitArray) into a single concatenated BitArray.

Internally accumulates into a BytesTree, which bridges to Erlang’s iolist_to_binary on the BEAM and concatenates incrementally on JavaScript — both avoid the quadratic cost of materialising into a list and concatenating after the fact. Pulls every element; for terminating streams only. Returns <<>> on an empty stream.

to_list

</>

pub fn to_list(stream: datastream.Stream(a)) -> List(a)

Materialise the stream into a list, preserving source order.

Pulls every element. On a Stream of n elements this allocates O(n) cons cells. For terminating streams only.

to_string

</>

pub fn to_string(stream: datastream.Stream(String)) -> String

Materialise a Stream(String) into a single concatenated String.

Internally accumulates into a StringTree, so concatenation is O(total length) rather than the O(n²) shape that to_list |> string.concat produces on long streams. Pulls every element; for terminating streams only. Returns "" on an empty stream.

to_string_join

</>

pub fn to_string_join(
  stream stream: datastream.Stream(String),
  with separator: String,
) -> String

Materialise a Stream(String) into a single String with separator inserted between consecutive elements (but not before the first or after the last). Streaming counterpart of gleam/string.join/2.

Internally accumulates into a StringTree, so the cost stays O(total length) even for long streams — no intermediate list materialisation. Returns "" on an empty stream and the lone element on a single-element stream (no separator).

Use when emitting CSV / NDJSON / log lines / single-line digests that need a fixed delimiter between elements without breaking the laziness of the upstream pipeline. (#213)

to_string_tree

</>

pub fn to_string_tree(
  stream: datastream.Stream(String),
) -> string_tree.StringTree

Materialise a Stream(String) into a StringTree, preserving source order without flattening.

Useful when the caller wants to keep building the tree — prepending a header, joining with another tree — before producing the final String. Returns the empty tree on an empty stream.

to_string_tree_join

</>

pub fn to_string_tree_join(
  stream stream: datastream.Stream(String),
  with separator: String,
) -> string_tree.StringTree

StringTree variant of to_string_join for callers that want to keep building the tree — prepending a header, joining with another tree — before producing the final String. Same separator semantics as to_string_join. (#213)