@withremyinc/stream

Composable Web Streams utilities with streaming JSON and XML parsing.

Sources

Create ReadableStreams from data

arrayStream

Create a ReadableStream from an array of items.

arrayStream<T>(array: T[]): ReadableStream<T>
output 1 2 3 
import { arrayStream, collect } from "@withremyinc/stream";

const stream = arrayStream([1, 2, 3]);
const result = await collect(stream);
// [1, 2, 3]

Transforms

TransformStreams that process chunks

map

Applies a synchronous or asynchronous mapper to each chunk.

map<T, U>(mapper: (chunk: T, index: number) => U | Promise<U>): TransformStream<T, U>
input 1 2 3 map(x => x * 10)output 10 20 30 
import { arrayStream, collect, map } from "@withremyinc/stream";

const result = await collect(
  arrayStream([1, 2, 3]).pipeThrough(map(x => x * 10))
);
// [10, 20, 30]

filter

Filters chunks based on a synchronous or asynchronous predicate.

filter<T>(predicate: (chunk: T, index: number) => boolean | Promise<boolean>): TransformStream<T, T>
input 1 2 3 4 filter(x => x % 2 === 0)output 2 4 
import { arrayStream, collect, filter } from "@withremyinc/stream";

const result = await collect(
  arrayStream([1, 2, 3, 4]).pipeThrough(filter(x => x % 2 === 0))
);
// [2, 4]

filterMap

Maps each chunk to a value and drops only nullish results.

filterMap<T, U>(mapper: (chunk: T, index: number) => U | null | undefined | Promise<U | null | undefined>): TransformStream<T, NonNullable<U>>
input 1 2 3 4 filterMap(x => x > 2 ? x * 10 : null)output 30 40 
import { arrayStream, collect, filterMap } from "@withremyinc/stream";

const result = await collect(
  arrayStream([1, 2, 3, 4]).pipeThrough(filterMap(x => x > 2 ? x * 10 : null))
);
// [30, 40]

flatMap

Maps each chunk to an (async) iterable then flattens.

flatMap<T, U>(mapper: (chunk: T, index: number) => Iterable<U> | AsyncIterable<U> | U | Promise<…>): TransformStream<T, U>
input 1 2 flatMap(x => [x, x * 10])output 1 10 2 20 
import { arrayStream, collect, flatMap } from "@withremyinc/stream";

const result = await collect(
  arrayStream([1, 2]).pipeThrough(flatMap(x => [x, x * 10]))
);
// [1, 10, 2, 20]

scan

Accumulates chunks and emits each intermediate accumulator value (rolling reduce).

scan<T, U>(reducer: (acc: U, chunk: T, index: number) => U | Promise<U>, initialValue: U): TransformStream<T, U>
input 1 3 5 scan((acc, curr) => acc + curr, 0)output 1 4 9 
import { arrayStream, collect, scan } from "@withremyinc/stream";

const result = await collect(
  arrayStream([1, 3, 5]).pipeThrough(scan((acc, x) => acc + x, 0))
);
// [1, 4, 9]

reduce

Accumulates chunks into a single result, emitting it on completion.

reduce<T, U>(reducer: (acc: U, chunk: T, index: number) => U | Promise<U>, initialValue: U): TransformStream<T, U>
input 1 3 5 reduce((acc, curr) => acc + curr, 0)output 9 
import { arrayStream, collectFirst, reduce } from "@withremyinc/stream";

const result = await collectFirst(
  arrayStream([1, 3, 5]).pipeThrough(reduce((acc, x) => acc + x, 0))
);
// 9

take

Emits up to limit chunks then closes the stream.

take<T>(limit?: number): TransformStream<T, T>
input 1 2 3 4 take(2)output 1 2 
import { arrayStream, collect, take } from "@withremyinc/stream";

const result = await collect(
  arrayStream([1, 2, 3, 4]).pipeThrough(take(2))
);
// [1, 2]

takeLast

