API Reference

streamfu provides 53 functions organized in three categories. The canonical, versioned list lives in STABILITY.md.

Rule of thumb: If it returns a ReadableStream, it’s non-consuming. If it returns a Promise, it consumes the stream.

Creation

Functions that create new streams from various data sources.

Function	Description
`createReadable(iterable)`
`createWritable(fn)`
`range(min, max, step?)`
`words(chars, length)`
`iterate(producer)`	producer-driven stream until `null`
`lines(stream)` / `csvLines(stream)`	line splitting (RFC 4180-aware variant)
`fromEvents(target, name, opts?)`	source from DOM/EventTarget events
`fromPromise(p)`	source from a single `Promise<T>`
`fromResponse(r)`	source from a `fetch()` `Response`

Transformations (non-consuming)

These return a new ReadableStream — the original is not consumed.

Function	Description
`map(stream, ...fns)`
`mapAsync(stream, fn, { concurrency, ordered })`	bounded-concurrency async map
`mapSettled(stream, fn, opts?)`	per-chunk `SettledResult` tagging
`catchError(stream, handler)`	stream-level recovery
`filter(stream, fn)`
`flat(stream, depth?)`
`flatMap(stream, fn)`
`slice(stream, start, end?)`
`splice(stream, start, count, ...items)`
`take(stream, n)` / `drop(stream, n)`	array-style aliases of slice
`tap(stream, fn)`	side-effect mid-pipe (chunks unchanged)
`batch(stream, size)`	group consecutive chunks into arrays
`scan(stream, fn, init)`	reduce with intermediate emissions
`unique(stream, keyFn?)` / `uniqueConsecutive(stream, keyFn?)`	Set-based vs O(1) consecutive dedup
`partition(stream, predicate)`	split into `[matching, nonMatching]`
`groupConsecutive(stream, keyFn)`	streaming grouping by key
`ndjson(stream)` / `csv(stream, opts?)`	parsers (NDJSON, RFC 4180)
`delay/throttle/debounce/bufferTime(stream, ms)`	time-based operators
`inspect(stream, label?)` (experimental)	debug-only logger passthrough
`concat(...streams \\| factories \\| promises)`
`zip(...streams)`
`zipLongest(...streams, { fillValue })`	zip padding short streams
`merge(...streams)`	interleave by arrival time
`pipe(stream, ...fns)`
`branch(stream, n, { maxBuffer? })`

Consumers (consuming)

These consume the stream and return a Promise — it cannot be reused afterward. Use branch() first if you need to consume a stream multiple times.

Function	Description
`reduce(stream, fn[, init])`
`list(stream)`
`at(stream, index)`
`some(stream, fn)`
`every(stream, fn)`
`find(stream, fn)`	first chunk matching the predicate
`count(stream)`	number of chunks
`toBuffer(stream)`	concatenate to a single `Uint8Array`
`groupBy(stream, keyFn)`	eager grouping → `Map<K, T[]>`
`forEach(stream, fn)`
`includes(stream, value)`
`indexOf(stream, value)`
`join(stream, separator?)`
`pipeTo(stream, ...fns, sink)`