bare-stream

stable

bare-stream — Streaming data for JavaScript.

Mirrors the Node.js stream module.

npm i bare-stream

Usage

const { Readable } = require('bare-stream')

const stream = new Readable({
  read(size) {
    // Push data, then push `null` to signal the end
    this.push('hello')
    this.push('world')
    this.push(null)
  }
})

stream.on('data', (data) => console.log(data))

API

Stream basics

`_destroy(err: Error | null, cb: StreamCallback): void`

Parameters

Parameter	Type	Default	Description
`err`	`Error` \| `null`	—	The error the stream is being destroyed with, or `null` for a clean destroy.
`cb`	`StreamCallback`	—	Called with an error, or `null` on success.

`_open(cb: StreamCallback): void`

Parameters

Parameter	Type	Default	Description
`cb`	`StreamCallback`	—	Called with an error, or `null`, once opening finishes.

`destroy(err?: Error | null): void`

Parameters

Parameter	Type	Default	Description
`err?`	`Error` \| `null`	—	The error to destroy the stream with; omit or pass `null` for a clean destroy.

Stream utilities

`Stream.addAbortSignal<S extends Stream>(signal: AbortSignal, stream: S): S`

Destroy stream when signal aborts, using signal.reason as the destruction error. Returns stream.

Parameters

Parameter	Type	Default	Description
`signal`	`AbortSignal`	—	The `AbortSignal` that destroys `stream` on abort.
`stream`	`S`	—	The stream to destroy when `signal` aborts.

`Stream.duplexPair(opts?: DuplexOptions): [Duplex, Duplex]`

Create a pair of linked Duplex streams. Data written to a is readable from b and vice versa. options are passed to each side.

Parameters

Parameter	Type	Default	Description
`opts?`	`DuplexOptions`	—	Passed to both ends of the pair.

`Stream.finished(stream: Stream, opts: { cleanup?: boolean }, cb: StreamCallback): () => void`

Invoke cb once stream is no longer readable or writable, or has errored. Returns a function that detaches the listeners.

Overloads:

Stream.finished(stream: Stream, opts: { cleanup?: boolean }, cb: StreamCallback): () => void
Stream.finished(stream: Stream, cb: StreamCallback): () => void

Parameters

Parameter	Type	Default	Description
`stream`	`Stream`	—	The stream to wait on.
`opts`	`{ cleanup?: boolean }`	—	Set `cleanup: true` to detach the listeners automatically once `cb` runs.
`cb`	`StreamCallback`	—	Called with an error, or `null`, once `stream` finishes.

`Stream.getStreamError(stream: Stream, opts?: { all?: boolean }): Error | null`

Return the error a stream was destroyed with, or null.

Parameters

Parameter	Type	Default	Description
`stream`	`Stream`	—	The stream to inspect.
`opts?`	`{ all?: boolean }`	—	—

`Stream.isDisturbed(stream: Stream): boolean`

Parameters

Parameter	Type	Default	Description
`stream`	`Stream`	—	The stream to test.

`Stream.isEnded(stream: Stream): boolean`

Parameters

Parameter	Type	Default	Description
`stream`	`Stream`	—	The stream to test.

`Stream.isErrored(stream: Stream): boolean`

Parameters

Parameter	Type	Default	Description
`stream`	`Stream`	—	The stream to test.

`Stream.isFinished(stream: Stream): boolean`

Parameters

Parameter	Type	Default	Description
`stream`	`Stream`	—	The stream to test.

`Stream.isReadable(stream: Stream): boolean`

Parameters

Parameter	Type	Default	Description
`stream`	`Stream`	—	The stream to test.

`Stream.isStream(stream: unknown): stream is Stream`

Parameters

Parameter	Type	Default	Description
`stream`	`unknown`	—	The stream to test.

`Stream.isWritable(stream: Stream): boolean`

Return true if stream is writable.

Parameters

Parameter	Type	Default	Description
`stream`	`Stream`	—	The stream to test.

`Stream.pipeline<S extends Writable>(streams: Pipeline<S>, cb?: StreamCallback): S`

Pipe a series of streams together, propagating errors and cleaning up on completion. streams is a Readable source, zero or more Duplex transforms, and a Writable destination. Returns the destination stream. cb is called when the pipeline finishes or errors.