Buffers the last count chunks and emits them on completion.

takeLast<T>(count?: number): TransformStream<T, T>
input 1 2 3 4 takeLast(2)output 3 4 
import { arrayStream, collect, takeLast } from "@withremyinc/stream";

const result = await collect(
  arrayStream([1, 2, 3, 4]).pipeThrough(takeLast(2))
);
// [3, 4]

drop

Skips the first limit chunks, then emits the rest.

drop<T>(limit: number): TransformStream<T, T>
input 1 2 3 4 drop(2)output 3 4 
import { arrayStream, collect, drop } from "@withremyinc/stream";

const result = await collect(
  arrayStream([1, 2, 3, 4]).pipeThrough(drop(2))
);
// [3, 4]

forEach

Executes a side-effect function for each chunk, re-emitting the chunk unchanged.

forEach<T>(fn: (chunk: T, index: number) => void | Promise<void>): TransformStream<T, T>
input 1 2 3 forEach(x => console.log(x))output 1 2 3 
import { arrayStream, collect, forEach } from "@withremyinc/stream";

const result = await collect(
  arrayStream([1, 2, 3]).pipeThrough(forEach(x => console.log(x)))
);
// logs: 1, 2, 3  →  result: [1, 2, 3]

some

Emits true if any chunk satisfies predicate, else false.

some<T>(predicate: (chunk: T, index: number) => boolean | Promise<boolean>): TransformStream<T, boolean>
input 1 4 2 some(x => x > 3)output true 
import { arrayStream, collectFirst, some } from "@withremyinc/stream";

const result = await collectFirst(
  arrayStream([1, 4, 2]).pipeThrough(some(x => x > 3))
);
// true

every

Emits false if any chunk fails predicate, else true.

every<T>(predicate: (chunk: T, index: number) => boolean | Promise<boolean>): TransformStream<T, boolean>
input 2 4 1 every(x => x > 1)output false 
import { arrayStream, collectFirst, every } from "@withremyinc/stream";

const result = await collectFirst(
  arrayStream([2, 4, 1]).pipeThrough(every(x => x > 1))
);
// false

find

Finds the first chunk satisfying a predicate, emits it or undefined.

find<T>(predicate: (chunk: T, index: number) => boolean | Promise<boolean>): TransformStream<T, T | undefined>
input 1 4 2 find(x => x > 3)output 4 
import { arrayStream, collectFirst, find } from "@withremyinc/stream";

const result = await collectFirst(
  arrayStream([1, 4, 2]).pipeThrough(find(x => x > 3))
);
// 4

toArray

Collects all chunks into an array and emits it on completion.

toArray<T>(): TransformStream<T, T[]>
input 1 2 3 toArray()output [1,2,3] 
import { arrayStream, collectFirst, toArray } from "@withremyinc/stream";

const result = await collectFirst(
  arrayStream([1, 2, 3]).pipeThrough(toArray())
);
// [1, 2, 3]

toString

Concatenates string chunks into one string, emits on completion.

toString(): TransformStream<string, string>
input he ll o toString()output hello 
import { arrayStream, collectFirst, toString } from "@withremyinc/stream";

const result = await collectFirst(
  arrayStream(["he", "ll", "o"]).pipeThrough(toString())
);
// "hello"

extractDelimiter

Extracts the body of the first matching fenced block (e.g. markdown code fences) as a string stream. Opening and closing fence lines are removed.

extractDelimiter(options?: ExtractDelimiterOptions): TransformStream<string, string>
input ``` {…} ``` extractDelimiter()output {…} 
import { arrayStream, collectToString, extractDelimiter } from "@withremyinc/stream";

const md = `Here is some JSON:
\`\`\`json
{"name": "stream"}
\`\`\`
`;
const result = await collectToString(
  arrayStream([md]).pipeThrough(extractDelimiter({ allowLanguages: ["json"] }))
);
// '{"name": "stream"}\n'

extractFrontmatter

