Reference

std/stream/reader

std/stream/src/reader.trb

Reader: the items of a wire protocol - a run of bytes, a line, a frame behind its length - read off a byte stream wherever the borders of its chunks fall, and FramingError, why a reader answered no item.

A protocol reads a socket item by item: Postgres a type byte and then a frame, Redis a line and then the run of bytes the line announced, HTTP a head of lines and then a body. The bytes of one item can arrive in many chunks and the chunk that ends it can hold the start of the next, so the reader keeps what arrived and was not read yet, and the next read starts there. A frame is Bytes, which std/binary's ByteReader reads the fields of.

type FramingError

type FramingError with Show, Error

Why a Reader answered no item: the stream ended inside one, or the item is longer than its limit. The words are those of std/binary's ReadError, whose name this one does not take, so a file can import both.

A protocol converts it into its own error through From, which is what Reader asks of its Failure:

type WireError {
  case Framing(problem: FramingError)
}

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

case Truncated

case Truncated(expected: Int, available: Int)

The stream ended after available bytes of a run that needs expected: the bytes of Reader.bytes, the length prefix or the body of Reader.frame, or a line of Reader.line. A line's length is not known before its line feed, so there expected is one more than available - at least the line feed is missing.

case LimitExceeded

case LimitExceeded(limit: Int)

A line or a frame is longer than the limit its read allows, in bytes.

fn show

fn show(): String

One line: what was missing, or which limit an item broke.

type Reader

shared type Reader<Failure: From<FramingError>> with Close

A buffer over a source of bytes that reads a run of count bytes, a line, or a frame behind its length, whatever the borders of the chunks are: what a wire protocol reads a socket with.

Every read answers Ok(Some(item)), Ok(None) where the stream ended cleanly before the item's first byte, or a failure: the source's own, or a FramingError converted into Failure - an end inside an item, or an item longer than its limit. A read that fails consumes nothing, as a read of std/binary's ByteReader does, so Reader.buffered still holds the bytes it saw. What a read did not consume stays buffered for the next one.

The protocol's error is the Failure of both the source and the reader: a socket's failure is converted first (socket.mapFailure(WireError.from)), and WireError: From<FramingError> gives the reader its own words.

Examples

var reader = Reader.of socket.mapFailure(WireError.from)
while const Some(frame) = reader.frame(prefix: 4, limit: 65536).await()? {
  handle ByteReader(frame)
}

Pitfalls

  • A reader pulls from its source when a read needs more bytes, and never ahead of that. Bytes it holds are bytes the source already gave away: a protocol that switches the transport (a TLS upgrade) takes them from Reader.buffered first.

Related

  • lines() - the stage that splits a whole stream into lines, for a reader that wants nothing else.
  • Source - what the reader reads from.

fn of

static fn of(var source: Source<Bytes, Failure>): Self

A reader over source, with nothing buffered.

fn buffered

fn buffered(): Bytes

The bytes that arrived and no read consumed yet, in the order they came. The reader keeps them: this is a copy, for a protocol that hands the connection on - after a TLS request, the transport that replaces the source starts with these.

fn bytes

var fn bytes(count: Int): Task<Result<Bytes?, Failure>>

The next count bytes. Ok(None) where the stream ended before the first of them; a count of 0 answers Ok(Some([])) without waiting.

Panics

  • For a count below 0: a mistake of the caller, not of the bytes.

Errors

  • FramingError.Truncated where the stream ended after fewer than count bytes, converted into Failure.
  • The source's own failure, as it is.

fn line

var fn line(limit: Int): Task<Result<Bytes?, Failure>>

The bytes up to the next \n, without it and without one \r in front of it, so a CR LF protocol reads like an LF one. Ok(None) where the stream ended before the line's first byte.

Errors

  • FramingError.LimitExceeded for a line of more than limit bytes, the line break not counted: found as soon as that many bytes arrived without a line feed, so a peer that never sends one cannot make the buffer grow.
  • FramingError.Truncated where the stream ended inside a line, with expected one more than available.
  • The source's own failure, as it is.

Pitfalls

  • A last line without a line break is truncated here, and a line for lines(): a protocol's line is not complete until its line feed arrived, a file's last line is.

fn frame

var fn frame(prefix: Int, limit: Int, order: ByteOrder = ByteOrder.BigEndian): Task<Result<Bytes?, Failure>>

The next frame: a length of prefix bytes in order, then that many bytes, which are the answer - the prefix is not part of it. Ok(None) where the stream ended before the prefix's first byte. The length counts the bytes after the prefix only; a protocol whose length counts itself reads the prefix with Reader.bytes instead.

Examples

const message = reader.frame(prefix: 4, limit: 1 << 20).await()?

Panics

  • For a prefix below 1 or above 8: a constant of the protocol, never a number from the bytes.

Errors

  • FramingError.LimitExceeded for a length above limit, decided from the prefix alone: a peer cannot make the buffer grow by announcing a long frame. The prefix stays buffered.
  • FramingError.Truncated where the stream ended inside the prefix (expected is prefix) or inside the frame (expected is the length, available the bytes after the prefix).
  • The source's own failure, as it is.

fn close

var fn close()

Nothing of its own: source is a field, and the release of the reader releases it right after this.