`std/iteration/staged`

std/iteration/src/staged.trb

The synchronous driver of a Stage: what Iterate.through(stage) answers.

A stage pushes, an Iterate is pulled, so the driver has two halves and they are not the same code:

collect is fused. The stage wraps the collector's accumulator and the values go straight through, one push per value, nothing in between. This is the path every terminal operation of Iterate ends in.
iterate() needs a queue, because one value pushed in can become none or many coming out while the caller asks for exactly one. The queue holds what one input produced and is drained before the next input is pulled, so it stays as small as the widest stage of the pipeline (Clojure's sequence works the same way).

type Queueing

type Queueing<Item> with Accumulator<Item, Void>

The tail of a pipeline that somebody pulls from: every value the stage pushes lands in a list the driver also holds. The list is in a Shared box because an accumulator is a value - handing it over would hand over a copy, and the driver would never see what arrived.

field box

var box: Shared<List<Item>>

Shared with the driver that reads it back out, one next() at a time.

fn add

var fn add(value: Item)

Appends the value to the queue the driver reads from.

fn finish

fn finish()

Nothing to do: the values are already in the box.