std/stream

Die generierte Referenz von std/stream zeigt jede öffentliche Deklaration mit Signatur, Doc-Kommentar und Beispielen.

"Stream" is the word for a flow in one direction, not a type: what a signature names is one of its two ends, Source<Item, Failure> to read from or Sink<Item, Failure> to write into. They are the asynchronous siblings of std/iteration's Iterator and Accumulator and carry the same verbs. Source, Sink and Bytes are in the prelude; Utf8Error, lines, textOf, Reader and FramingError are imported from here.

Import#

use Source, Sink, Bytes, Utf8Error, lines, textOf, Reader, FramingError from "std/stream"

fn totalLength(channel: Channel<String>): Task<Int> {
  var reading = channel.source()
  var total = 0
  // The reading end of a channel cannot fail, so every pull is an `Ok`
  while const Ok(Some(word)) = reading.next().await() {
    total = total + word.byteLength()
  }
  total
}

Declarations#

Source#

shared trait Source<Item, Failure> with Close {
  var fn next(): Task<Result<Item?, Failure>>
  var fn through<Output>(stage: Stage<Item, Output>): Source<Output, Failure>
  var fn map<Output>(transform: Transform<Item, Output>): Source<Output, Failure>
  var fn filter(predicate: Predicate<Item>): Source<Item, Failure>
  var fn filterMap<Output>(transform: Transform<Item, Output?>): Source<Output, Failure>
  var fn mapWhile<Output>(transform: Transform<Item, Output?>): Source<Output, Failure>
  var fn take(amount: Int): Source<Item, Failure>
  var fn takeWhile(predicate: Predicate<Item>): Source<Item, Failure>
  var fn skip(amount: Int): Source<Item, Failure>
  var fn indexed(): Source<(index: Int, item: Item), Failure>
  var fn chunked(size: Int): Source<List<Item>, Failure>
  var fn then<Output>(step: (value: Item) => Task<Result<Output, Failure>>): Source<Output, Failure>
  var fn mapFailure<Other>(transform: (failure: Failure) => Other): Source<Item, Other>
  var fn collect<Output>(into: Accumulator<Item, Output>): Task<Result<Output, Failure>>
  var fn toList(): Task<Result<List<Item>, Failure>>
  var fn count(): Task<Result<Int, Failure>>
  var fn fold<State>(initial: State, combine: (State, Item) => State): Task<Result<State, Failure>>
  var fn forEach(action: Action<Item>): Task<Result<Void, Failure>>
  var fn find(predicate: Predicate<Item>): Task<Result<Item?, Failure>>
  var fn into(var sink: Sink<Item, Failure>): Task<Result<Void, Failure>>
  static fn from(items: Iterate<Item>): Source<Item, Failure>
  static fn pulling(step: () => Task<Result<Item?, Failure>>): Source<Item, Failure>
  static fn produce(capacity: Int = 0, body: (var sink: Sink<Item, Failure>) => Task<Result<Void, Failure>>): Source<Item, Failure> where Failure: From<ChannelClosed>
  static fn empty(): Source<Item, Failure>
}
extend<Item, Problem, Failure: From<Problem>> Source<Result<Item, Problem>, Failure> {
  var fn checked(): Source<Item, Failure>
}

The reading end of a stream. next() answers Ok(Some(item)) for the next item, Ok(None) at the end, and Fail(problem) for a failure that ends the stream for good. A source has an identity and is consumed once, so it is a shared type, and reading from it needs no var binding: a change of an object is not a question of the path it is reached through (see Shared types), so every holder reads from the one stream. Wrapping (map, filter, through) hands the source to a wrapper that pulls from it afterwards, and a chain reads as one expression. Backpressure is the pull: nothing is read before somebody asks for it. checked() (an extension for Source<Result<Item, Problem>, Failure>) turns Result items into the stream's own failure, ending the stream there. produce is the one way to write a producer without generators: body runs as a task of its own and writes into a Channel (see std/task), with capacity: 0 handing every item over directly, in lock-step with the consumer.