Overloads:

Stream.pipeline<S extends Writable>(streams: Pipeline<S>, cb?: StreamCallback): S
Stream.pipeline<S extends Writable>(...args: Pipeline<S>): S
Stream.pipeline<S extends Writable>(...args: [...Pipeline<S>, cb: StreamCallback]): S

Parameters

Parameter	Type	Default	Description
`streams`	`Pipeline<S>`	—	A `Readable` source, zero or more `Duplex` transforms, and a `Writable` destination.
`cb?`	`StreamCallback`	—	Called with an error, or `null`, once the pipeline finishes or errors.

Readable streams

`new Readable(opts?: ReadableOptions)`

A readable stream.

Parameters

Parameter	Type	Default	Description
`opts?`	`ReadableOptions`	—	—

`_read(size: number): void`

Parameters

Parameter	Type	Default	Description
`size`	`number`	—	—

`Readable.closed: boolean`

true when the stream is no longer readable.

`Readable.errored: Error | null`

The error the stream was destroyed with, or null.

`pause(): this`

`pipe<S extends Writable>(dest: S, cb?: StreamCallback): S`

Parameters

Parameter	Type	Default	Description
`dest`	`S`	—	The destination stream to write into.
`cb?`	`StreamCallback`	—	Called with an error, or `null` on success.

`push(data: unknown | null, encoding?: BufferEncoding): boolean`

Push data into the stream's internal buffer. If data is a string, it is encoded to a Buffer using encoding, defaulting to 'utf8'. Push null to signal the end of the stream.

Parameters

Parameter	Type	Default	Description
`data`	`unknown` \| `null`	—	Data to add to the buffer, or `null` to end the stream.
`encoding?`	`BufferEncoding`	—	Encoding used to convert a string `data` to a `Buffer`; defaults to `'utf8'`.

`read(): unknown | null`

`Readable.from`

Readable.from(data: unknown | unknown[] | AsyncIterable<unknown>, opts?: ReadableOptions): Readable

Create a readable stream from data, which may be a value, an array of values, or an async iterable.

Parameters

Parameter	Type	Default	Description
`data`	`unknown \| unknown[] \| AsyncIterable<unknown>`	—	A value, array of values, or async iterable to read from.
`opts?`	`ReadableOptions`	—	—

`Readable.fromWeb(readableStream: ReadableStream, opts?: ReadableFromWebOptions): Readable`

Convert a web ReadableStream into a Readable.

Parameters

Parameter	Type	Default	Description
`readableStream`	`ReadableStream`	—	The web `ReadableStream` to convert.
`opts?`	`ReadableFromWebOptions`	—	Options for the conversion; supports `encoding` and `signal`, matching `ReadableOptions`.

`Readable.isBackpressured(rs: Readable): boolean`

Parameters

Parameter	Type	Default	Description
`rs`	`Readable`	—	The stream to check.

`Readable.isPaused(rs: Readable): boolean`

Parameters

Parameter	Type	Default	Description
`rs`	`Readable`	—	The stream to check.

`Readable.toWeb(readable: Readable, opts?: ReadableToWebOptions): ReadableStream`

Convert a Readable into a web ReadableStream.

Parameters

Parameter	Type	Default	Description
`readable`	`Readable`	—	The `Readable` to convert.
`opts?`	`ReadableToWebOptions`	—	Options for the conversion; `strategy` is a custom queuing strategy passed through to the `ReadableStream` constructor.

`resume(): this`

`setEncoding(encoding: BufferEncoding): void`

Parameters

Parameter	Type	Default	Description
`encoding`	`BufferEncoding`	—	Encoding used to decode emitted data to strings.

`unshift(data: unknown | null, encoding?: BufferEncoding): boolean`

Like stream.push() but prepends data to the internal buffer so it is read before any already buffered data.

Parameters

Parameter	Type	Default	Description
`data`	`unknown` \| `null`	—	Data to prepend to the buffer, or `null` to end the stream.
`encoding?`	`BufferEncoding`	—	Encoding used to convert a string `data` to a `Buffer`; defaults to `'utf8'`.

Writable streams

`new Writable(opts?: WritableOptions)`

A writable stream.

Parameters

