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> implements AsyncIterable<T>
The readable end.
T is scoped (owner-futures.md §"Streams"), so a stream may carry
owners. A single write parks its element in the stream's one slot
until a read takes it, and resolves (true, _); if the reader stops
first, the write hands the element back, (false, value), as
tryComplete hands back its loser. So the stream never releases or
drops an element. The bulk operations, read(buf) and writeAll,
copy through arrays and so need a first-class T; a bulk write parks
a slice of its writer's buffer instead of one element.
Both operations are exclusive: one read and one write may be
pending at a time, and 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: a parked write gets
its element back, later writes are refused the same way, a parked
bulk write completes false, and later reads complete 0 or
none. 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. At a
restricted T the stream has one reader, so .stream cannot be read
off the writer; make the pair with StreamWriter.create instead.
A write's future is the backpressure: it completes when its elements are consumed into reads — or refuses them when the reader has stopped, which is data, not a failure.
Properties
zena
stream
Methods
zena
static create<A>(): inline (StreamWriter<A>, Stream<A>)
A writer and its stream's readable end. At a restricted T, such
as a stream of owners, this is how the pair is made: the stream
belongs to its reader, and the writer keeps a reference that only
puts into it.
zena
write(value: T): Future<Result<void, T>>
Delivers value to the reader: (true, _) once it is consumed,
or (false, value), handing it back, 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.
StreamLimitError
zena
final class StreamLimitError extends Error
A stream longer than the most a read of it would take (readAllBytes).
TextStream
zena
final class TextStream extends Stream<u8>
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>, limit: i64 = as i64): 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.
With a limit of zero or more, a stream that runs past that many
bytes is stopped, and the read fails with StreamLimitError, so a
body someone else sends cannot fill memory.
readAllText
zena
async function readAllText(stream: Stream<u8>, limit: i64 = as i64): 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.
bytesStream
zena
function bytesStream(bytes: FixedArray<u8>): Stream<u8>
A byte stream carrying bytes, then end-of-stream: a response or
request body, or a file's new contents, built from bytes in hand.
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.