zena:stream
zena
import {…} from 'zena:stream';
Asynchronous data streaming between producers and consumers.
Provides Stream<T> for reading data and StreamWriter<T> for writing data
with built-in backpressure. Elements move directly from writers to readers,
with writers awaiting until written data has been consumed.
Examples
zena
import { StreamWriter } from 'zena:stream';
let writer = new StreamWriter<String>();
let stream = writer.stream;
writer.write('first chunk');
writer.close();
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
readOne(): Future<Option<T>>
One element, or none at end-of-stream: a read of one that needs
no buffer of the caller's, for a consumer that takes elements one
at a time — an event loop, or the pump that carries a typed
stream across a component boundary. The same exclusivity as
read: one read of either kind may be pending.
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
readAllBytes
zena
async function readAllBytes(stream: Stream<u8>): Future<FixedArray<u8>>
Every byte of stream from here to end-of-stream, in one array:
reads until a read returns 0. A response body is the usual case; a
stream that never ends never resolves. Bytes rather than any T
because a read needs a scratch buffer, and a buffer of T needs a
value of T to fill it with.
readAllText
zena
async function readAllText(stream: Stream<u8>): Future<String>
Every byte of stream to end-of-stream, decoded as text. Zena
strings are WTF-8 views over byte arrays, so nothing is re-encoded:
the bytes are collected and wrapped.
textStream
zena
function textStream(text: String): Stream<u8>
A byte stream carrying text, then end-of-stream: a response or
request body built from a string. The bytes are written by a
background task as the reader takes them, so the stream is handed
out at once and the caller never awaits the write.
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.