Splits a Markdown-style frontmatter header from the body that follows it. Emits onFrontmatter as soon as the closing delimiter line is complete, then forwards body text as onBody deltas without buffering the whole body. Delimiters are recognized only as complete lines and may be split across any number of chunks; end of input counts as a line boundary, so a closing delimiter in the final bytes needs no trailing newline. The header arrives as raw text — parse it however you like, so no YAML dependency is implied.

extractFrontmatter(options?: ExtractFrontmatterOptions): TransformStream<string, FrontmatterExtractOutput>
chunks --- k: v --- hi eventsextractFrontmatter()→ onFrontmatter "k: v"→ onBody "hi"
import { arrayStream, collect, extractFrontmatter } from "@withremyinc/stream";

const events = await collect(
  arrayStream(["---\nbehav", "ior: reply\n---\nHel", "lo"]).pipeThrough(
    extractFrontmatter()
  )
);
// [
//   { type: "onFrontmatter", raw: "behavior: reply" },
//   { type: "onBody", value: "Hel" },
//   { type: "onBody", value: "lo" },
// ]

// Parse the header however you like:
for (const event of events) {
  if (event.type === "onFrontmatter") meta = YAML.parse(event.raw);
}

tee

Duplicates the stream into two branches, processes them with a callback, and emits the results.

tee<T0, T1>(callback: (branch1: ReadableStream<T0>, branch2: ReadableStream<T0>) => ReadableStream<T1>): TransformStream<T0, T1>
input 1 2 3 tee((a, b) => merge([a, b]))output 1 1 2 2 3 3 
import { arrayStream, collect, tee, merge } from "@withremyinc/stream";

const result = await collect(
  arrayStream([1, 2, 3]).pipeThrough(
    tee((a, b) => merge([a, b]))
  )
);
// [1, 1, 2, 2, 3, 3] (order may vary)

Combinators

Merge and compose streams

merge

Merges multiple ReadableStreams into a single stream of chunks as they arrive.

merge<T>(streams: ReadableStream<T>[]): ReadableStream<T>
a a b b 1 2 output a 1 b 2 
import { arrayStream, collect, merge } from "@withremyinc/stream";

const result = await collect(
  merge([arrayStream(["a", "b"]), arrayStream(["1", "2"])])
);
// ["a", "b", "1", "2"] (order may vary)

mergeKeyed

Merges an object of ReadableStreams into a single stream of keyed chunks.

mergeKeyed<V>(streamsObj: { [K in keyof V]: ReadableStream<V[K]> }): ReadableStream<Partial<V>>
a a:1 a:2 b b:x b:y output {a:1} {b:x} {a:2} {b:y} 
import { arrayStream, collect, mergeKeyed } from "@withremyinc/stream";

const result = await collect(
  mergeKeyed({
    letters: arrayStream(["a", "b"]),
    numbers: arrayStream([1, 2]),
  })
);
// [{ letters: "a" }, { numbers: 1 }, …]

concat

Concatenates multiple ReadableStreams into a single stream, in order.

concat<T>(streams: ReadableStream<T>[]): ReadableStream<T>
a a b b 1 2 output a b 1 2 
import { arrayStream, collect, concat } from "@withremyinc/stream";

const result = await collect(
  concat([arrayStream(["a", "b"]), arrayStream(["1", "2"])])
);
// ["a", "b", "1", "2"]

pipeThrough

Compose N TransformStreams into a single TransformStream.

pipeThrough<In, Out>(...streams: TransformStream[]): TransformStream<In, Out>
input hi ok pipeThrough(upper, bracket)output [HI] [OK] 
import { pipeThrough, map } from "@withremyinc/stream";

const upper = map(s => s.toUpperCase());
const bracket = map(s => `[${s}]`);
const composed = pipeThrough(upper, bracket);

Collectors

Consume streams into values

collect

Consumes a ReadableStream and returns an array of all chunks.

