it-pipe

import { pipe } from 'it-pipe'; async function runPipelineExample() { const result = await pipe( // A source is just an iterable, this is shorthand for () => [1, 2, 3] [1, 2, 3], // A transform takes a source, and returns a source. // This transform doubles each value asynchronously. async function transform (source: AsyncIterable<number>): Promise<AsyncIterable<number>> { return (async function * (): AsyncIterable<number> { for await (const val of source) yield val * 2; })(); }, // A sink, it takes a source and consumes it, optionally returning a value. // This sink buffers up all the values from the source and returns them. async function collect (source: AsyncIterable<number>): Promise<number[]> { const vals: number[] = []; for await (const val of source) { vals.push(val); } return vals; } ); console.log('Pipelined Result:', result); // Expected: [2, 4, 6] // Demonstrating piping an async generator directly async function* generateStrings(): AsyncIterable<string> { yield 'hello'; yield 'world'; } const collectedStrings = await pipe( generateStrings(), // An async iterable source async function toUpperCase(source: AsyncIterable<string>): Promise<AsyncIterable<string>> { return (async function*(): AsyncIterable<string> { for await (const s of source) { yield s.toUpperCase(); } })(); }, async function collectAndJoin(source: AsyncIterable<string>): Promise<string> { let finalString = ''; for await (const s of source) { finalString += s + ' '; } return finalString.trim(); } ); console.log('String Pipeline Result:', collectedStrings); // Expected: "HELLO WORLD" } runPipelineExample().catch(console.error);