Parameter	Type	Default	Description
`opts?`	`WritableOptions`	—	—

`_final(cb: StreamCallback): void`

Parameters

Parameter	Type	Default	Description
`cb`	`StreamCallback`	—	Called with an error, or `null` on success.

`_write(data: unknown, encoding: StreamEncoding, cb: StreamCallback): void`

Parameters

Parameter	Type	Default	Description
`data`	`unknown`	—	The chunk to write.
`encoding`	`StreamEncoding`	—	Encoding of `data`, or `'buffer'` if it is not a string.
`cb`	`StreamCallback`	—	Called with an error, or `null` on success.

`_writev(batch: { chunk: unknown; encoding: StreamEncoding }[], cb: StreamCallback): void`

Parameters

Parameter	Type	Default	Description
`batch`	`{ chunk: unknown; encoding: StreamEncoding }[]`	—	Queued chunks to write, each with its `chunk` and `encoding`.
`cb`	`StreamCallback`	—	Called with an error, or `null` on success.

`Writable.closed: boolean`

true when the stream is no longer writable.

`cork(): void`

`end(cb?: StreamCallback): this`

Signal that no more data will be written. If data is provided it is written first. The optional cb is called once the stream has finished.

Overloads:

end(cb?: StreamCallback): this
end(data: unknown, encoding?: BufferEncoding, cb?: StreamCallback): this
end(data: unknown, cb?: StreamCallback): this

Parameters

Parameter	Type	Default	Description
`cb?`	`StreamCallback`	—	Called with an error, or `null`, once the stream has finished.

`Writable.errored: Error | null`

The error the stream was destroyed with, or null.

`uncork(): void`

`Writable.drained(ws: Writable): Promise<boolean>`

Returns a promise that resolves once the stream has drained.

Parameters

Parameter	Type	Default	Description
`ws`	`Writable`	—	The stream to wait on.

`Writable.fromWeb(writableStream: WritableStream, opts?: WritableFromWebOptions): Writable`

Convert a web WritableStream into a Writable.

Parameters

Parameter	Type	Default	Description
`writableStream`	`WritableStream`	—	The web `WritableStream` to convert.
`opts?`	`WritableFromWebOptions`	—	Options for the conversion; supports `signal`, matching `WritableOptions`.

`Writable.isBackpressured(ws: Writable): boolean`

Parameters

Parameter	Type	Default	Description
`ws`	`Writable`	—	The stream to check.

`Writable.toWeb(writable: Writable): WritableStream`

Convert a Writable into a web WritableStream.

Parameters

Parameter	Type	Default	Description
`writable`	`Writable`	—	The `Writable` to convert.

`write(data: unknown, encoding?: BufferEncoding, cb?: StreamCallback): boolean`

Write data to the stream. If data is a string, it is encoded using encoding, defaulting to 'utf8'. Returns false if the stream is backpressured. The optional cb is called once the write has drained.

Overloads:

write(data: unknown, encoding?: BufferEncoding, cb?: StreamCallback): boolean
write(data: unknown, cb?: StreamCallback): boolean

Parameters

Parameter	Type	Default	Description
`data`	`unknown`	—	Data to write. If a string, it is encoded using `encoding`.
`encoding?`	`BufferEncoding`	—	Encoding used to convert a string `data` to a `Buffer`; defaults to `'utf8'`.
`cb?`	`StreamCallback`	—	Called with an error, or `null`, once the write has drained.

Duplex and Transform streams

`new Duplex(opts?: DuplexOptions)`

A stream that is both readable and writable. Accepts the combined options of Readable and Writable.

Parameters

Parameter	Type	Default	Description
`opts?`	`DuplexOptions`	—	—

`Duplex.fromWeb`

Duplex.fromWeb({ readable: ReadableStream, writable: Writable }, opts?: DuplexFromWebOptions): Readable

Convert a pair of web ReadableStream and WritableStream into a Duplex.

Parameters

Parameter	Type	Default	Description
`{ readable: ReadableStream, writable: Writable }`	`any`	—	—
`opts?`	`DuplexFromWebOptions`	—	Options for the conversion; combines the `Readable` and `Writable` conversion options (`encoding`, `signal`).

`Duplex.toWeb(readable: Readable, opts?: ReadableToWebOptions): ReadableStream`