collect<T>(stream: ReadableStream<T>): Promise<T[]>
stream 1 2 3 result[1, 2, 3]
import { arrayStream, collect } from "@withremyinc/stream";

const result = await collect(arrayStream([1, 2, 3]));
// [1, 2, 3]

collectToString

Consumes a ReadableStream of strings and concatenates them.

collectToString(stream: ReadableStream<string>): Promise<string>
stream he ll o resulthello
import { arrayStream, collectToString } from "@withremyinc/stream";

const result = await collectToString(arrayStream(["he", "ll", "o"]));
// "hello"

collectFirst

Retrieves the first chunk from a ReadableStream.

collectFirst<T>(stream: ReadableStream<T>): Promise<T | undefined>
stream 1 2 3 result1
import { arrayStream, collectFirst } from "@withremyinc/stream";

const result = await collectFirst(arrayStream([1, 2, 3]));
// 1

collectLast

Retrieves the last chunk from a ReadableStream.

collectLast<T>(stream: ReadableStream<T>): Promise<T | undefined>
stream 1 2 3 result3
import { arrayStream, collectLast } from "@withremyinc/stream";

const result = await collectLast(arrayStream([1, 2, 3]));
// 3

Parsers

Streaming JSON & XML parsing

parseJSON

Streaming JSON/JSONC parser. Emits SAX-style events (onObjectBegin, onObjectEnd, onArrayBegin, onArrayEnd, onLiteralValue, onObjectProperty, onError) with full JSONPath tracking. Pass { emitPartialStrings: true } to also emit onPartialLiteralValue events for open strings at chunk boundaries (useful for LLM token streams).

parseJSON(options?: JSONParserOptions): TransformStream<string, JSONParserOutput>
chunks { na me : s } eventsparseJSON()→ onObjectBegin→ onObjectProperty "name"→ onLiteralValue "s"→ onObjectEnd
import { arrayStream, collect, parseJSON } from "@withremyinc/stream";

const events = await collect(
  arrayStream(['{"na', 'me":"', 'stream"}']).pipeThrough(parseJSON())
);
// [
//   { type: "onObjectBegin", path: [] },
//   { type: "onObjectProperty", name: "name", path: [] },
//   { type: "onLiteralValue", value: "stream", path: ["name"] },
//   { type: "onObjectEnd", path: [] },
// ]

jsonToJSObject

Folds a stream of JSONParserOutput events back into a JavaScript value, emitting the value reconstructed so far every time it changes — including the partial strings from parseJSON({ emitPartialStrings: true }). Add takeLast(1) for just the completed document. Every emission is the same live accumulator rather than a copy, which is what makes emitting on every event free; copy anything you retain and treat emitted values as read-only.

jsonToJSObject(): TransformStream<JSONParserOutput, any>
chunks {… prop val …} eventsjsonToJSObject()→ → {}→ → {name: "str"}→ → {name: "stream"}
import { arrayStream, collect, collectLast, jsonToJSObject, map, parseJSON } from "@withremyinc/stream";

// Every change, as it arrives.
const snapshots = await collect(
  arrayStream(['{"name":"str', 'eam"}'])
    .pipeThrough(parseJSON({ emitPartialStrings: true }))
    .pipeThrough(jsonToJSObject())
    .pipeThrough(map((value) => structuredClone(value)))
);
// [{}, { name: "str" }, { name: "stream" }]

// Just the completed document.
const result = await collectLast(
  arrayStream(['{"name":"stream"}'])
    .pipeThrough(parseJSON())
    .pipeThrough(jsonToJSObject())
);
// { name: "stream" }

parseXML

Tolerant streaming XML parser. Emits SAX-style events (onDocumentBegin, onElementBegin, onAttribute, onText, onElementEnd, onComment, onProcessingInstruction, onCDATA, onDocumentEnd, onError).

