Creation Functions

Functions that create new streams from various data sources.

createReadable

Creates a readable stream from an iterable. Every item in the iterable becomes a chunk in the stream.

function createReadable(s: string): ReadableStream<string>
function createReadable<T>(it: AsyncIterable<T>): ReadableStream<T>
function createReadable<T>(it: Iterable<T>): ReadableStream<T>

Parameter	Type	Description
`iterable`	`string \| Iterable<T> \| AsyncIterable<T>`	Anything iterable: arrays, generators, sets, strings, etc.

Returns: ReadableStream<T> (or ReadableStream<string> for strings)

Since v0.9.0 each input shape has its own overload, so the chunk type is inferred precisely — including for AsyncIterable<T>, which previously resolved to never.

import { createReadable } from "@sgmonda/streamfu"

// From an array
const numbers = createReadable([1, 2, 3, 4, 5])

// From a Set
const unique = createReadable(new Set([1, 2, 3]))

// From a string (each character becomes a chunk)
const chars = createReadable("hello")

// From an async generator
const asyncStream = createReadable(async function* () {
  yield 1
  yield 2
  yield 3
}())

createWritable

Creates a writable stream from either a callback function (for simple sinks) or a config object (for sinks that need cleanup hooks or a configurable buffer).

function createWritable<T>(fn: (chunk: T) => void | Promise<void>): WritableStream<T>
function createWritable<T>(opts: CreateWritableOptions<T>): WritableStream<T>

Function overload

Parameter	Type	Description
`fn`	`(chunk: T) => void \| Promise<void>`	Called for each chunk written to the stream

Returns: WritableStream<T>

import { createWritable } from "@sgmonda/streamfu"

const logStream = createWritable(console.log)

// Use with pipeTo
await someReadable.pipeTo(logStream)

Config object overload

Option	Type	Required	Description
`write`	`(chunk: T) => void \| Promise<void>`	yes	Called for each chunk written to the stream
`close`	`() => void \| Promise<void>`	no	Called once when the writer closes normally (all chunks processed). Use to flush buffered state or release resources
`abort`	`(reason?: unknown) => void \| Promise<void>`	no	Called when the writer is aborted (explicitly or by an upstream error in `pipeTo`). Receives the abort reason
`highWaterMark`	`number`	no	Internal queue size. Defaults to `1`. Higher values let more chunks queue before backpressure kicks in

close and abort are mutually exclusive: a sink sees one of the two, never both. When write() itself throws, the writable enters an errored state and neither hook is invoked — wrap the failing operation in try/catch inside write if you need to react to it.

import { createWritable } from "@sgmonda/streamfu"

let batch: Row[] = []
const dbSink = createWritable<Row>({
  write: async (row) => {
    batch.push(row)
    if (batch.length >= 100) {
      await db.insertMany(batch)
      batch = []
    }
  },
  close: async () => {
    if (batch.length > 0) await db.insertMany(batch)
  },
  abort: () => {
    batch = []
  },
  highWaterMark: 100,
})

range

Generates a stream of numbers in a range.

function range(min: number, max: number, step?: number): ReadableStream<number>

Parameter	Type	Default	Description
`min`	`number`	—	The minimum number (inclusive)
`max`	`number`	—	The maximum number (inclusive)
`step`	`number`	`1`	The step between each number

Returns: ReadableStream<number>

import { list, range } from "@sgmonda/streamfu"

const nums = await list(range(1, 5)) // [1, 2, 3, 4, 5]
const evens = await list(range(0, 10, 2)) // [0, 2, 4, 6, 8, 10]

words

Generates a stream of random strings.

function words(chars: number, length: number): ReadableStream<string>

Parameter	Type	Description
`chars`	`number`	The number of characters in each word
`length`	`number`	The number of words to generate

Returns: ReadableStream<string>

import { list, words } from "@sgmonda/streamfu"

const randomWords = await list(words(5, 3)) // e.g. ["abcde", "fghij", "klmno"]