Parameters

Parameter	Type	Default	Description
`readable`	`Readable`	—	—
`opts?`	`ReadableToWebOptions`	—	—

`new Transform(opts?: TransformOptions)`

A duplex stream where output is computed from input.

Parameters

Parameter	Type	Default	Description
`opts?`	`TransformOptions`	—	—

`_flush(cb: StreamCallback): void`

Parameters

Parameter	Type	Default	Description
`cb`	`StreamCallback`	—	Called with an error, or `null`, once flushing finishes.

`_transform(data: unknown, encoding: StreamEncoding, cb: StreamCallback): void`

Parameters

Parameter	Type	Default	Description
`data`	`unknown`	—	The chunk to transform.
`encoding`	`StreamEncoding`	—	Encoding of `data`, or `'buffer'` if it is not a string.
`cb`	`StreamCallback`	—	Called with an error, or `null` on success.

Events

`StreamEvents`

interface StreamEvents extends EventMap {
  close: []
  error: [err: Error]
}

`ReadableEvents`

interface ReadableEvents extends StreamEvents {
  data: [data: unknown]
  end: []
  readable: []
  piping: [dest: Writable]
}

`WritableEvents`

interface WritableEvents extends StreamEvents {
  drain: []
  finish: []
  pipe: [src: Readable]
}

`DuplexEvents`

interface DuplexEvents extends ReadableEvents, WritableEvents {}

`TransformEvents`

interface TransformEvents extends DuplexEvents {}

Options

`StreamOptions`

interface StreamOptions<S extends Stream = Stream> {
  eagerOpen?: boolean
  signal?: AbortSignal
  open?(this: S, cb: StreamCallback): void
  predestroy?(this: S): void
  destroy?(this: S, err: Error | null, cb: StreamCallback): void
}

`ReadableOptions`

interface ReadableOptions<S extends Readable = Readable> extends StreamOptions<S> {
  encoding?: BufferEncoding
  highWaterMark?: number
  read?(this: S, size: number): void
}

`WritableOptions`

interface WritableOptions<S extends Writable = Writable> extends StreamOptions<S> {
  write?(this: S, data: unknown, encoding: StreamEncoding, cb: StreamCallback): void
  writev?(this: S, batch: { chunk: unknown; encoding: StreamEncoding }[], cb: StreamCallback): void
  final?(this: S, cb: StreamCallback): void
}

`DuplexOptions`

interface DuplexOptions<S extends Duplex = Duplex> extends ReadableOptions<S>, WritableOptions<S> {}

`TransformOptions`

interface TransformOptions<S extends Transform = Transform> extends DuplexOptions<S> {
  transform?(this: S, data: unknown, encoding: StreamEncoding, cb: StreamCallback): void
  flush?(this: S, cb: StreamCallback): void
}

`bare-stream/web`

ReadableStreamDefaultReader

`new ReadableStreamDefaultReader(stream: ReadableStream)`

A reader over a ReadableStream, exposing closed, read(), releaseLock(), and cancel([reason]).

Parameters

Parameter	Type	Default	Description
`stream`	`ReadableStream`	—	The `ReadableStream` to read from.

`ReadableStreamDefaultReader.cancel(reason?: unknown): Promise<void>`

Parameters

Parameter	Type	Default	Description
`reason?`	`unknown`	—	Reason for the cancellation, passed to the stream's `destroy()`; defaults to a `TypeError` if omitted.

`ReadableStreamDefaultReader.closed: Promise<void>`

`read(): Promise<{ value: unknown; done: boolean }>`

Returns Promise<{ value: unknown; done: boolean }> — Resolves with the next chunk as { value, done: false }, or { value: undefined, done: true } once the stream ends; rejects with the stream's error if the stream is errored.

`ReadableStreamDefaultReader.releaseLock(): void`

ReadableStreamDefaultController

`new ReadableStreamDefaultController(stream: ReadableStream)`

The controller passed to start and pull, exposing desiredSize, enqueue(data), close(), and error([err]).

Parameters

Parameter	Type	Default	Description
`stream`	`ReadableStream`	—	The `ReadableStream` the controller manages.

`close(): void`

`ReadableStreamDefaultController.desiredSize: number`

