Streams

Stream[T] is a shared handle to a bounded FIFO channel. Copies of the handle refer to the same control block, which makes it suitable for communication between spawned producers and consumers.

events: Stream[int] = Stream.new[int](16)

The capacity defaults to 16. A non-positive capacity is normalized to 1. Capacity is fixed for the lifetime of the stream. It bounds live buffered elements, not the total number that can pass through over time.

events := Stream.new[Event](64)
producer := spawn {
    for event in load_events() {
        if !events.send_await(event) {
            return
        }
    }
    events.close()
}

Sending

send(value) is non-blocking:

accepted: bool = events.send(42)

It returns false when the stream is full or closed. The value is enqueued and a waiting consumer is woken when it returns true.

send_await(value) applies backpressure:

accepted := events.send_await(42)

It suspends while the channel is full, then returns true once the value is stored. It returns false if the stream is or becomes closed.

Use send when dropping, coalescing, or retrying is an explicit producer policy. Use send_await when every accepted value must be delivered and the producer should slow down with the consumer.

Receiving

recv() is a non-blocking buffer operation:

match events.recv() {
    Some(event) => consume(event)
    None => {}
}

It removes the oldest buffered item. None means that no item is currently buffered; by itself it does not distinguish an open empty stream from a closed drained stream. Check is_closed() when that distinction matters.

To wait for incoming data, use the stream as a select source. Selection blocks until the stream becomes readable or closes.

select {
    event := events => consume(event)
    _ := shutdown => stop()
}

recv() frees one ring slot and wakes a producer parked in send_await. Received values move out of the channel; the dead slot is not read again.

Closing

close() rejects future sends and wakes waiting producers and consumers. Buffered elements remain available:

events.close()

for Some(value) := events.recv() {
    consume(value)
}

Closing more than once leaves the stream closed. Producers should treat send_await(...) == false as a request to stop producing.

Close does not discard buffered values. Consumers can drain them with repeated recv() calls and finish when both is_closed() and recv() == None. Because state can change between calls, keep the decision inside one cooperative consumer loop rather than splitting it across tasks.

State

The current inspection methods are:

These are snapshots in concurrent code. Do not check is_empty() and then assume a later recv() will observe the same state.

Selection boundary

A closed empty stream wakes a blocked select. The current selectable adapter must materialize a T for every winning arm; on that close-only wake it uses a zeroed value. Null-like scalar and handle representations are drop-safe, but a value enum whose zero-tag variant owns payload data is not yet a sound selected element type for this path.

Use a separate shutdown task, a scalar sentinel whose zero value is reserved for closure, or check stream closure outside selection when clean end-of-stream must be distinguished from a real element.

Lifetime

Drain values before abandoning a stream when their destruction matters. The shared handle deliberately has no ordinary per-copy destructor, because several copied handles alias the same channel control block.

That design means abandoning a stream with buffered managed values can retain their allocations until the enclosing runtime unit is reclaimed. Close stops new sends, but draining transfers and eventually drops each buffered value.

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