Streams

Stream[T] is a shared handle to a bounded FIFO ring buffer. Every copy of the handle — passed as an argument, stored in a field, captured by a spawned block — refers to the same control block, which is what makes it a channel rather than a value.

fn round_trip(): int {
    events: Stream[int] = Stream.new(4)
    events.send(1)
    events.send(2)
    events.close()

    mut total := 0
    for Some(v) := events.recv() {
        total = total + v
    }
    return total
}

for Some(v) := events.recv() is the drain idiom: it loops as long as the pattern matches and stops on the first None.

Constructing

Stream.new takes a capacity, defaulting to 16. The element type comes from an explicit type argument or from the annotation.

fn shapes(): int {
    from_annotation: Stream[int] = Stream.new(4)
    from_type_arg := Stream.new[string](2)
    with_default_capacity := Stream.new[int]()
    return from_annotation.len() + from_type_arg.len() + with_default_capacity.len()
}

Capacity is fixed for the stream’s lifetime and bounds live buffered elements, not the total number that pass through. A capacity of zero or less is normalized to 1.

Choose it from the largest burst you intend to absorb and the size of T:

• a small capacity couples producer progress tightly to consumer progress;
• a larger capacity smooths bursts but keeps more values live;
• no capacity removes the need for a close-and-shutdown protocol;
• capacity does not limit how many handles or producers share the channel.

Sending

There are two sends, and the difference is what happens when the ring is full.

fn try_publish(events: Stream[int], value: int): bool {
    return events.send(value)
}

fn publish(events: Stream[int], value: int): bool {
    return events.send_await(value)
}

Use send when dropping, coalescing, or retrying is a deliberate producer policy — a metrics feed where the newest sample matters more than every sample. Use send_await when every accepted value must be delivered and the producer should slow to the consumer’s pace.

Both return false on a closed stream, which is the canonical “stop producing” signal:

fn produce(out: Stream[int], count: int): int {
    mut sent := 0
    mut i := 0
    for i < count {
        if !out.send_await(i) {
            return sent
        }
        sent = sent + 1
        i = i + 1
    }
    out.close()
    return sent
}

Ignoring the return value of a send is how items silently vanish. Branch on it.

Receiving

recv() is a non-blocking buffer operation returning T?. It removes the oldest buffered item, or gives None when nothing is buffered right now.

fn consume_once(events: Stream[int]): int {
    match events.recv() {
        Some(event) => return event
        None => return -1
    }
}

None on its own does not distinguish “open but empty” from “closed and drained”. Ask is_closed() when that distinction matters:

fn drain_all(events: Stream[int]): int {
    mut total := 0
    mut running := true
    for running {
        match events.recv() {
            Some(event) => total = total + event
            None => {
                if events.is_closed() { running = false }
            }
        }
    }
    return total
}

recv() frees a ring slot, which wakes one producer parked in send_await. The received value moves out of the channel; the vacated slot is never read again.

To wait for data rather than poll for it, use the stream as a select source. Selection blocks until the stream becomes readable or is closed.

fn handle(event: int): void { println("event ${event}") }

fn pump(events: Stream[int], shutdown: Stream[int]): void {
    mut running := true
    for running {
        select {
            event := events => handle(event)
            _ := shutdown => running = false
        }
    }
}

Closing

close() rejects further sends, wakes every parked producer and consumer, and leaves buffered elements drainable.

fn close_then_drain(): int {
    events := Stream.new[int](4)
    events.send(1)
    events.send(2)
    events.close()

    mut total := 0
    for Some(value) := events.recv() {
        total = total + value
    }
    return total
}

Closing twice leaves the stream closed. Producers should treat send_await(...) == false as a request to stop.

A consumer that must not miss a value and must not spin combines the two: poll the buffer, park in a select when it is empty, and stop once the stream is both closed and drained.

fn handle(event: int): void { println("event ${event}") }

fn consume(events: Stream[int]): int {
    mut seen := 0
    for {
        match events.recv() {
            Some(event) => {
                handle(event)
                seen = seen + 1
            }
            None if events.is_closed() => break
            None => {
                select {
                    event := events => {
                        handle(event)
                        seen = seen + 1
                    }
                }
            }
        }
    }
    return seen
}

