@alessiofrittoli/stream-reader v2.5.0

Stream Reader 📚

Easly read pushed data from a Stream

The StreamReader class is a utility for reading data from a ReadableStream in a structured and event-driven manner. It extends the EventEmitter class, providing events for stream lifecycle management and error handling.

Table of Contents

• Getting started
• API Reference
  • StreamReader Class API Reference
    • Constructor
    • Properties
    • Methods
    • Static Methods
    • Listening Event
    • Usage
    • Error handling
  • Types API Reference
• Development
  • Install depenendencies
  • Build the source code
  • ESLint
  • Jest
• Contributing
• Security
• Credits

Getting started

Run the following command to start using stream-reader in your projects:

npm i @alessiofrittoli/stream-reader

or using pnpm

pnpm i @alessiofrittoli/stream-reader

API Reference

Importing the library

import { StreamReader } from '@alessiofrittoli/stream-reader'
import type { ... } from '@alessiofrittoli/stream-reader/types'

StreamReader Class API Reference

Constructor

The StreamReader class constructor accepts a ReadableStream argument. You can optionally pass types arguments I and O to define the type of the streamed data being read and the type of the transformed output chunk.

const reader  = new StreamReader<Buffer>( ... )
const reader2 = new StreamReader<Buffer, string>( ... )

type Input    = Buffer
type Output   = Buffer
const stream  = new TransformStream<Input, Output>()
const reader  = new StreamReader( stream.readable ) // type of `StreamReader<Output, Output>`

Properties

Here are listed the StreamReader class instance accessible properties:

Methods

StreamReader.read()

The StreamReader.read() method read the on-demand pushed data from the given stream.

It internally uses the StreamReader.readChunks() method to read the received chunks.

It emits usefull events such as:

• data - Emitted when a chunk is received from the stream and processed by the optional transform function.
• close - Emitted when the stream is closed.
• error - Emitted when an error occurs while reading.

• See Listening Events section for further info.

| Paramenter | Type | Description | | transform | TransformChunk<I, O> | (Optional) A function that transforms each chunk. |

Type: Promise<ReadChunks<O>>

A new Promise with an Array of read and eventually transformed chunks, resolved once the stream is closed.

StreamReader.readChunks()

The StreamReader.readChunks() method read the on-demand pushed data from the given stream.

Type: AsyncGenerator<ReadChunk<I>>

An async iterable object for consuming chunks of data.

StreamReader.cancel()

Deprecated. Use StreamReader.abort() method instead.

StreamReader.abort()

The StreamReader.abort() method it's pretty usefull when stream data reading is no longer needed, regardless of stream writer state.

This method will cancel the reader, release the lock, emit an 'abort' event, and remove data, close and abort event listeners.

It emits the abort event.

• See Listening Events section for further info.

Static Methods

StreamReader.generatorToReadableStream()

The StreamReader.generatorToReadableStream() method is a utiliy function that converts a Generator or AsyncGenerator to a ReadableStream.

Listening Events

The StreamReader class extends the EventEmitter class, providing events for stream lifecycle management and error handling.

data event

const reader = new StreamReader( ... )
reader.on( 'data', chunk => {
  console.log( 'received chunk', chunk )
} )

close event

const reader = new StreamReader( ... )
reader.on( 'close', chunks => {
  console.log( 'chunks', chunks )
} )

error event

const reader = new StreamReader( ... )
reader.on( 'error', error => {
  console.error( error )
} )

abort event

const reader = new StreamReader( ... )
reader.on( 'abort', reason => {
  console.log( 'reading aborted', reason.message )
} )

Usage

In the following examples we reference streamData which is an async function that writes data and closes the stream once finished:

const sleep = ( ms: number ) => new Promise<void>( resolve => setTimeout( resolve, ms ) )

const defaultChunks = [ 'data 1', 'data 2' ]
const erroredChunks = [ 'data 1', new Error( 'Test Error' ), 'data 2' ]
    
const streamData = async (
  { writer, chunks }: {
    writer: WritableStreamDefaultWriter<Buffer>
    chunks?: ( string | Error )[]
  }
) => {
  chunks ||= defaultChunks
  for await ( const chunk of chunks ) {
    if ( chunk instanceof Error ) {
      throw chunk
    }
    await writer.write( Buffer.from( chunk ) )
    await sleep( 50 )
  }
  await writer.close()
  writer.releaseLock()
}

const stream  = new TransformStream<Buffer, Buffer>()
const writer  = stream.writable.getWriter()
const reader  = new StreamReader( stream.readable )

streamData( { writer } )

const chunks = await reader.read()

const response = await fetch( ... )
let resourceSize = 0

if ( response.body ) {
    const reader  = new StreamReader( response.body )
    const decoder = new TextDecoder()
    reader.on( 'data', chunk => {
        const decoded = decoder.decode( chunk, { stream: true } )
        resourceSize += chunk.BYTES_PER_ELEMENT * chunk.length
    } )
    const chunks = await reader.read()
}

