Async Generator Pipeline
Hard · Streams
Node.js streams are async iterables, and stream.pipeline can connect them with async generator functions as transform steps. Build a tiny version of that.
Each stage is a function that takes an async iterable and returns a new async iterable (write them as async function*):
- map(fn): yields await fn(x) for every item.
- filter(fn): yields the items for which await fn(x) is truthy.
- batch(n): yields arrays of n items; a last, shorter batch is yielded at the end if items are left over.
- take(n): yields the first n items and then stops without pulling another item. take(0) must not pull anything at all.
- pipeline(source, ...stages) connects them left to right and returns the last iterable.
Why this matters: when a consumer stops early (or a stage throws), for await calls return() on the iterator it was reading. That travels up the chain and runs the source's finally block, which is where real sources close files and sockets. If a stage reads one item too many, you lose data; if cleanup does not run, you leak handles.
The driver runPipeline(items, stages) is already written. Its source logs open, pull <item> and close (in finally). Stages are given as data, like ["map", "double"], ["filter", "even"], ["batch", 2] or ["take", 3]. It returns { output, log }, and if the pipeline throws it logs error: <message>.