Node Stream Test

Install

npm install node-stream-test

ESM only, requires Node.js 22 or newer.

Usage

readable

creates test Readable stream, simulating sync/async behaviors (options: MakeReadableOptions) => (readableOptions: ReadableOptions) => (iterable: Iterable<any>) => Readable

type MakeReadableOptions = {
  eager: boolean               // lazy or eager stream behavior
  log?: typeof console.log     // provide debug logger or noop
  delayMs?: number             // simulate async stream behavior
  errorAtStep?: number         // emit 'error' event at certain step
  continueOnError?: boolean    // whether should stream continue on error or break
}

Lazy stream pushes one chunk of data on every read.
Eager stream pushes all chunks in a synchronous loop on read.

delayMs is a delay between read call and actual chunk push.
This simulates asynchronous stream behavior.
If the stream is eager, it will push all chunks in a loop after first delay

import { readable } from 'node-stream-test'

// create test-readable stream
const testReadable = readable({
  log: console.log,            // output debug info to console
  delayMs: 10,                 // delay 10ms
  eager: false                 // eager or lazy stream 
})({
  objectMode: true             // provide Node Readable configuration
})(
  [1, 2, 3, 4, 5]              // provide data to stream
)

// subscribe to test-readable
testReadable
  .on('data', () => {})
  .on('end', () => {})

writable

creates test Writable stream, simulating sync/async behaviors (options: MakeWritableOptions) => (writableOptions: WritableOptions) => (sink: (chunk: any) => void) => Writable

type MakeWritableOptions = {
  log?: typeof console.log,      // provide debug logger or noop
  delayMs?: number               // simulate async
  errorAtStep?: number           // emit 'error' event at certain step
}

delayMs is a delay between write call and passing chunk to a sink.
This simulates long async writes.

import { writable } from 'node-stream-test'

// We have the following stream
declare var stream: ReadableStream

const testWritable = writable({ 
  log: console.log,              // output debug info to console
  delayMs: 10                    // delay 10ms
})({
  objectMode: true               // provide Node Writable configuration
})(
  (chunk) => {}                  // sink, called on every write
)

// pipe the stream into test-writable
stream.pipe(testWritable)
  .on('finish', () => {})

producer

writes chunks to a stream (options: ProducerOptions) => (iterable: Iterable<any>) => (stream: WritableStream) => () => void

type ProducerOptions = {
  log?: typeof console.log,       // provide debug logger or noop
  eager: boolean,                 // eager or lazy producer
  continueOnError?: boolean       // whether should producer continue on error or break
}

eager producer writes chunks in a synchronous loop until highWatermark reached.
lazy producer writes one chunk, then waits for the write callback or drain event.

import { producer } from 'node-stream-test'

// We have the following writable stream
declare var stream: WritableStream

// create a producer
const beginProduce = producer({
  log: console.log,                // output debug info to console
  eager: true                      // eager producer
})(
  [1, 2, 3, 4, 5]                 // data to write
)(
  stream                           // write to this stream
)

// start producing
beginProduce()

pushConsumer

simple on('data') consumer with logging (options: PushConsumerOptions) => (sink: (chunk: any) => void) => (stream: ReadableStream) => () => void

type PushConsumerOptions = {
  log?: typeof console.log,      // provide debug logger or noop
  continueOnError?: boolean      // whether should consumer continue on error or break
}

import { pushConsumer } from 'node-stream-test'

// We have the following stream
declare var stream: ReadableStream

const subscribeConsumer = pushConsumer({
  log: console.log           // output debug info to console
})(
  (chunk: string) => {}      // your callback on every `data` event
)(
  stream // stream to consume
)

// start consuming
subscribeConsumer()

pullConsumer

simple on('readable') consumer with sync/async behavior and logging (options: PullConsumerOptions) => (sink: (chunk: any) => void) => (stream: ReadableStream) => () => void

type PullConsumerOptions = {
  log?: typeof console.log,      // provide debug logger or noop
  eager: boolean,                // eager or lazy behavior
  delayMs?: number,              // simulate async
  readSizeLimit?: number,        // how much data to read on each 'readable' event
  continueOnError?: boolean      // whether should consumer continue on error or break
}

delayMs is a time between readable event and actual read call on stream.

eager consumer calls read in synchronous loop until null returned.
Then waits for the next readable.
lazy consumer reads one chunk, then waits.

import { pullConsumer } from 'node-stream-test'

// We have the following stream
declare var stream: ReadableStream

const subscribeConsumer = pullConsumer({
  log: console.log,                // print debug info to console
  delayMs: 10,                     // delay 10ms
  eager: false,                    // lazy behavior
  readSizeLimit: undefined         // read all available data
})(
  (chunk: string) => {}            // your callback on `read` call, after `readable` event
)(
  stream // stream to consume
)

// start consuming
subscribeConsumer()