`ReadableStreamDefaultController.enqueue(data: unknown): void`

Parameters

Parameter	Type	Default	Description
`data`	`unknown`	—	The chunk to enqueue.

`ReadableStreamDefaultController.error(error?: unknown): void`

Parameters

Parameter	Type	Default	Description
`error?`	`unknown`	—	The error to destroy the stream with.

ReadableStream

`new ReadableStream(underlyingSource?: UnderlyingSource, queuingStrategy?: CustomQueuingStrategy)`

A web readable stream. underlyingSource may provide start, pull, and cancel methods, or be an existing streamx stream to wrap. queuingStrategy defaults to a CountQueuingStrategy.

Parameters

Parameter	Type	Default	Description
`underlyingSource?`	`UnderlyingSource`	—	May provide `start`, `pull`, and `cancel` methods, or be an existing `streamx` stream to wrap.
`queuingStrategy?`	`CustomQueuingStrategy`	—	Defaults to a `CountQueuingStrategy` if omitted.

`ReadableStream.cancel(reason?: unknown): Promise<void>`

Parameters

Parameter	Type	Default	Description
`reason?`	`unknown`	—	Reason for the cancellation, passed to the stream's `destroy()`; defaults to a `TypeError` if omitted.

`getReader(): ReadableStreamDefaultReader`

Acquire a ReadableStreamDefaultReader. Throws if the stream is already locked.

Throws

TypeError — thrown if the stream already has an active reader (locked is true).

`ReadableStream.locked: boolean`

`pipeTo(destination: WritableStream): Promise<void>`

Pipe the stream to a WritableStream, returning a promise that resolves once piping completes.

Parameters

Parameter	Type	Default	Description
`destination`	`WritableStream`	—	The `WritableStream` to pipe into.

`ReadableStream.from(iterable: unknown | unknown[] | AsyncIterable<unknown>): ReadableStream`

Create a ReadableStream from an iterable or async iterable.

Parameters

Parameter	Type	Default	Description
`iterable`	`unknown \| unknown[] \| AsyncIterable<unknown>`	—	A value, array of values, or async iterable to read from.

`tee(): [ReadableStream, ReadableStream]`

Split the stream into two independent ReadableStream branches.

QueuingStrategy

`new QueuingStrategy(opts?: QueuingStrategyOptions)`

Parameters

Parameter	Type	Default	Description
`opts?`	`QueuingStrategyOptions`	—	—

`highWaterMark: number`

`size(chunk: unknown): number`

Parameters

Parameter	Type	Default	Description
`chunk`	`unknown`	—	The chunk to measure.

WritableStreamDefaultWriter

`new WritableStreamDefaultWriter(stream: WritableStream)`

A writer over a WritableStream, exposing desiredSize, closed, ready, write(chunk), releaseLock(), close(), and abort([reason]).

Parameters

Parameter	Type	Default	Description
`stream`	`WritableStream`	—	The `WritableStream` to write to.

`WritableStreamDefaultWriter.abort(reason?: unknown): Promise<void>`

Parameters

Parameter	Type	Default	Description
`reason?`	`unknown`	—	Reason for the abort, passed to the stream's `destroy()`; defaults to a `TypeError` if omitted.

`WritableStreamDefaultWriter.close(): Promise<void>`

Returns Promise<void> — Resolves once the stream has finished closing.

`WritableStreamDefaultWriter.closed: Promise<void>`

`WritableStreamDefaultWriter.desiredSize: number`

`ready: Promise<void>`

`WritableStreamDefaultWriter.releaseLock(): void`

`write(chunk: unknown): Promise<void>`

Parameters

Parameter	Type	Default	Description
`chunk`	`unknown`	—	The chunk to write.

Returns Promise<void> — Resolves once chunk has been written and the stream has drained; rejects with the stream's error if the stream is or becomes errored.

WritableStreamDefaultController

`new WritableStreamDefaultController(stream: WritableStream)`

The controller passed to start and write, exposing error([err]).

Parameters

Parameter	Type	Default	Description
`stream`	`WritableStream`	—	The `WritableStream` the controller manages.

`error(err?: unknown): void`

Parameters

Parameter	Type	Default	Description
`err?`	`unknown`	—	The error to destroy the stream with.

WritableStream