The stages read exactly like Iterate's: filterMap transforms and drops the items that answer None, mapWhile transforms until one does and ends the stream there, take/takeWhile/skip bound how much is read, indexed pairs every item with its position, and chunked groups items into lists of at most size, the last group being whatever is left. The terminal operations beyond toList are count (how many items arrived), fold (every item combined into a running state, left to right), forEach (an action run on every item as it arrives), and find (stops at the first match and leaves the rest of the stream unread).

Sink#

shared trait Sink<Item, Failure> with Close {
  var fn add(item: Item): Task<Result<Void, Failure>>
  var fn end(): Task<Result<Void, Failure>>
  var fn addAll(items: Iterate<Item>): Task<Result<Void, Failure>>
  var fn fill(var source: Source<Item, Failure>): Task<Result<Void, Failure>>
  var fn mapFailure<Other>(transform: (failure: Failure) => Other): Sink<Item, Other>
  var fn buffered(capacity: Int = 64): Buffered<Item, Failure>
  static fn pushing(accept: (item: Item) => Task<Result<Void, Failure>>, complete: () => Task<Result<Void, Failure>>): Sink<Item, Failure>
  static fn discarding(): Sink<Item, Failure>
}
extend<Failure> Sink<Bytes, Failure> {
  var fn addText(text: String): Task<Result<Void, Failure>>
  var fn addLine(text: String = ""): Task<Result<Void, Failure>>
}

The writing end, the asynchronous sibling of Accumulator. Backpressure is the await on add: the task finishes once the target has taken the item, so a writer faster than its target waits by itself. end() is the graceful end

• everything buffered is written, and whatever went wrong is reported here at the latest; close() (from Close) is the abrupt end the last release runs - no program calls it - cannot fail, and may leave less written than end() would have. buffered(capacity:) answers a Buffered, whose own flush() is where "when was it actually written" gets an answer; end() flushes too, but close() does not.

Failure is the stream's own failure and nothing else. A read or a write can be cancelled like every wait, but a cancellation is not a failure a stream answers: await() passes it on to the task that waits (see std/task), so every pull and every write is one .await()?, and a stream that cannot fail says Never.

Bytes, Utf8Error#

type Bytes = Buffer<UInt8>
type Utf8Error with Show, Error {
  offset: Int
  fn show(): String
}

Bytes is an alias for Buffer<UInt8>, not a type of its own, so every list operation already works on a chunk - and a chunk belongs to whoever received it, since a list is a value and there is no borrowed buffer to copy before the next await. Utf8Error is bytes that are not UTF-8, with the offset of the byte that broke it; a String is never anything else. The offset is counted from the start of what was decoded, and which that is belongs to the stage that failed: textOf counts from the start of its chunk, lines() from the start of the line, and decodedText() from the start of the whole stream, because a chunk border is nobody's choice and an offset inside one would name a byte the caller cannot point at.

lines, decodedText, textOf, encodedText#

fn lines(): Stage<Bytes, Result<String, Utf8Error>>
fn decodedText(): Stage<Bytes, Result<String, Utf8Error>>
fn textOf(bytes: Bytes): Result<String, Utf8Error>
fn encodedText(): Stage<String, Bytes>

Bytes to lines, split at \n with a trailing \r dropped so CRLF files read like LF files; a last line without a break is still a line. decodedText cuts at character borders instead, holding back a character split across two chunks. Both are resumable Stages (see std/iteration) and work on any Iterate<Bytes> as well as on a Source<Bytes, Failure> - they need no Task themselves, only reading the Source they sit in front of does. A sequence that is still incomplete when the stream ends is reported as invalid UTF-8, at the offset it starts on - there is no more input coming to complete it. textOf is the short form for a whole chunk that has already arrived; encodedText is the reverse, and needs no decision because a String already is UTF-8.