parseXML(options?: XMLParserOptions): TransformStream<string, XMLParserOutput>
chunks <r oo t> hi </ > eventsparseXML()→ onElementBegin "root"→ onText "hi"→ onElementEnd "root"
import { arrayStream, collect, parseXML } from "@withremyinc/stream";

const events = await collect(
  arrayStream(["<root>", "hello", "</root>"]).pipeThrough(parseXML())
);

extractXML

Extracts a flat stream of allowlisted XML tags from mixed text. Nested markup inside allowed tags is surfaced as a single onText payload.

extractXML(options: XMLExtractOptions): TransformStream<string, XMLExtractOutput>
chunks Hi <a> ok </a> eventsextractXML({ allowTags: ["a"] })→ onElementBegin "a"→ onText "ok"→ onElementEnd "a"
import { arrayStream, collect, extractXML } from "@withremyinc/stream";

const events = await collect(
  arrayStream(["Hello <code>world</code>"])
    .pipeThrough(extractXML({ allowTags: ["code"] }))
);

Types

Shared type definitions

JSONParserOutput 
type JSONParserOutput =
  | { type: "onObjectBegin"; path: JSONPath }
  | { type: "onObjectProperty"; name: string | number; path: JSONPath }
  | { type: "onObjectEnd"; path: JSONPath }
  | { type: "onArrayBegin"; path: JSONPath }
  | { type: "onArrayEnd"; path: JSONPath }
  | { type: "onLiteralValue"; value: any; path: JSONPath }
  | { type: "onPartialLiteralValue"; value: string; path: JSONPath }
  | { type: "onError"; error: ParseErrorCode }
JSONParserOptions 
type JSONParserOptions = {
  /** Emit onPartialLiteralValue events for unterminated string literals at chunk boundaries. */
  emitPartialStrings?: boolean;
}
JSONPath 
type Segment = string | number;
type JSONPath = Segment[];
XMLParserOutput 
type XMLParserOutput =
  | { type: "onDocumentBegin" }
  | { type: "onDocumentEnd" }
  | { type: "onElementBegin"; name: string; attributes: XMLAttribute[] }
  | { type: "onElementEnd"; name: string }
  | { type: "onText"; text: string }
  | { type: "onComment"; text: string }
  | { type: "onProcessingInstruction"; name: string; body: string }
  | { type: "onCDATA"; text: string }
  | { type: "onError"; message: string }
XMLExtractOutput 
type XMLExtractOutput = Extract<
  XMLParserOutput,
  { type: "onElementBegin" | "onElementEnd" | "onText" | "onError" }
>
XMLParserOptions 
type XMLParserOptions = {
  /** Tags whose contents should be treated as opaque foreign text. */
  foreignTags?: readonly string[];
  /** "coalesced" (default) buffers text runs; "delta" emits text as it streams. */
  textMode?: XMLTextMode;
}
XMLExtractOptions 
type XMLExtractOptions = {
  /** Tag names to extract from mixed text. Contents are surfaced as opaque onText. */
  allowTags: readonly string[];
  /** "coalesced" (default) buffers text runs; "delta" emits text as it streams. */
  textMode?: XMLTextMode;
}
XMLTextMode 
type XMLTextMode = "coalesced" | "delta";
XMLAttribute 
type XMLAttribute = { name: string; value: string }
ExtractDelimiterOptions 
type ExtractDelimiterOptions = {
  /** Fence marker. Defaults to triple backticks. */
  delimiter?: string;
  /** Allowed fence labels (e.g. "json", "xml"). Case-insensitive. */
  allowLanguages?: readonly string[];
}
FrontmatterExtractOutput 
type FrontmatterExtractOutput =
  | { type: "onFrontmatter"; raw: string }
  | { type: "onBody"; value: string }
ExtractFrontmatterOptions 
type ExtractFrontmatterOptions = {
  /** Line marker used for both delimiters. Defaults to "---". */
  delimiter?: string;
  /** Cap on characters buffered while waiting for the closing delimiter. Defaults to 65536. */
  maxHeaderChars?: number;
}