gossamer - v10.0.0

gossamer/stream/readable_stream

The source side of the Streams API. Build one from a pull-based source via the Builder, or bridge from a Gleam Yielder / AsyncYielder. Reads happen through a Reader acquired with get_reader.

Types

Builder

opaque </>

The configuration for a ReadableStream.

pub opaque type Builder(a)

PipeOptions

</>

Per-call options for pipe_through and pipe_to.

pub type PipeOptions {
  PipeOptions(
    prevent_abort: Bool,
    prevent_cancel: Bool,
    prevent_close: Bool,
    signal: option.Option(abort_signal.AbortSignal),
  )
}

Constructors

ReadableStream

</>

A JavaScript ReadableStream — a pull-based stream of bytes or objects used as a source for data.

See ReadableStream on MDN.

pub type ReadableStream(a)

Values

build

</> 
pub fn build(
  builder: Builder(a),
) -> Result(ReadableStream(a), stream.StreamLifecycleError)

Creates a ReadableStream from the configured Builder. Returns Errored if the start callback throws synchronously; the thrown value is the variant’s reason.

cancel

</> 
pub fn cancel(
  stream: ReadableStream(a),
  reason reason: r,
) -> promise.Promise(Result(Nil, stream.StreamLifecycleError))

Signals consumer disinterest in the stream. Returns Locked if the stream is locked to a reader, or Errored if the underlying source’s cancel callback throws or returns a rejecting promise; the thrown or rejection value is the reason.

controlled

</> 
pub fn controlled() -> #(
  ReadableStream(a),
  default_controller.DefaultController(a),
)

Creates a ReadableStream together with its DefaultController, letting chunks be enqueued from outside the stream rather than produced on demand by a pull callback. The stream applies no automatic backpressure, so gate enqueues on default_controller.desired_size when the consumer may be slower than the producer.

from_async_yielder

</> 
pub fn from_async_yielder(
  yielder: async_yielder.AsyncYielder(a),
) -> ReadableStream(a)

Creates a ReadableStream from an AsyncYielder. Equivalent to JavaScript’s ReadableStream.from with an async iterable. Panics on Bun — see https://github.com/oven-sh/bun/issues/3700.

from_pull

</> 
pub fn from_pull(
  pull: fn(default_controller.DefaultController(a)) -> b,
) -> ReadableStream(a)

Creates a ReadableStream from only a pull callback — use when chunks are produced on demand. If the callback returns a Promise, the stream waits for it to resolve before pulling again.

from_start

</> 
pub fn from_start(
  start: fn(default_controller.DefaultController(a)) -> b,
) -> Result(ReadableStream(a), stream.StreamLifecycleError)

Creates a ReadableStream from only a start callback — use when all chunks can be enqueued up front. If the callback returns a Promise, the stream waits for it to resolve before any pulls occur. Returns Errored if start throws synchronously.

from_yielder

</> 
pub fn from_yielder(
  yielder: yielder.Yielder(a),
) -> ReadableStream(a)

Creates a ReadableStream from a Yielder. Values are pulled from the yielder as the stream is read. Equivalent to JavaScript’s ReadableStream.from with a synchronous iterable. Panics on Bun — see https://github.com/oven-sh/bun/issues/3700.

get_reader

</> 
pub fn get_reader(
  stream: ReadableStream(a),
) -> Result(reader.Reader(a), stream.StreamLifecycleError)

Acquires a Reader that locks the stream. Returns Locked if the stream is already locked.

is_locked

</> 
pub fn is_locked(stream: ReadableStream(a)) -> Bool

Checks whether the stream is locked to a reader.

new

</> 
pub fn new() -> Builder(a)

Creates a Builder with no underlying-source callbacks set.

pipe_options

</> 
pub fn pipe_options() -> PipeOptions

A PipeOptions with every flag at its default. The three prevent flags default to False; no abort signal is attached.

pipe_through

</> 
pub fn pipe_through(
  stream: ReadableStream(a),
  transform: #(
    ReadableStream(b),
    writable_stream.WritableStream(a),
  ),
  with options: PipeOptions,
) -> Result(ReadableStream(b), stream.StreamLifecycleError)

