NodeStream unstable

Adapters between Node streams and Effect streams, channels, and readables.

This module is the stream boundary for Node APIs. It wraps Readable and Duplex values as Effect Streams and Channels, pipes Effect streams through Node duplex streams, exposes an Effect Stream back to Node as a Readable, and collects readable payloads into strings, array buffers, or Uint8Arrays with optional byte limits.

11 exports Since v4.0.0 Source

Combinators

pipeThroughDuplex

unstable

Since v4.0.0

Pipes an Effect Stream through a Node Duplex, writing the stream's chunks to the duplex and emitting chunks read back from it.

Signature

declare const pipeThroughDuplex: {
  <B = Uint8Array<ArrayBufferLike>, E2 = UnknownError>(options: {
    readonly chunkSize?: number;
    readonly encoding?: BufferEncoding;
    readonly endOnDone?: boolean;
    readonly evaluate: LazyArg<Duplex>;
    readonly onError?: (error: unknown) => E2;
  }): <R, E, A>(self: Stream<A, E, R>) => Stream<B, E2 | E, R>;
  <R, E, A, B = Uint8Array<ArrayBufferLike>, E2 = UnknownError>(self: Stream<A, E, R>, options: {
    readonly chunkSize?: number;
    readonly encoding?: BufferEncoding;
    readonly endOnDone?: boolean;
    readonly evaluate: LazyArg<Duplex>;
    readonly onError?: (error: unknown) => E2;
  }): Stream<B, E | E2, R>;
}

pipeThroughSimple

unstable

Since v4.0.0

Pipes a stream of strings or bytes through a Node Duplex using default options and Cause.UnknownError for stream failures.

Signature

declare const pipeThroughSimple: {
  (duplex: LazyArg<Duplex>): <R, E>(self: Stream<string | Uint8Array<ArrayBufferLike>, E, R>) => Stream<Uint8Array<ArrayBufferLike>, UnknownError | E, R>;
  <R, E>(self: Stream<string | Uint8Array<ArrayBufferLike>, E, R>, duplex: LazyArg<Duplex>): Stream<Uint8Array<ArrayBufferLike>, UnknownError | E, R>;
}

Constructors

fromDuplex

unstable

Since v4.0.0

Creates a Channel over a Node Duplex, writing upstream chunks with backpressure while emitting chunks read from the duplex and optionally ending the writable side when upstream completes.

Signature

declare function fromDuplex<IE, I = Uint8Array<ArrayBufferLike>, O = Uint8Array<ArrayBufferLike>, E = UnknownError>(options: {
  readonly chunkSize?: number;
  readonly encoding?: BufferEncoding;
  readonly endOnDone?: boolean;
  readonly evaluate: LazyArg<Duplex>;
  readonly onError?: (error: unknown) => E;
}): Channel<readonly [O, O], IE | E, void, readonly [I, I], IE>

fromReadable

unstable

Since v4.0.0

Converts a Node readable stream into an Effect Stream, reading chunks with an optional chunk size, mapping stream errors with onError, and destroying the readable on completion unless closeOnDone is false.

Signature

declare function fromReadable<A = Uint8Array<ArrayBufferLike>, E = UnknownError>(options: {
  readonly chunkSize?: number;
  readonly closeOnDone?: boolean;
  readonly evaluate: LazyArg<Readable | ReadableStream>;
  readonly onError?: (error: unknown) => E;
}): Stream<A, E>

fromReadableChannel

unstable

Since v4.0.0

Creates a Channel that pulls chunks from a Node readable stream, mapping errors with onError and destroying the readable on completion unless closeOnDone is false.

Signature

declare function fromReadableChannel<A = Uint8Array<ArrayBufferLike>, E = UnknownError>(options: {
  readonly chunkSize?: number;
  readonly closeOnDone?: boolean;
  readonly evaluate: LazyArg<Readable | ReadableStream>;
  readonly onError?: (error: unknown) => E;
}): Channel<readonly [A, A], E>

Converting

toArrayBuffer

unstable

Since v4.0.0

Consumes a Node readable stream into an ArrayBuffer, failing through onError on stream errors or when maxBytes is exceeded and destroying the stream on interruption or failure. Set destroyOnOverflow to false to drain the remaining data instead of destroying the stream when the byte limit is exceeded.

Signature

declare function toArrayBuffer<E = UnknownError>(readable: LazyArg<Readable | ReadableStream>, options?: {
  readonly destroyOnOverflow?: boolean;
  readonly maxBytes?: Input;
  readonly onError?: (error: unknown) => E;
}): Effect<ArrayBuffer, E>

toReadable

unstable

Since v4.0.0

Converts an Effect Stream into a Node Readable, using the caller's Effect context to run the stream and destroying the readable if the stream fails.

Signature

declare function toReadable<E, R>(stream: Stream<string | Uint8Array<ArrayBufferLike>, E, R>): Effect<Readable, never, R>

toReadableNever

unstable

Since v4.0.0

Converts a service-free Effect Stream into a Node Readable using an empty Effect context.

Signature

declare function toReadableNever<E>(stream: Stream<string | Uint8Array<ArrayBufferLike>, E, never>): Readable

toString

unstable

Since v4.0.0

Consumes a Node readable stream into a string using the selected encoding, failing through onError on stream errors or when maxBytes is exceeded and destroying the stream on interruption or failure. Set destroyOnOverflow to false to drain the remaining data instead of destroying the stream when the byte limit is exceeded.

Signature

declare function toString<E = UnknownError>(readable: LazyArg<Readable | ReadableStream>, options?: {
  readonly destroyOnOverflow?: boolean;
  readonly encoding?: BufferEncoding;
  readonly maxBytes?: Input;
  readonly onError?: (error: unknown) => E;
}): Effect<string, E>

toUint8Array

unstable

Since v4.0.0

Consumes a Node readable stream into a Uint8Array, using the same error mapping and maxBytes handling as toArrayBuffer.

Signature

declare function toUint8Array<E = UnknownError>(readable: LazyArg<Readable | ReadableStream>, options?: {
  readonly destroyOnOverflow?: boolean;
  readonly maxBytes?: Input;
  readonly onError?: (error: unknown) => E;
}): Effect<Uint8Array<ArrayBufferLike>, E>

Errors

MaxBytesError

unstable

Since v4.0.0

Error indicating that a buffered Node stream exceeded its byte limit.

Signature

declare class MaxBytesError extends Error {
  constructor(maxBytes: number);
  readonly _tag: "MaxBytesError";
  readonly maxBytes: number;
}