Inspection

fn describe(events: Stream[int]): string {
    if events.is_closed() && events.is_empty() {
        return "finished"
    }
    return "${events.len()} buffered"
}

These are snapshots. In concurrent code, do not check is_empty() and then act as if a later recv() will observe the same state — keep the decision inside one cooperative consumer loop.

Producer and consumer as separate tasks

The point of the shared handle is crossing a task boundary.

fn produce(out: Stream[int], count: int): void {
    mut i := 1
    for i <= count {
        if !out.send_await(i) {
            return
        }
        i = i + 1
    }
    out.close()
}

fn run(count: int): int {
    events := Stream.new[int](2)
    producer := spawn { produce(events, count) }

    mut total := 0
    mut running := true
    for running {
        match events.recv() {
            Some(v) => total = total + v
            None => {
                if events.is_closed() { running = false }
            }
        }
    }

    producer.await()
    return total
}

Capacity 2 with count well above it means the producer really does park on backpressure. It resumes each time recv() frees a slot.

Multiple producers, one consumer

Handles copy freely, so several children can share one channel. Closing then becomes a coordination question: the last producer to finish should close, or the coordinator should close after joining all of them.

fn produce(out: Stream[int], base: int, count: int): int {
    mut sent := 0
    mut i := 0
    for i < count {
        if !out.send_await(base + i) {
            return sent
        }
        sent = sent + 1
        i = i + 1
    }
    return sent
}

fn collect(producers: int, per_producer: int): int {
    queue := Stream.new[int](8)

    mut workers: []Task[int] = []
    mut p := 0
    for p < producers {
        workers.add(spawn { produce(queue, p * 100, per_producer) })
        p = p + 1
    }

    mut consumed := 0
    mut drained := 0
    for drained < producers * per_producer {
        match queue.recv() {
            Some(v) => {
                consumed = consumed + v
                drained = drained + 1
            }
            None => {}
        }
    }

    for worker in workers {
        worker.await()
    }
    queue.close()
    return consumed
}

The coordinator owns close() here, because no single producer knows it is last.

The selection boundary

A closed, empty stream wakes a blocked select. The selectable adapter must materialize a T for the winning arm, and on that close-only wake it produces a zeroed value. Null-shaped scalar and handle representations are drop-safe, so Stream[int], Stream[string], and Stream[[]byte] are all fine. A value enum whose zero-tag variant owns payload data is not yet a sound selected element type, because a zero tag there is a real variant rather than a null sentinel.

The practical consequence: do not treat “my select arm fired” as proof that a producer sent something. Carry an explicit end marker, use a separate shutdown source, or check is_closed() outside the selection.

Re-checking is_closed() inside the arm is the cheapest version:

fn handle(value: int): void { println("item ${value}") }

fn consume(inbox: Stream[int]): int {
    mut total := 0
    mut running := true
    for running {
        select {
            raw := inbox => {
                if inbox.is_closed() && inbox.is_empty() {
                    running = false
                } else {
                    handle(raw)
                    total = total + raw
                }
            }
        }
    }
    return total
}

A separate shutdown source is clearer when the producer and the stopper are different components, because the arm that fires already tells you which happened:

fn handle(value: int): void { println("item ${value}") }

fn consume(inbox: Stream[int], shutdown: Stream[int]): int {
    mut total := 0
    mut running := true
    for running {
        select {
            value := inbox => {
                handle(value)
                total = total + value
            }
            _ := shutdown => running = false
        }
    }
    return total
}

Lifetime

A Stream[T] handle has no per-copy destructor, on purpose: several copies alias one control block, so dropping one must not free it. The consequence is that abandoning a stream with values still buffered retains those values until the enclosing runtime unit is reclaimed.

Drain before you abandon. close() stops new sends; only recv() transfers each buffered value out so it can be dropped.

The stream handle is for in-process structured tasks. Its selectable handle is runtime state, not a serializable channel identifier.

Protocol checklist

Every stream API should answer these, in its own documentation:

Stream[T] provides transport and backpressure. Ordering across multiple producers, deduplication, retries, acknowledgements, and durable delivery are protocols you layer on top.