🌊 Keeps only the chunks of a stream that satisfy a predicate.
Syntax
TypeScript
import { filterStream } from '@opentf/std'; filterStream<T>( stream: ReadableStream<T>, fn: (chunk: T, index: number) => boolean | Promise<boolean>, ): ReadableStream<T>;
Parameters
| Name | Type | Description |
|---|---|---|
| stream | ReadableStream<T> | The source stream. |
| fn | Function | The predicate to test each chunk. May be async. |
Returns
A stream of the chunks that passed.
The same result as iterToStream(filterIterAsync(streamToIter(s), fn)), with one call instead of three.
Info
Built on the ReadableStream constructor rather than TransformStream, which the rest of this module also avoids: it is missing from some runtimes, and referring to it would throw when the module is imported rather than when it is used.
Rejected chunks are read and discarded as the reader asks for more, so a run of them costs reads but no memory. Cancelling the result cancels the source, and a predicate that throws does the same before erroring the stream.
Examples
TypeScript
const errors = filterStream(lines, (line) => line.startsWith('ERROR')); await streamToArray(errors) //=> ['ERROR one', 'ERROR two']
Composing with mapStream:
TypeScript
const report = mapStream( filterStream(events, (e) => e.level === 'error'), (e) => `${e.time} ${e.message}`, );