`new WritableStream(underlyingSink?: UnderlyingSink, queuingStrategy?: CustomQueuingStrategy)`

A web writable stream. underlyingSink may provide start, write, close, and abort methods, or be an existing streamx stream to wrap.

Parameters

Parameter	Type	Default	Description
`underlyingSink?`	`UnderlyingSink`	—	May provide `start`, `write`, `close`, and `abort` methods, or be an existing `streamx` stream to wrap.
`queuingStrategy?`	`CustomQueuingStrategy`	—	—

`WritableStream.abort(reason?: unknown): Promise<void>`

Parameters

Parameter	Type	Default	Description
`reason?`	`unknown`	—	Reason for the abort, passed to the stream's `destroy()`; defaults to a `TypeError` if omitted.

`WritableStream.close(): Promise<void>`

Returns Promise<void> — Resolves once the stream has finished closing.

`getWriter(): WritableStreamDefaultWriter`

Acquire a WritableStreamDefaultWriter. Throws if the stream is already locked.

Throws

TypeError — thrown if the stream already has an active writer (locked is true).

Parameter	Type	Default	Description
`stream`	`TransformStream`	—	The `TransformStream` the controller manages.

`TransformStreamDefaultController.desiredSize: number`

`TransformStreamDefaultController.enqueue(data: unknown): void`

Parameters

Parameter	Type	Default	Description
`data`	`unknown`	—	The chunk to enqueue.

`TransformStreamDefaultController.error(error?: unknown): void`

Parameters

Parameter	Type	Default	Description
`error?`	`unknown`	—	The error to destroy the stream with.

new TransformStream(transformer?: Transformer, writableStrategy?: CustomQueuingStrategy, readableStrategy?: CustomQueuingStrategy)

A web transform stream. transformer may provide start, transform, and flush methods. Exposes readable and writable properties.

Parameters

Parameter	Type	Default	Description
`transformer?`	`Transformer`	—	May provide `start`, `transform`, and `flush` methods.
`writableStrategy?`	`CustomQueuingStrategy`	—	—
`readableStrategy?`	`CustomQueuingStrategy`	—	—

Parameter	Type	Default	Description
`value`	`unknown`	—	The value to test.

`isReadableStreamErrored(stream: ReadableStream): boolean`

Parameters

Parameter	Type	Default	Description
`stream`	`ReadableStream`	—	The stream to test.

`isReadableStreamDisturbed(stream: ReadableStream): boolean`

Parameters

Parameter	Type	Default	Description
`stream`	`ReadableStream`	—	The stream to test.

`isWritableStream(value: unknown): value is WritableStream`

Parameters

Parameter	Type	Default	Description
`value`	`unknown`	—	The value to test.

`isTransformStream(value: unknown): value is TransformStream`

Return true if value is a web TransformStream.

Parameters

Parameter	Type	Default	Description
`value`	`unknown`	—	The value to test.

Types

`UnderlyingSource`

interface UnderlyingSource<S extends ReadableStream = ReadableStream> {
  start?(this: S, controller: ReadableStreamDefaultController): void
  pull?(this: S, controller: ReadableStreamDefaultController): void
  cancel?(this: S, reason?: unknown): void
}

`CustomQueuingStrategy`

interface CustomQueuingStrategy {
  highWaterMark?: number
  size?: (chunk: unknown) => number
}

`QueuingStrategyOptions`

interface QueuingStrategyOptions {
  highWaterMark?: number
}

`UnderlyingSink`

interface UnderlyingSink<S extends WritableStream = WritableStream> {
  start?(this: S, controller: WritableStreamDefaultController): void
  write?(this: S, chunk: unknown, controller: WritableStreamDefaultController): void
  close?(this: S): void
  abort?(this: S, reason?: unknown): void
}

`Transformer`

interface Transformer<S extends TransformStream = TransformStream> {
  start?(this: S, controller: TransformStreamDefaultController): void
  transform?(this: S, chunk: unknown, controller: TransformStreamDefaultController): void
  flush?(this: S, controller: TransformStreamDefaultController): void
}

Classes

`CountQueuingStrategy`

class CountQueuingStrategy {
}

`ByteLengthQueuingStrategy`

class ByteLengthQueuingStrategy {
}

bare-stream

On this page