const stream  = new TransformStream<Buffer, Buffer>()
const writer  = stream.writable.getWriter()
const reader  = new StreamReader<Buffer, string>( stream.readable, {
  transform( chunk ) {
    return chunk.toString( 'base64url' )
  }
} )

streamData( { writer } )

reader.on( 'data', chunk => {
  console.log( chunk ) // chunk is now a base64url string
} )
const chunks = await reader.read() // `string[]`

const inMemory  = false
const stream    = new TransformStream<Buffer, Buffer>()
const writer    = stream.writable.getWriter()
const reader    = new StreamReader( stream.readable, { inMemory } )

streamData( { writer } )

reader.on( 'data', chunk => {
  console.log( chunk )
} )
const chunks = await reader.read() // empty `[]`

const stream  = new TransformStream<Buffer, Buffer>()
const writer  = stream.writable.getWriter()
const reader  = new StreamReader( stream.readable )

streamData( { writer } )

reader.read()

abortButton.addEventListener( 'click', () => {
  reader.abort( 'Reading no longer needed' )
} )

Error handling

When an error occurs while reading stream data (such as unexpected stream abort), the StreamReader uses an internal error function which handles the thrown Error.

By default, if no listener is attached to the error event, the StreamReader.read() method will re-throw the caught error.

In that case, you need to await and wrap the StreamReader.read() method call in a trycatch block like so:

try {
  const chunks = await reader.read()
} catch ( err ) {
  const error = err as Error
  console.error( 'An error occured', error.message )
}

with error event listener:

reader.read()
reader.on( 'error', error => {
  console.error( 'An error occured', error.message )
} )

Types API Reference

ReadChunk<T>

Represents a chunk of data that can be read, which can either be of type T or a promise that resolves to T.

• Type Parameter:
  • T: The type of the data chunk. Defaults to unknown if not specified.

ReadChunks<T>

Represents an array of ReadChunk objects.

• Type Parameter:
  • T: The type of data contained in each ReadChunk.

TransformChunk<I, O>

A function that transforms a chunk of data.

• Type Parameters:

  • I: The type of the input chunk. Defaults to unknown.
  • O: The type of the output chunk. Defaults to I.

• Parameters:

  • chunk: The chunk of data to be transformed.

• Returns:
  A transformed chunk of data, which can be either a synchronous result or a promise that resolves to the result.

StreamReaderEvents<O>

Defines event types emitted by the StreamReader.

• Type Parameter:

  • O: The type of data being read from the stream and eventually transformed before the event is emitted.

• Event Types:

  • data: Emitted when a chunk of data is read.
    • Parameters: chunk (ReadChunk<O>)
  • close: Emitted when the stream is closed.
    • Parameters: chunks (ReadChunks<O>)
  • error: Emitted when an error occurs during reading.
    • Parameters: error (Error)
  • abort: Emitted when the reading process is aborted.
    • Parameters: reason (AbortError)

Listener<K, O>

A listener function for events emitted by the StreamReader.

• Type Parameters:

  • K: The event type to listen for.
  • O: The type of data being read from the stream.

• Parameters:

  • ...args: The arguments emitted with the event, based on the event type K.

OnDataEventListener<O>

Listener for the data event.

• Type Parameter:

  • O: The type of data being read.

• Parameters:

  • chunk (ReadChunk<O>): The chunk of data that was read.

OnCloseEventListener<O>

Listener for the close event.

• Type Parameter:

  • O: The type of data being read.

• Parameters:

  • chunks (ReadChunks<O>): An array of chunks read before the stream was closed.

OnCancelEventListener

Deprecated. Use OnAbortEventListener type instead.

OnAbortEventListener

Listener for the abort event.

• Parameters:
  • reason (AbortError): Explains the reason for aborting the operation.

OnErrorEventListener

Listener for the error event.

• Parameters:
  • error (Error): The error that occurred during reading.

StreamGenerator<T>

A generator that produces chunks of data asynchronously. It can be either a regular generator or an async generator.

• Type Parameter:

  • T: The type of data produced by the generator.

Development

Install depenendencies

npm install

or using pnpm

pnpm i

Build the source code

Run the following command to test and build code for distribution.

pnpm build

ESLint

warnings / errors check.

pnpm lint

Jest

Run all the defined test suites by running the following:

# Run tests and watch file changes.
pnpm test:watch

# Run tests in a CI environment.
pnpm test:ci

• See package.json file scripts for more info.

Run tests with coverage.

An HTTP server is then started to serve coverage files from ./coverage folder.

⚠️ You may see a blank page the first time you run this command. Simply refresh the browser to see the updates.

test:coverage:serve

Contributing

Contributions are truly welcome!

Please refer to the Contributing Doc for more information on how to start contributing to this project.

Help keep this project up to date with GitHub Sponsor.

Security

If you believe you have found a security vulnerability, we encourage you to responsibly disclose this and NOT open a public issue. We will investigate all legitimate reports. Email security@alessiofrittoli.it to disclose any security vulnerabilities.

Made with ☕