Sink<Bytes, Failure>.addText, addLine#

An extension of the instantiated trait, so it is there for every byte sink - a file, standard output, a socket, the input of a child process - and nowhere else. addText encodes text as UTF-8 and writes it, the text form of add; addLine calls it with \n appended, the byte a wire uses and not the line break of the operating system running the program.

Reader, FramingError#

shared type Reader<Failure: From<FramingError>> with Close {
  static fn of(var source: Source<Bytes, Failure>): Self
  fn buffered(): Bytes
  var fn bytes(count: Int): Task<Result<Bytes?, Failure>>
  var fn line(limit: Int): Task<Result<Bytes?, Failure>>
  var fn frame(prefix: Int, limit: Int, order: ByteOrder = ByteOrder.BigEndian): Task<Result<Bytes?, Failure>>
  var fn close()
}
type FramingError with Show, Error {
  case Truncated(expected: Int, available: Int)
  case LimitExceeded(limit: Int)
  fn show(): String
}

What a wire protocol reads a socket with: a stage splits a whole stream one way, a Reader reads it item by item, and the protocol decides the shape of the next item from the last one. bytes(count) is the next run of count bytes, line(limit) the bytes up to the next \n (without it, and without one \r in front of it), and frame(prefix, limit, order) a length of prefix bytes - big endian unless order says otherwise - followed by that many bytes, which are the answer. The bytes of one item may arrive in any number of chunks; what arrived and was not read yet stays buffered for the next read, and buffered() answers it - what a protocol hands on when it switches the transport, after a TLS request.

Every read answers Ok(None) where the stream ended cleanly before the item's first byte, and a failure where it ended inside the item (FramingError.Truncated) or the item is longer than its limit (FramingError.LimitExceeded, decided before the buffer grows past it). A read that fails consumes nothing, as a read of std/binary's ByteReader does. The Failure is the protocol's own: the source's failure passes through as it is, and the reader's words arrive through the protocol's From<FramingError>. A frame is Bytes, which a ByteReader reads the fields of. A last line without a line break is a truncated line here, and still a line for lines(): a protocol's line is not complete until its line feed arrived.

use Reader, FramingError from "std/stream"

type WireError {
  case Closed
  case Framing(problem: FramingError)
}

extend WireError with From<FramingError> {
  static fn from(value: FramingError): WireError {
    WireError.Framing value
  }
}

/** The kind and the length of every message up to the clean end: a kind byte, then a frame behind four bytes. */
fn summary(reader: Reader<WireError>): Task<Result<List<String>, WireError>> {
  var reading = reader
  var seen: List<String> = []
  while const Some(kind) = reading.bytes(1).await()? {
    // A kind byte the stream ends behind is a message cut short, and the protocol's own failure
    if const Some(body) = reading.frame(prefix: 4, limit: 65536).await()? {
      seen.append "{kind[0]}: {body.length()} bytes"
    } else {
      return Fail WireError.Closed
    }
  }
  Ok seen
}

What is missing#

Nothing that this page declares is missing from the two back ends. A Source or a Sink held as a trait-typed value runs like a concrete one: through reaches Stage.onto<Final>, a generic member through a trait-typed value, which the witness table holds once per list of arguments the program calls it with (CONCURRENCY gap 11), and the static members the trait gives a body - Source.from, Source.pulling, Source.produce - run as those bodies when they are called through the trait's name (tests/conformance/source-trait-value.trb). Bytes, Utf8Error, textOf, lines() and decodedText() do not depend on Task at all and work wherever a Stage does - over an Iterate<Bytes> in a test, for instance.

Related#

• std/iteration - Iterator, Accumulator and Stage, the synchronous siblings this package reuses.
• std/task - Task and Channel, which every Source/Sink verb rides on.
• std/fs - File, which is both a Source and a Sink.
• The standard library - the other packages.