Pipes the stream through a transform (a writable+readable pair), returning the readable side. Returns Locked if this stream or the writable side of the transform is already locked.

pipe_to

</> 
pub fn pipe_to(
  stream: ReadableStream(a),
  destination: writable_stream.WritableStream(a),
  with options: PipeOptions,
) -> promise.Promise(Result(Nil, stream.StreamLifecycleError))

Pipes the stream to a WritableStream. Returns Locked if either side is already locked, Aborted(reason) if the pipe is cancelled via the AbortSignal on options, or Errored(reason) if either side enters an errored state during the pipe.

read_bytes

</> 
pub fn read_bytes(
  stream: ReadableStream(BitArray),
) -> promise.Promise(
  Result(BitArray, stream.StreamLifecycleError),
)

Consumes a byte stream and returns the assembled bytes as a single BitArray. Returns Locked if the stream is already locked, or Errored if the stream enters an errored state during read.

read_text

</> 
pub fn read_text(
  stream: ReadableStream(BitArray),
) -> promise.Promise(Result(String, stream.StreamLifecycleError))

Consumes a byte stream and decodes the assembled bytes as UTF-8. Invalid bytes are replaced with U+FFFD. Returns Locked if the stream is already locked, or Errored if the stream enters an errored state during read.

set_prevent_abort

</> 
pub fn set_prevent_abort(
  opts: PipeOptions,
  prevent_abort: Bool,
) -> PipeOptions

Sets whether errors in the destination should propagate back to abort the source.

set_prevent_cancel

</> 
pub fn set_prevent_cancel(
  opts: PipeOptions,
  prevent_cancel: Bool,
) -> PipeOptions

Sets whether errors in the source should propagate forward to cancel the destination.

set_prevent_close

</> 
pub fn set_prevent_close(
  opts: PipeOptions,
  prevent_close: Bool,
) -> PipeOptions

Sets whether closing the source should also close the destination.

set_signal

</> 
pub fn set_signal(
  opts: PipeOptions,
  signal: abort_signal.AbortSignal,
) -> PipeOptions

Sets an abort signal that, when triggered, aborts the pipe.

tee

</> 
pub fn tee(
  stream: ReadableStream(a),
) -> Result(
  #(ReadableStream(a), ReadableStream(a)),
  stream.StreamLifecycleError,
)

Splits the stream into two independent streams reading the same data. Returns Locked if the stream is already locked.

to_async_yielder

</> 
pub fn to_async_yielder(
  stream: ReadableStream(a),
) -> Result(
  async_yielder.AsyncYielder(a),
  stream.StreamLifecycleError,
)

Returns an AsyncYielder that reads from the stream. The yielder locks the stream until reading completes. Returns Locked if the stream is already locked. Equivalent to JavaScript’s ReadableStream.values (or stream[Symbol.asyncIterator]()).

with_cancel

</> 
pub fn with_cancel(
  builder: Builder(a),
  cancel: fn(dynamic.Dynamic) -> b,
) -> Builder(a)

Registers the cancel callback that runs when the consumer aborts. Receives the cancellation reason. If the callback returns a Promise, the stream waits for it before considering the cancel complete.

with_pull

</> 
pub fn with_pull(
  builder: Builder(a),
  pull: fn(default_controller.DefaultController(a)) -> b,
) -> Builder(a)

Registers the pull callback that runs whenever the consumer requests more chunks. If the callback returns a Promise, the stream waits for it to resolve before pulling again.

with_queuing_strategy

</> 
pub fn with_queuing_strategy(
  builder: Builder(a),
  strategy: stream.QueuingStrategy,
) -> Builder(a)

Sets the queuing strategy controlling backpressure on the stream’s internal queue. Without this, the stream uses the default strategy (chunk count, high water mark of 1).

with_start

</> 
pub fn with_start(
  builder: Builder(a),
  start: fn(default_controller.DefaultController(a)) -> b,
) -> Builder(a)

Registers the start callback that runs once at construction. Use to enqueue initial chunks or set up state. If the callback returns a Promise, the stream waits for it to resolve before any pulls occur.

✨ Search Document