zena:stream
zena
import {…} from 'zena:stream';
Classes
Stream
zena
class Stream<T>
The readable end.
Both operations are exclusive: one read and one write may be
pending at a time — the ends are affine in intent, and until move
checking enforces that, a concurrent operation throws. The read
buffer belongs to the caller: a read overwrites a prefix and
reports the count, and only [0, count) is meaningful.
Methods
zena
read(buf: FixedArray<T>): Future<i32>
Between 1 and buf.length elements into buf, or 0 at
end-of-stream. Completes the moment any elements exist — awaiting
it still costs the queue hop, per the always-async rule.
zena
stop(): void
The reader disclaims the rest of the stream: the pending write
and subsequent writes complete false, and subsequent reads
complete 0. Under cancellation this is what releasing the read
end performs; at a component boundary it is the readable-end
drop.
StreamWriter
zena
class StreamWriter<T>
The write capability. Hand out .stream; keep this — and keep it
singular: fan-in from several producers is an explicit multiplexing
adapter over one writer, not shared use of this object.
A write's future is the backpressure: it completes when its
elements are consumed into reads — or with false when the reader
has stopped, which is data, not a failure.
Properties
zena
stream
Methods
zena
write(value: T): Future<boolean>
Delivers value to the reader: true once it is consumed,
false if the reader stopped before consuming it.
zena
writeAll(buf: FixedArray<T>, count: i32 = buf.length): Future<boolean>
Delivers the first count elements of buf (all of it by
default); the buffer belongs to the stream until the future
completes. true = every element was consumed; false = the
reader stopped first, and a prefix may already have been
delivered.
zena
close(): void
Ends the stream: after the pending write drains, reads complete 0.
Multicast
zena
class Multicast<T>
Reads one source stream and delivers every element to each
subscription, under that subscription's policy. Subscribe before
or during run(); a subscription sees elements from the next
batch on — a multicast broadcasts "from now on", never a replay.
Constructors
zena
new(source: Stream<T>, scratch: FixedArray<T>)
Methods
zena
subscribe(policy: Policy): Subscription<T>
A fresh subscription delivering under policy, starting at the
next element the source produces.
zena
async run(): Future<void>
Drives delivery until the source ends, then closes every subscription and returns once their remaining elements drain. Spawn it in the scope that owns the multicast; cancelling that scope tears the whole thing down.
Subscription
zena
class Subscription<T> implements Disposable
A subscriber's handle: the stream to read, and the detach. Dropping
a subscription must detach it — under Waits an abandoned
subscriber would park the pump forever — so this implements
Disposable and belongs in a using. (The design wants a
resource class here; until implicit drop lands, resource-ness
would add Own handles to the API without adding safety, so the
explicit using form ships first.)
Policy
zena
sealed class Policy
What absorbs a subscriber's divergence from the producer.
Conflate absorbs unlimited divergence by dropping intermediates;
Buffered absorbs its capacity by memory; Waits absorbs none
and couples the producer to this subscriber.
Conflate
zena
final class Conflate extends Policy
Waits
zena
final class Waits extends Policy
Buffered
zena
final class Buffered(capacity: i32) extends Policy
Properties
zena
capacity: i32
Functions
buffered
zena
function buffered(n: i32): Policy
Lossless up to n queued elements, then the pump waits.
Variables
conflate
zena
let conflate: Policy
The newest undelivered element wins; the pump never waits.
waits
zena
let waits: Policy
The pump does not advance past an element this subscriber has not read, which holds the producer to the slowest such subscriber.