trending/scriptc
1
0
Fork 0
mirror of https://github.com/vercel-labs/scriptc.git synced 2026-10-29 07:04:09 +08:00
Code Issues Packages Projects Releases Wiki Activity

Add async generator stages to pipeline

Browse Source 
pipeline() accepts a run of async generator function stages between
streams, following pipeline.js: each stage is called during setup with the
previous stream (a Readable is async iterable) or the previous stage's
generator, and the last generator is pumped into the next stream with
pumpToNode's write, drain and end waits. An error in the pump loop returns
the generator before the pipeline finishes, so the stage's finally runs.

A stage's source parameter is typed as what the pipeline passes it and may
only head for await loops. Function sources and destinations, stages that
take the signal argument, and stages separated by streams are refused.
This commit is contained in:
Chris Tate
2026-10-15 18:46:41 -05:00
parent 16d687a42d
commit ba3fcfb7f9
29 changed files with 1197 additions and 82 deletions
Download Patch File Download Diff File Expand all files Collapse all files
docs/content/limitations/builtin-apis/streams-and-readline.mdx
+1 -1
View File
@@ -1 +1 @@
- **Streams and readline:** Stream chunks are Buffers or strings. `Readable.from` accepts arrays, generators and async generators whose values are strings or Buffers; a stream that mixes strings and Buffers is unsupported. Object-mode streams of other values, `Readable.from` over other iterables, and generator stages in `pipeline` are unsupported. `Readable.from` accepts `objectMode` and `highWaterMark` options. The `Readable` operators (`map`, `filter`, `forEach`, `toArray`, `some`, `find`, `every`, `flatMap`, `drop`, `take`, and `reduce`) accept `concurrency`, `highWaterMark`, and `signal` options; their callbacks take only the chunk, `map` returns strings or Buffers, and `flatMap` returns strings or arrays of them. `readable.compose` is unsupported. `stream.compose` accepts two or more streams, or one `Duplex`; functions, iterables, and web streams as stages are unsupported, and the `ERR_INVALID_ARG_VALUE` message for a stream that is no longer readable or writable names the stream's class instead of printing the stream. `pipeline` from `node:stream/promises` accepts a trailing `{ signal, end }` object, and `finished` accepts `readable`, `writable`, `error`, `cleanup`, and `signal` options, with boolean literals for the flags. Stream constructors accept a `signal` option in an options object literal, and `stream.addAbortSignal` accepts any stream. `node:readline` and `node:readline/promises` interfaces read `process.stdin` or a readable stream without terminal mode: `question`, `prompt`, `write`, `pause`, `resume`, `'line'`/`'close'`/`'pause'`/`'resume'`/`'error'` listeners, and `for await` compile. Terminal mode, `completer`, history options, and chained listener registration are unsupported.
- **Streams and readline:** Stream chunks are Buffers or strings. `Readable.from` accepts arrays, generators and async generators whose values are strings or Buffers; a stream that mixes strings and Buffers is unsupported. Object-mode streams of other values and `Readable.from` over other iterables are unsupported. `pipeline` accepts one run of async generator function stages between streams (`async function* (source)`, iterating `source` with `for await`); functions as the source or destination, stages that take the `{ signal }` argument, and function stages together with pipeline options are unsupported. `Readable.from` accepts `objectMode` and `highWaterMark` options. The `Readable` operators (`map`, `filter`, `forEach`, `toArray`, `some`, `find`, `every`, `flatMap`, `drop`, `take`, and `reduce`) accept `concurrency`, `highWaterMark`, and `signal` options; their callbacks take only the chunk, `map` returns strings or Buffers, and `flatMap` returns strings or arrays of them. `readable.compose` is unsupported. `stream.compose` accepts two or more streams, or one `Duplex`; functions, iterables, and web streams as stages are unsupported, and the `ERR_INVALID_ARG_VALUE` message for a stream that is no longer readable or writable names the stream's class instead of printing the stream. `pipeline` from `node:stream/promises` accepts a trailing `{ signal, end }` object, and `finished` accepts `readable`, `writable`, `error`, `cleanup`, and `signal` options, with boolean literals for the flags. Stream constructors accept a `signal` option in an options object literal, and `stream.addAbortSignal` accepts any stream. `node:readline` and `node:readline/promises` interfaces read `process.stdin` or a readable stream without terminal mode: `question`, `prompt`, `write`, `pause`, `resume`, `'line'`/`'close'`/`'pause'`/`'resume'`/`'error'` listeners, and `for await` compile. Terminal mode, `completer`, history options, and chained listener registration are unsupported.
internal/compatibility/generated/node-v24-backlog.json
+90 -30
View File
@@ -61644,7 +61644,9 @@
"tests/corpus/1688-stream-writable-basics.ts",
"tests/corpus/1814-stream-pipeline.cjs",
"tests/corpus/5302-stream-promises-options.ts",
"tests/corpus/5303-stream-finished-callback-opts.cjs"
"tests/corpus/5303-stream-finished-callback-opts.cjs",
"tests/corpus/5425-pipeline-generator-stage.ts",
"tests/corpus/5428-pipeline-generator-stages.cjs"
]
}
}
@@ -61675,7 +61677,9 @@
"tests/corpus/1688-stream-writable-basics.ts",
"tests/corpus/1814-stream-pipeline.cjs",
"tests/corpus/5302-stream-promises-options.ts",
"tests/corpus/5303-stream-finished-callback-opts.cjs"
"tests/corpus/5303-stream-finished-callback-opts.cjs",
"tests/corpus/5425-pipeline-generator-stage.ts",
"tests/corpus/5428-pipeline-generator-stages.cjs"
]
}
}
@@ -61706,7 +61710,9 @@
"tests/corpus/1688-stream-writable-basics.ts",
"tests/corpus/1814-stream-pipeline.cjs",
"tests/corpus/5302-stream-promises-options.ts",
"tests/corpus/5303-stream-finished-callback-opts.cjs"
"tests/corpus/5303-stream-finished-callback-opts.cjs",
"tests/corpus/5425-pipeline-generator-stage.ts",
"tests/corpus/5428-pipeline-generator-stages.cjs"
]
}
}
@@ -61737,7 +61743,9 @@
"tests/corpus/1688-stream-writable-basics.ts",
"tests/corpus/1814-stream-pipeline.cjs",
"tests/corpus/5302-stream-promises-options.ts",
"tests/corpus/5303-stream-finished-callback-opts.cjs"
"tests/corpus/5303-stream-finished-callback-opts.cjs",
"tests/corpus/5425-pipeline-generator-stage.ts",
"tests/corpus/5428-pipeline-generator-stages.cjs"
]
}
}
@@ -61768,7 +61776,9 @@
"tests/corpus/1688-stream-writable-basics.ts",
"tests/corpus/1814-stream-pipeline.cjs",
"tests/corpus/5302-stream-promises-options.ts",
"tests/corpus/5303-stream-finished-callback-opts.cjs"
"tests/corpus/5303-stream-finished-callback-opts.cjs",
"tests/corpus/5425-pipeline-generator-stage.ts",
"tests/corpus/5428-pipeline-generator-stages.cjs"
]
}
}
@@ -61799,7 +61809,9 @@
"tests/corpus/1688-stream-writable-basics.ts",
"tests/corpus/1814-stream-pipeline.cjs",
"tests/corpus/5302-stream-promises-options.ts",
"tests/corpus/5303-stream-finished-callback-opts.cjs"
"tests/corpus/5303-stream-finished-callback-opts.cjs",
"tests/corpus/5425-pipeline-generator-stage.ts",
"tests/corpus/5428-pipeline-generator-stages.cjs"
]
}
}
@@ -62216,7 +62228,9 @@
"tests/corpus/1688-stream-writable-basics.ts",
"tests/corpus/1814-stream-pipeline.cjs",
"tests/corpus/5302-stream-promises-options.ts",
"tests/corpus/5303-stream-finished-callback-opts.cjs"
"tests/corpus/5303-stream-finished-callback-opts.cjs",
"tests/corpus/5425-pipeline-generator-stage.ts",
"tests/corpus/5428-pipeline-generator-stages.cjs"
]
}
}
@@ -62272,7 +62286,9 @@
"tests/corpus/1688-stream-writable-basics.ts",
"tests/corpus/1814-stream-pipeline.cjs",
"tests/corpus/5302-stream-promises-options.ts",
"tests/corpus/5303-stream-finished-callback-opts.cjs"
"tests/corpus/5303-stream-finished-callback-opts.cjs",
"tests/corpus/5425-pipeline-generator-stage.ts",
"tests/corpus/5428-pipeline-generator-stages.cjs"
]
}
}
@@ -62303,7 +62319,9 @@
"tests/corpus/1688-stream-writable-basics.ts",
"tests/corpus/1814-stream-pipeline.cjs",
"tests/corpus/5302-stream-promises-options.ts",
"tests/corpus/5303-stream-finished-callback-opts.cjs"
"tests/corpus/5303-stream-finished-callback-opts.cjs",
"tests/corpus/5425-pipeline-generator-stage.ts",
"tests/corpus/5428-pipeline-generator-stages.cjs"
]
}
}
@@ -62334,7 +62352,9 @@
"tests/corpus/1688-stream-writable-basics.ts",
"tests/corpus/1814-stream-pipeline.cjs",
"tests/corpus/5302-stream-promises-options.ts",
"tests/corpus/5303-stream-finished-callback-opts.cjs"
"tests/corpus/5303-stream-finished-callback-opts.cjs",
"tests/corpus/5425-pipeline-generator-stage.ts",
"tests/corpus/5428-pipeline-generator-stages.cjs"
]
}
}
@@ -62365,7 +62385,9 @@
"tests/corpus/1688-stream-writable-basics.ts",
"tests/corpus/1814-stream-pipeline.cjs",
"tests/corpus/5302-stream-promises-options.ts",
"tests/corpus/5303-stream-finished-callback-opts.cjs"
"tests/corpus/5303-stream-finished-callback-opts.cjs",
"tests/corpus/5425-pipeline-generator-stage.ts",
"tests/corpus/5428-pipeline-generator-stages.cjs"
]
}
}
@@ -62396,7 +62418,9 @@
"tests/corpus/1688-stream-writable-basics.ts",
"tests/corpus/1814-stream-pipeline.cjs",
"tests/corpus/5302-stream-promises-options.ts",
"tests/corpus/5303-stream-finished-callback-opts.cjs"
"tests/corpus/5303-stream-finished-callback-opts.cjs",
"tests/corpus/5425-pipeline-generator-stage.ts",
"tests/corpus/5428-pipeline-generator-stages.cjs"
]
}
}
@@ -62813,7 +62837,9 @@
"tests/corpus/1688-stream-writable-basics.ts",
"tests/corpus/1814-stream-pipeline.cjs",
"tests/corpus/5302-stream-promises-options.ts",
"tests/corpus/5303-stream-finished-callback-opts.cjs"
"tests/corpus/5303-stream-finished-callback-opts.cjs",
"tests/corpus/5425-pipeline-generator-stage.ts",
"tests/corpus/5428-pipeline-generator-stages.cjs"
]
}
}
@@ -62894,7 +62920,9 @@
"tests/corpus/1688-stream-writable-basics.ts",
"tests/corpus/1814-stream-pipeline.cjs",
"tests/corpus/5302-stream-promises-options.ts",
"tests/corpus/5303-stream-finished-callback-opts.cjs"
"tests/corpus/5303-stream-finished-callback-opts.cjs",
"tests/corpus/5425-pipeline-generator-stage.ts",
"tests/corpus/5428-pipeline-generator-stages.cjs"
]
}
}
@@ -62952,7 +62980,9 @@
"tests/corpus/1688-stream-writable-basics.ts",
"tests/corpus/1814-stream-pipeline.cjs",
"tests/corpus/5302-stream-promises-options.ts",
"tests/corpus/5303-stream-finished-callback-opts.cjs"
"tests/corpus/5303-stream-finished-callback-opts.cjs",
"tests/corpus/5425-pipeline-generator-stage.ts",
"tests/corpus/5428-pipeline-generator-stages.cjs"
]
}
}
@@ -62983,7 +63013,9 @@
"tests/corpus/1688-stream-writable-basics.ts",
"tests/corpus/1814-stream-pipeline.cjs",
"tests/corpus/5302-stream-promises-options.ts",
"tests/corpus/5303-stream-finished-callback-opts.cjs"
"tests/corpus/5303-stream-finished-callback-opts.cjs",
"tests/corpus/5425-pipeline-generator-stage.ts",
"tests/corpus/5428-pipeline-generator-stages.cjs"
]
}
}
@@ -63014,7 +63046,9 @@
"tests/corpus/1688-stream-writable-basics.ts",
"tests/corpus/1814-stream-pipeline.cjs",
"tests/corpus/5302-stream-promises-options.ts",
"tests/corpus/5303-stream-finished-callback-opts.cjs"
"tests/corpus/5303-stream-finished-callback-opts.cjs",
"tests/corpus/5425-pipeline-generator-stage.ts",
"tests/corpus/5428-pipeline-generator-stages.cjs"
]
}
}
@@ -63070,7 +63104,9 @@
"tests/corpus/1688-stream-writable-basics.ts",
"tests/corpus/1814-stream-pipeline.cjs",
"tests/corpus/5302-stream-promises-options.ts",
"tests/corpus/5303-stream-finished-callback-opts.cjs"
"tests/corpus/5303-stream-finished-callback-opts.cjs",
"tests/corpus/5425-pipeline-generator-stage.ts",
"tests/corpus/5428-pipeline-generator-stages.cjs"
]
}
}
@@ -63606,7 +63642,9 @@
"tests/corpus/1688-stream-writable-basics.ts",
"tests/corpus/1814-stream-pipeline.cjs",
"tests/corpus/5302-stream-promises-options.ts",
"tests/corpus/5303-stream-finished-callback-opts.cjs"
"tests/corpus/5303-stream-finished-callback-opts.cjs",
"tests/corpus/5425-pipeline-generator-stage.ts",
"tests/corpus/5428-pipeline-generator-stages.cjs"
]
}
}
@@ -63637,7 +63675,9 @@
"tests/corpus/1688-stream-writable-basics.ts",
"tests/corpus/1814-stream-pipeline.cjs",
"tests/corpus/5302-stream-promises-options.ts",
"tests/corpus/5303-stream-finished-callback-opts.cjs"
"tests/corpus/5303-stream-finished-callback-opts.cjs",
"tests/corpus/5425-pipeline-generator-stage.ts",
"tests/corpus/5428-pipeline-generator-stages.cjs"
]
}
}
@@ -63668,7 +63708,9 @@
"tests/corpus/1688-stream-writable-basics.ts",
"tests/corpus/1814-stream-pipeline.cjs",
"tests/corpus/5302-stream-promises-options.ts",
"tests/corpus/5303-stream-finished-callback-opts.cjs"
"tests/corpus/5303-stream-finished-callback-opts.cjs",
"tests/corpus/5425-pipeline-generator-stage.ts",
"tests/corpus/5428-pipeline-generator-stages.cjs"
]
}
}
@@ -63699,7 +63741,9 @@
"tests/corpus/1688-stream-writable-basics.ts",
"tests/corpus/1814-stream-pipeline.cjs",
"tests/corpus/5302-stream-promises-options.ts",
"tests/corpus/5303-stream-finished-callback-opts.cjs"
"tests/corpus/5303-stream-finished-callback-opts.cjs",
"tests/corpus/5425-pipeline-generator-stage.ts",
"tests/corpus/5428-pipeline-generator-stages.cjs"
]
}
}
@@ -63730,7 +63774,9 @@
"tests/corpus/1688-stream-writable-basics.ts",
"tests/corpus/1814-stream-pipeline.cjs",
"tests/corpus/5302-stream-promises-options.ts",
"tests/corpus/5303-stream-finished-callback-opts.cjs"
"tests/corpus/5303-stream-finished-callback-opts.cjs",
"tests/corpus/5425-pipeline-generator-stage.ts",
"tests/corpus/5428-pipeline-generator-stages.cjs"
]
}
}
@@ -63761,7 +63807,9 @@
"tests/corpus/1688-stream-writable-basics.ts",
"tests/corpus/1814-stream-pipeline.cjs",
"tests/corpus/5302-stream-promises-options.ts",
"tests/corpus/5303-stream-finished-callback-opts.cjs"
"tests/corpus/5303-stream-finished-callback-opts.cjs",
"tests/corpus/5425-pipeline-generator-stage.ts",
"tests/corpus/5428-pipeline-generator-stages.cjs"
]
}
}
@@ -63792,7 +63840,9 @@
"tests/corpus/1688-stream-writable-basics.ts",
"tests/corpus/1814-stream-pipeline.cjs",
"tests/corpus/5302-stream-promises-options.ts",
"tests/corpus/5303-stream-finished-callback-opts.cjs"
"tests/corpus/5303-stream-finished-callback-opts.cjs",
"tests/corpus/5425-pipeline-generator-stage.ts",
"tests/corpus/5428-pipeline-generator-stages.cjs"
]
}
}
@@ -63823,7 +63873,9 @@
"tests/corpus/1688-stream-writable-basics.ts",
"tests/corpus/1814-stream-pipeline.cjs",
"tests/corpus/5302-stream-promises-options.ts",
"tests/corpus/5303-stream-finished-callback-opts.cjs"
"tests/corpus/5303-stream-finished-callback-opts.cjs",
"tests/corpus/5425-pipeline-generator-stage.ts",
"tests/corpus/5428-pipeline-generator-stages.cjs"
]
}
}
@@ -63854,7 +63906,9 @@
"tests/corpus/1688-stream-writable-basics.ts",
"tests/corpus/1814-stream-pipeline.cjs",
"tests/corpus/5302-stream-promises-options.ts",
"tests/corpus/5303-stream-finished-callback-opts.cjs"
"tests/corpus/5303-stream-finished-callback-opts.cjs",
"tests/corpus/5425-pipeline-generator-stage.ts",
"tests/corpus/5428-pipeline-generator-stages.cjs"
]
}
}
@@ -65146,7 +65200,9 @@
"tests/corpus/1688-stream-writable-basics.ts",
"tests/corpus/1814-stream-pipeline.cjs",
"tests/corpus/5302-stream-promises-options.ts",
"tests/corpus/5303-stream-finished-callback-opts.cjs"
"tests/corpus/5303-stream-finished-callback-opts.cjs",
"tests/corpus/5425-pipeline-generator-stage.ts",
"tests/corpus/5428-pipeline-generator-stages.cjs"
]
}
}
@@ -65302,7 +65358,9 @@
"tests/corpus/1688-stream-writable-basics.ts",
"tests/corpus/1814-stream-pipeline.cjs",
"tests/corpus/5302-stream-promises-options.ts",
"tests/corpus/5303-stream-finished-callback-opts.cjs"
"tests/corpus/5303-stream-finished-callback-opts.cjs",
"tests/corpus/5425-pipeline-generator-stage.ts",
"tests/corpus/5428-pipeline-generator-stages.cjs"
]
}
}
@@ -65333,7 +65391,9 @@
"tests/corpus/1688-stream-writable-basics.ts",
"tests/corpus/1814-stream-pipeline.cjs",
"tests/corpus/5302-stream-promises-options.ts",
"tests/corpus/5303-stream-finished-callback-opts.cjs"
"tests/corpus/5303-stream-finished-callback-opts.cjs",
"tests/corpus/5425-pipeline-generator-stage.ts",
"tests/corpus/5428-pipeline-generator-stages.cjs"
]
}
}
internal/compatibility/generated/node-v24-internal.json
+93 -31
View File
@@ -85812,7 +85812,9 @@
"tests/corpus/5342-stream-compose.ts",
"tests/corpus/5343-stream-compose-edges.ts",
"tests/corpus/5420-readable-from-generators.ts",
"tests/corpus/5421-readable-from-generators.cjs"
"tests/corpus/5421-readable-from-generators.cjs",
"tests/corpus/5425-pipeline-generator-stage.ts",
"tests/corpus/5428-pipeline-generator-stages.cjs"
]
},
"anchorSource": "chapter"
@@ -85934,7 +85936,9 @@
"tests/corpus/1688-stream-writable-basics.ts",
"tests/corpus/1814-stream-pipeline.cjs",
"tests/corpus/5302-stream-promises-options.ts",
"tests/corpus/5303-stream-finished-callback-opts.cjs"
"tests/corpus/5303-stream-finished-callback-opts.cjs",
"tests/corpus/5425-pipeline-generator-stage.ts",
"tests/corpus/5428-pipeline-generator-stages.cjs"
]
},
"anchorSource": "exact"
@@ -85964,7 +85968,9 @@
"tests/corpus/1688-stream-writable-basics.ts",
"tests/corpus/1814-stream-pipeline.cjs",
"tests/corpus/5302-stream-promises-options.ts",
"tests/corpus/5303-stream-finished-callback-opts.cjs"
"tests/corpus/5303-stream-finished-callback-opts.cjs",
"tests/corpus/5425-pipeline-generator-stage.ts",
"tests/corpus/5428-pipeline-generator-stages.cjs"
]
},
"anchorSource": "exact"
@@ -85994,7 +86000,9 @@
"tests/corpus/1688-stream-writable-basics.ts",
"tests/corpus/1814-stream-pipeline.cjs",
"tests/corpus/5302-stream-promises-options.ts",
"tests/corpus/5303-stream-finished-callback-opts.cjs"
"tests/corpus/5303-stream-finished-callback-opts.cjs",
"tests/corpus/5425-pipeline-generator-stage.ts",
"tests/corpus/5428-pipeline-generator-stages.cjs"
]
},
"anchorSource": "exact"
@@ -86047,7 +86055,9 @@
"tests/corpus/1688-stream-writable-basics.ts",
"tests/corpus/1814-stream-pipeline.cjs",
"tests/corpus/5302-stream-promises-options.ts",
"tests/corpus/5303-stream-finished-callback-opts.cjs"
"tests/corpus/5303-stream-finished-callback-opts.cjs",
"tests/corpus/5425-pipeline-generator-stage.ts",
"tests/corpus/5428-pipeline-generator-stages.cjs"
]
},
"anchorSource": "exact"
@@ -86077,7 +86087,9 @@
"tests/corpus/1688-stream-writable-basics.ts",
"tests/corpus/1814-stream-pipeline.cjs",
"tests/corpus/5302-stream-promises-options.ts",
"tests/corpus/5303-stream-finished-callback-opts.cjs"
"tests/corpus/5303-stream-finished-callback-opts.cjs",
"tests/corpus/5425-pipeline-generator-stage.ts",
"tests/corpus/5428-pipeline-generator-stages.cjs"
]
},
"anchorSource": "exact"
@@ -86107,7 +86119,9 @@
"tests/corpus/1688-stream-writable-basics.ts",
"tests/corpus/1814-stream-pipeline.cjs",
"tests/corpus/5302-stream-promises-options.ts",
"tests/corpus/5303-stream-finished-callback-opts.cjs"
"tests/corpus/5303-stream-finished-callback-opts.cjs",
"tests/corpus/5425-pipeline-generator-stage.ts",
"tests/corpus/5428-pipeline-generator-stages.cjs"
]
},
"anchorSource": "exact"
@@ -86496,7 +86510,9 @@
"tests/corpus/1688-stream-writable-basics.ts",
"tests/corpus/1814-stream-pipeline.cjs",
"tests/corpus/5302-stream-promises-options.ts",
"tests/corpus/5303-stream-finished-callback-opts.cjs"
"tests/corpus/5303-stream-finished-callback-opts.cjs",
"tests/corpus/5425-pipeline-generator-stage.ts",
"tests/corpus/5428-pipeline-generator-stages.cjs"
]
},
"anchorSource": "exact"
@@ -86549,7 +86565,9 @@
"tests/corpus/1688-stream-writable-basics.ts",
"tests/corpus/1814-stream-pipeline.cjs",
"tests/corpus/5302-stream-promises-options.ts",
"tests/corpus/5303-stream-finished-callback-opts.cjs"
"tests/corpus/5303-stream-finished-callback-opts.cjs",
"tests/corpus/5425-pipeline-generator-stage.ts",
"tests/corpus/5428-pipeline-generator-stages.cjs"
]
},
"anchorSource": "exact"
@@ -86579,7 +86597,9 @@
"tests/corpus/1688-stream-writable-basics.ts",
"tests/corpus/1814-stream-pipeline.cjs",
"tests/corpus/5302-stream-promises-options.ts",
"tests/corpus/5303-stream-finished-callback-opts.cjs"
"tests/corpus/5303-stream-finished-callback-opts.cjs",
"tests/corpus/5425-pipeline-generator-stage.ts",
"tests/corpus/5428-pipeline-generator-stages.cjs"
]
},
"anchorSource": "exact"
@@ -86632,7 +86652,9 @@
"tests/corpus/1688-stream-writable-basics.ts",
"tests/corpus/1814-stream-pipeline.cjs",
"tests/corpus/5302-stream-promises-options.ts",
"tests/corpus/5303-stream-finished-callback-opts.cjs"
"tests/corpus/5303-stream-finished-callback-opts.cjs",
"tests/corpus/5425-pipeline-generator-stage.ts",
"tests/corpus/5428-pipeline-generator-stages.cjs"
]
},
"anchorSource": "ancestor"
@@ -86662,7 +86684,9 @@
"tests/corpus/1688-stream-writable-basics.ts",
"tests/corpus/1814-stream-pipeline.cjs",
"tests/corpus/5302-stream-promises-options.ts",
"tests/corpus/5303-stream-finished-callback-opts.cjs"
"tests/corpus/5303-stream-finished-callback-opts.cjs",
"tests/corpus/5425-pipeline-generator-stage.ts",
"tests/corpus/5428-pipeline-generator-stages.cjs"
]
},
"anchorSource": "ancestor"
@@ -86692,7 +86716,9 @@
"tests/corpus/1688-stream-writable-basics.ts",
"tests/corpus/1814-stream-pipeline.cjs",
"tests/corpus/5302-stream-promises-options.ts",
"tests/corpus/5303-stream-finished-callback-opts.cjs"
"tests/corpus/5303-stream-finished-callback-opts.cjs",
"tests/corpus/5425-pipeline-generator-stage.ts",
"tests/corpus/5428-pipeline-generator-stages.cjs"
]
},
"anchorSource": "ancestor"
@@ -87081,7 +87107,9 @@
"tests/corpus/1688-stream-writable-basics.ts",
"tests/corpus/1814-stream-pipeline.cjs",
"tests/corpus/5302-stream-promises-options.ts",
"tests/corpus/5303-stream-finished-callback-opts.cjs"
"tests/corpus/5303-stream-finished-callback-opts.cjs",
"tests/corpus/5425-pipeline-generator-stage.ts",
"tests/corpus/5428-pipeline-generator-stages.cjs"
]
},
"anchorSource": "ancestor"
@@ -87180,7 +87208,9 @@
"tests/corpus/1688-stream-writable-basics.ts",
"tests/corpus/1814-stream-pipeline.cjs",
"tests/corpus/5302-stream-promises-options.ts",
"tests/corpus/5303-stream-finished-callback-opts.cjs"
"tests/corpus/5303-stream-finished-callback-opts.cjs",
"tests/corpus/5425-pipeline-generator-stage.ts",
"tests/corpus/5428-pipeline-generator-stages.cjs"
]
},
"anchorSource": "exact"
@@ -87236,7 +87266,9 @@
"tests/corpus/1688-stream-writable-basics.ts",
"tests/corpus/1814-stream-pipeline.cjs",
"tests/corpus/5302-stream-promises-options.ts",
"tests/corpus/5303-stream-finished-callback-opts.cjs"
"tests/corpus/5303-stream-finished-callback-opts.cjs",
"tests/corpus/5425-pipeline-generator-stage.ts",
"tests/corpus/5428-pipeline-generator-stages.cjs"
]
},
"anchorSource": "exact"
@@ -87266,7 +87298,9 @@
"tests/corpus/1688-stream-writable-basics.ts",
"tests/corpus/1814-stream-pipeline.cjs",
"tests/corpus/5302-stream-promises-options.ts",
"tests/corpus/5303-stream-finished-callback-opts.cjs"
"tests/corpus/5303-stream-finished-callback-opts.cjs",
"tests/corpus/5425-pipeline-generator-stage.ts",
"tests/corpus/5428-pipeline-generator-stages.cjs"
]
},
"anchorSource": "exact"
@@ -87296,7 +87330,9 @@
"tests/corpus/1688-stream-writable-basics.ts",
"tests/corpus/1814-stream-pipeline.cjs",
"tests/corpus/5302-stream-promises-options.ts",
"tests/corpus/5303-stream-finished-callback-opts.cjs"
"tests/corpus/5303-stream-finished-callback-opts.cjs",
"tests/corpus/5425-pipeline-generator-stage.ts",
"tests/corpus/5428-pipeline-generator-stages.cjs"
]
},
"anchorSource": "exact"
@@ -87349,7 +87385,9 @@
"tests/corpus/1688-stream-writable-basics.ts",
"tests/corpus/1814-stream-pipeline.cjs",
"tests/corpus/5302-stream-promises-options.ts",
"tests/corpus/5303-stream-finished-callback-opts.cjs"
"tests/corpus/5303-stream-finished-callback-opts.cjs",
"tests/corpus/5425-pipeline-generator-stage.ts",
"tests/corpus/5428-pipeline-generator-stages.cjs"
]
},
"anchorSource": "exact"
@@ -87938,7 +87976,9 @@
"tests/corpus/1688-stream-writable-basics.ts",
"tests/corpus/1814-stream-pipeline.cjs",
"tests/corpus/5302-stream-promises-options.ts",
"tests/corpus/5303-stream-finished-callback-opts.cjs"
"tests/corpus/5303-stream-finished-callback-opts.cjs",
"tests/corpus/5425-pipeline-generator-stage.ts",
"tests/corpus/5428-pipeline-generator-stages.cjs"
]
},
"anchorSource": "exact"
@@ -87968,7 +88008,9 @@
"tests/corpus/1688-stream-writable-basics.ts",
"tests/corpus/1814-stream-pipeline.cjs",
"tests/corpus/5302-stream-promises-options.ts",
"tests/corpus/5303-stream-finished-callback-opts.cjs"
"tests/corpus/5303-stream-finished-callback-opts.cjs",
"tests/corpus/5425-pipeline-generator-stage.ts",
"tests/corpus/5428-pipeline-generator-stages.cjs"
]
},
"anchorSource": "exact"
@@ -87998,7 +88040,9 @@
"tests/corpus/1688-stream-writable-basics.ts",
"tests/corpus/1814-stream-pipeline.cjs",
"tests/corpus/5302-stream-promises-options.ts",
"tests/corpus/5303-stream-finished-callback-opts.cjs"
"tests/corpus/5303-stream-finished-callback-opts.cjs",
"tests/corpus/5425-pipeline-generator-stage.ts",
"tests/corpus/5428-pipeline-generator-stages.cjs"
]
},
"anchorSource": "exact"
@@ -88028,7 +88072,9 @@
"tests/corpus/1688-stream-writable-basics.ts",
"tests/corpus/1814-stream-pipeline.cjs",
"tests/corpus/5302-stream-promises-options.ts",
"tests/corpus/5303-stream-finished-callback-opts.cjs"
"tests/corpus/5303-stream-finished-callback-opts.cjs",
"tests/corpus/5425-pipeline-generator-stage.ts",
"tests/corpus/5428-pipeline-generator-stages.cjs"
]
},
"anchorSource": "exact"
@@ -88058,7 +88104,9 @@
"tests/corpus/1688-stream-writable-basics.ts",
"tests/corpus/1814-stream-pipeline.cjs",
"tests/corpus/5302-stream-promises-options.ts",
"tests/corpus/5303-stream-finished-callback-opts.cjs"
"tests/corpus/5303-stream-finished-callback-opts.cjs",
"tests/corpus/5425-pipeline-generator-stage.ts",
"tests/corpus/5428-pipeline-generator-stages.cjs"
]
},
"anchorSource": "exact"
@@ -88088,7 +88136,9 @@
"tests/corpus/1688-stream-writable-basics.ts",
"tests/corpus/1814-stream-pipeline.cjs",
"tests/corpus/5302-stream-promises-options.ts",
"tests/corpus/5303-stream-finished-callback-opts.cjs"
"tests/corpus/5303-stream-finished-callback-opts.cjs",
"tests/corpus/5425-pipeline-generator-stage.ts",
"tests/corpus/5428-pipeline-generator-stages.cjs"
]
},
"anchorSource": "exact"
@@ -88118,7 +88168,9 @@
"tests/corpus/1688-stream-writable-basics.ts",
"tests/corpus/1814-stream-pipeline.cjs",
"tests/corpus/5302-stream-promises-options.ts",
"tests/corpus/5303-stream-finished-callback-opts.cjs"
"tests/corpus/5303-stream-finished-callback-opts.cjs",
"tests/corpus/5425-pipeline-generator-stage.ts",
"tests/corpus/5428-pipeline-generator-stages.cjs"
]
},
"anchorSource": "exact"
@@ -88148,7 +88200,9 @@
"tests/corpus/1688-stream-writable-basics.ts",
"tests/corpus/1814-stream-pipeline.cjs",
"tests/corpus/5302-stream-promises-options.ts",
"tests/corpus/5303-stream-finished-callback-opts.cjs"
"tests/corpus/5303-stream-finished-callback-opts.cjs",
"tests/corpus/5425-pipeline-generator-stage.ts",
"tests/corpus/5428-pipeline-generator-stages.cjs"
]
},
"anchorSource": "exact"
@@ -88178,7 +88232,9 @@
"tests/corpus/1688-stream-writable-basics.ts",
"tests/corpus/1814-stream-pipeline.cjs",
"tests/corpus/5302-stream-promises-options.ts",
"tests/corpus/5303-stream-finished-callback-opts.cjs"
"tests/corpus/5303-stream-finished-callback-opts.cjs",
"tests/corpus/5425-pipeline-generator-stage.ts",
"tests/corpus/5428-pipeline-generator-stages.cjs"
]
},
"anchorSource": "exact"
@@ -89593,7 +89649,9 @@
"tests/corpus/1688-stream-writable-basics.ts",
"tests/corpus/1814-stream-pipeline.cjs",
"tests/corpus/5302-stream-promises-options.ts",
"tests/corpus/5303-stream-finished-callback-opts.cjs"
"tests/corpus/5303-stream-finished-callback-opts.cjs",
"tests/corpus/5425-pipeline-generator-stage.ts",
"tests/corpus/5428-pipeline-generator-stages.cjs"
]
},
"anchorSource": "exact"
@@ -89853,7 +89911,9 @@
"tests/corpus/1688-stream-writable-basics.ts",
"tests/corpus/1814-stream-pipeline.cjs",
"tests/corpus/5302-stream-promises-options.ts",
"tests/corpus/5303-stream-finished-callback-opts.cjs"
"tests/corpus/5303-stream-finished-callback-opts.cjs",
"tests/corpus/5425-pipeline-generator-stage.ts",
"tests/corpus/5428-pipeline-generator-stages.cjs"
]
},
"anchorSource": "ancestor"
@@ -89883,7 +89943,9 @@
"tests/corpus/1688-stream-writable-basics.ts",
"tests/corpus/1814-stream-pipeline.cjs",
"tests/corpus/5302-stream-promises-options.ts",
"tests/corpus/5303-stream-finished-callback-opts.cjs"
"tests/corpus/5303-stream-finished-callback-opts.cjs",
"tests/corpus/5425-pipeline-generator-stage.ts",
"tests/corpus/5428-pipeline-generator-stages.cjs"
]
},
"anchorSource": "ancestor"
internal/compatibility/static-support.json
+1 -1
View File
@@ -350,7 +350,7 @@
{ "symbols": ["AsyncLocalStorage", "asyncLocalStorage.run", "asyncLocalStorage.exit", "asyncLocalStorage.getStore", "asyncLocalStorage.enterWith", "asyncLocalStorage.disable"], "status": "partial", "evidence": ["tests/corpus/2213-async-local-storage.cjs"] },
{ "symbols": ["StringDecoder", "stringDecoder.write", "stringDecoder.end"], "status": "partial", "evidence": ["tests/corpus/1662-string-decoder-encodings.ts", "tests/corpus/node24-string-decoder.ts"] },
{ "symbols": ["setImmediate", "setInterval", "setTimeout", "clearImmediate", "clearInterval", "clearTimeout", "timersPromises.setTimeout", "timersPromises.setImmediate", "timersPromises.setInterval", "immediate.hasRef", "immediate.ref", "immediate.unref", "timeout.hasRef", "timeout.ref", "timeout.unref", "timeout.refresh"], "status": "partial", "evidence": ["tests/corpus/1804-timers-module.ts", "tests/corpus/1805-timer-callback-args.ts", "tests/corpus/2093-timers-promises.ts", "tests/corpus/2852-timers-promises-interval.ts"] },
{ "symbols": ["stream.finished", "stream.pipeline", "stream.getDefaultHighWaterMark", "readable.push", "readable.unshift", "readable.read", "readable.pause", "readable.resume", "readable.isPaused", "readable.setEncoding", "readable.pipe", "readable.unpipe", "readable.destroy", "writable.write", "writable.end", "writable.cork", "writable.uncork", "writable.setDefaultEncoding"], "status": "partial", "evidence": ["tests/corpus/1685-stream-readable-basics.ts", "tests/corpus/1688-stream-writable-basics.ts", "tests/corpus/1814-stream-pipeline.cjs", "tests/corpus/5302-stream-promises-options.ts", "tests/corpus/5303-stream-finished-callback-opts.cjs"] },
{ "symbols": ["stream.finished", "stream.pipeline", "stream.getDefaultHighWaterMark", "readable.push", "readable.unshift", "readable.read", "readable.pause", "readable.resume", "readable.isPaused", "readable.setEncoding", "readable.pipe", "readable.unpipe", "readable.destroy", "writable.write", "writable.end", "writable.cork", "writable.uncork", "writable.setDefaultEncoding"], "status": "partial", "evidence": ["tests/corpus/1685-stream-readable-basics.ts", "tests/corpus/1688-stream-writable-basics.ts", "tests/corpus/1814-stream-pipeline.cjs", "tests/corpus/5302-stream-promises-options.ts", "tests/corpus/5303-stream-finished-callback-opts.cjs", "tests/corpus/5425-pipeline-generator-stage.ts", "tests/corpus/5428-pipeline-generator-stages.cjs"] },
{ "symbols": ["net.createServer", "net.connect", "net.createConnection", "net.getDefaultAutoSelectFamilyAttemptTimeout", "net.setDefaultAutoSelectFamilyAttemptTimeout", "server.address", "server.close", "server.listen", "socket.destroy", "socket.end", "socket.pause", "socket.resume", "socket.setEncoding", "socket.setTimeout", "socket.write"], "status": "partial", "evidence": ["tests/corpus/2500-net-autosel-timeout.ts", "tests/corpus/2596-net-arg-ladders.cjs"] },
{ "symbols": ["dgram.createSocket", "socket.address", "socket.bind", "socket.close", "socket.connect", "socket.ref", "socket.remoteAddress", "socket.send", "socket.unref"], "status": "partial", "evidence": ["tests/corpus/2597-dgram-send-ladders.cjs"] },
{ "symbols": ["readline.createInterface", "rl.close"], "status": "partial", "evidence": ["tests/corpus/1475-readline-closed-stdin.ts"] },
packages/compiler/src/backend/llvm/lib-dispatch.ts
+1
View File
@@ -440,6 +440,7 @@ export function emitLibCall(host: LlvmEmitterContext, e: LibCallExpr): LlValue {
if (e.fn === "process.nextTick") return host.emitAsyncContextLibCall(e);
if (
e.fn === "sp.pipeline" ||
e.fn === "sp.pipelineStages" ||
e.fn.startsWith("sp.pipelineOpts") ||
e.fn.startsWith("sp.finishedOpts")
)
packages/compiler/src/backend/llvm/lib-shared.ts
+3
View File
@@ -1380,6 +1380,9 @@ export const USES_TIMERS_LIB_FNS = new Set<string>([
"stream.compose",
"sp.finished",
"sp.pipeline",
"sp.pipelineStages",
"stream.pipelineStages",
"stream.pipelineStagesDyn",
"sp.pipelineOpts",
"sp.pipelineOptsSignal",
"sp.finishedOpts",
packages/compiler/src/backend/llvm/lib-stream.ts
+59
View File
@@ -633,6 +633,65 @@ export function emitStreamLibCall(host: LlvmEmitterContext, e: LibCallExpr): LlV
host.emitPendingCheck();
return out;
}
if (
e.fn === "sp.pipelineStages" ||
e.fn === "stream.pipelineStages" ||
e.fn === "stream.pipelineStagesDyn"
) {
// (n, at, k, s1..sn, stage1..stagek, next, read, return[, cb]):
// the streams and the stage closures ride stack arrays.
const [nArg, atArg, kArg] = e.args;
if (nArg?.kind !== "numLit" || atArg?.kind !== "numLit" || kArg?.kind !== "numLit")
throw new InternalCompilerError(`llvm emitter bug: ${e.fn} counts not literals`);
const n = nArg.value;
const k = kArg.value;
const args = e.args.map((a) => host.emitExpr(a));
const packed = (from: number, count: number): string => {
const arr = B.slot();
B.entryAllocas.push(`${arr} = alloca [${count} x ptr]`);
for (let i = 0; i < count; i++) {
const p = B.tmp();
B.line(`${p} = getelementptr inbounds [${count} x ptr], ptr ${arr}, i64 0, i64 ${i}`);
B.line(`store ptr ${args[from + i]!.name}, ptr ${p}`);
}
return arr;
};
const streams = packed(3, n);
const stages = packed(3 + n, k);
const [next, read, ret] = [3 + n + k, 4 + n + k, 5 + n + k].map((i) => args[i]!.name);
const head = `double ${f64Lit(n)}, ptr ${streams}, double ${f64Lit(atArg.value)}, double ${f64Lit(k)}, ptr ${stages}, ptr ${next}, ptr ${read}, ptr ${ret}`;
const t = B.tmp();
if (e.fn === "sp.pipelineStages") {
host.declare(
`declare ptr @scr_sp_pipeline_stages(double, ptr, double, double, ptr, ptr, ptr, ptr)`,
);
B.line(`${t} = call ptr @scr_sp_pipeline_stages(${head})`);
} else if (e.fn === "stream.pipelineStagesDyn") {
host.declare(
`declare ptr @scr_stream_pipeline_stages_dyn(double, ptr, double, double, ptr, ptr, ptr, ptr, ptr)`,
);
B.line(
`${t} = call ptr @scr_stream_pipeline_stages_dyn(${head}, ptr ${args[args.length - 1]!.name})`,
);
} else {
const cbT = e.args[e.args.length - 1]!.type;
if (cbT.kind !== "func")
throw new InternalCompilerError(
"llvm emitter bug: stream.pipelineStages callback not a func",
);
host.moveTemp(args[args.length - 1]!);
const thunk = host.streamCbThunkFor("e", cbT);
host.declare(
`declare ptr @scr_stream_pipeline_stages(double, ptr, double, double, ptr, ptr, ptr, ptr, ptr, ptr)`,
);
B.line(
`${t} = call ptr @scr_stream_pipeline_stages(${head}, ptr ${args[args.length - 1]!.name}, ptr @${thunk})`,
);
}
const out = host.own({ name: t, type: e.type });
host.emitPendingCheck();
return out;
}
if (e.fn === "stream.pipeline" || e.fn === "stream.pipelineDyn") {
// pipeline(count, s1..sn, cb): the destination answers +1. The
// stream list rides a stack array (the C compound literal).
packages/compiler/src/frontend/lowering/call-signatures.ts
+4
View File
@@ -140,6 +140,8 @@ export function generatorMeta(
* - `...xs: T[]`: the ABI type is the array; call sites pack the surplus.
*/
export function paramShape(lowerer: Lowerer, param: ts.ParameterDeclaration): ParamShape {
const stage = lowerer.stageSourceType(param.name);
if (stage) return { type: stage, mode: "required" };
if (isJsSourceFile(param.getSourceFile()) && ts.isSetAccessorDeclaration(param.parent))
return { type: DYN, mode: "required" };
if (
@@ -344,6 +346,8 @@ export function paramShapes(
return params.filter((p) => !isThisParameter(p)).map((param) => lowerer.paramShape(param));
const symbols = signature.getParameters();
return params.map((declParam, i) => {
const stage = lowerer.stageSourceType(declParam.name);
if (stage) return { type: stage, mode: "required" as const };
const symbol = symbols[i];
const tsType = symbol
? lowerer.checker.getTypeOfSymbol(symbol)
packages/compiler/src/frontend/lowering/lower-async-iteration.ts
+2 -1
View File
@@ -136,7 +136,8 @@ function literalBooleanOption(
}
function nodeReadablePlan(lowerer: Lowerer, expression: ts.Expression): NodeReadablePlan | null {
const direct = lowerer.mapTypeOf(lowerer.typeOf(expression));
const direct =
lowerer.stageSourceType(expression) ?? lowerer.mapTypeOf(lowerer.typeOf(expression));
if (direct?.kind === "object") {
const sides = streamSidesOf(lowerer, lowerer.classes.get(direct.className));
if (sides === "r" || sides === "rw") {
packages/compiler/src/frontend/lowering/lower-stmts.ts
+3 -1
View File
@@ -7849,7 +7849,9 @@ export function lowerForOf(lowerer: Lowerer, stmt: ts.ForOfStatement): IrStmt {
!!stmt.awaitModifier && lowerer.mapTypeOf(lowerer.typeOf(stmt.expression))?.kind === "array";
if (stmt.awaitModifier && !awaitArray) {
{
const genT = lowerer.mapTypeOf(lowerer.typeOf(stmt.expression));
const genT =
lowerer.stageSourceType(stmt.expression) ??
lowerer.mapTypeOf(lowerer.typeOf(stmt.expression));
if (genT?.kind === "generator" && genT.async) {
const iterable = lowerer.lowerExpr(stmt.expression);
if (iterable.type.kind === "generator" && iterable.type.async) {
packages/compiler/src/frontend/lowering/lower-stream.ts
+204
View File
@@ -2026,6 +2026,23 @@ export function lowerStreamModuleCall(
? pipelineOptions(lowerer, lastArg)
: null;
const stageArgs = opts ? args.slice(0, -1) : args;
if (stageArgs.some(isPipelineFunctionStage)) {
if (opts) {
lowerer.noLowering(
"pipeline with function stages and options",
lastArg,
"pipeline(source, async function* (source) { ... }, destination) without options is supported",
);
}
const staged = lowerPipelineStages(lowerer, stageArgs, loc);
return {
kind: "libCall",
fn: "sp.pipelineStages",
args: staged.args,
type: { kind: "promise", inner: VOID },
loc,
};
}
let streams: IrExpr[];
try {
streams = stageArgs.map((a) => lowerStreamArg(lowerer, a, "pipeline"));
@@ -2200,6 +2217,22 @@ export function lowerStreamModuleCall(
"the supported form is pipeline(source, ...streams, callback)",
);
}
if (args.slice(0, -1).some(isPipelineFunctionStage)) {
const staged = lowerPipelineStages(lowerer, args.slice(0, -1), loc);
const { cb, dyn } = lowerEosCallback(
lowerer,
"pipeline",
args[args.length - 1]!,
staged.last,
);
return {
kind: "libCall",
fn: dyn ? "stream.pipelineStagesDyn" : "stream.pipelineStages",
args: [...staged.args, cb],
type: staged.last,
loc,
};
}
let streams: IrExpr[];
try {
streams = args.slice(0, -1).map((a) => lowerStreamArg(lowerer, a, "pipeline"));
@@ -2223,6 +2256,177 @@ export function lowerStreamModuleCall(
return null;
}
/** A pipeline() argument written as a function: an async generator
* function stage. */
function isPipelineFunctionStage(node: ts.Expression): boolean {
return ts.isFunctionExpression(node) || ts.isArrowFunction(node);
}
/** pipeline(s1, ..., stage1, ..., stagek, ..., sn): one run of async
* generator function stages between node streams (pipeline.js). Each
* stage's `source` parameter is typed as what the pipeline passes it:
* makeAsyncIterable answers the previous stream itself (a Readable is
* async iterable) or the previous stage's generator. Answers the libcall
* head [n, at, k, streams..., stages..., next, read, return] and the last
* stream's type. */
function lowerPipelineStages(
lowerer: Lowerer,
stageArgs: readonly ts.Expression[],
loc: SrcLoc,
): { args: IrExpr[]; last: IrType } {
const first = stageArgs.findIndex(isPipelineFunctionStage);
let end = first;
while (end < stageArgs.length && isPipelineFunctionStage(stageArgs[end]!)) end++;
if (stageArgs.slice(end).some(isPipelineFunctionStage)) {
lowerer.noLowering(
"pipeline with function stages separated by streams",
stageArgs[end]!,
"one run of function stages between two streams is supported",
);
}
if (first === 0) {
lowerer.noLowering(
"pipeline with a function as its source",
stageArgs[0]!,
"the source must be a stream; Readable.from(generator()) turns a generator into one",
);
}
if (end === stageArgs.length) {
lowerer.noLowering(
"pipeline with a function as its destination",
stageArgs[end - 1]!,
"the destination must be a stream",
);
}
const before: IrExpr[] = [];
const after: IrExpr[] = [];
try {
for (const a of stageArgs.slice(0, first)) before.push(lowerStreamArg(lowerer, a, "pipeline"));
} catch (e) {
if (e instanceof StreamArgTypeThrow)
lowerer.noLowering("pipeline over this value", stageArgs[0]!);
throw e;
}
let srcT: IrType = before[before.length - 1]!.type;
const stages: IrExpr[] = [];
for (const node of stageArgs.slice(first, end)) {
const fn = node as ts.FunctionExpression | ts.ArrowFunction;
const isAsyncGen =
ts.isFunctionExpression(fn) &&
fn.asteriskToken !== undefined &&
(fn.modifiers ?? []).some((m) => m.kind === ts.SyntaxKind.AsyncKeyword);
if (!isAsyncGen) {
lowerer.noLowering(
"pipeline stages that are not async generator functions",
fn,
"write the stage as async function* (source) { for await (const chunk of source) ... }",
);
}
if (fn.parameters.length > 1) {
lowerer.noLowering(
"pipeline stages that declare the options argument",
fn.parameters[1]!,
"the { signal } argument has no lowering — take only the source",
);
}
const lowered = lowerStageFunction(lowerer, fn, srcT);
stages.push(lowered.closure);
srcT = lowered.out;
}
try {
for (const a of stageArgs.slice(end)) after.push(lowerStreamArg(lowerer, a, "pipeline"));
} catch (e) {
if (e instanceof StreamArgTypeThrow)
lowerer.noLowering("pipeline over this value", stageArgs[end]!);
throw e;
}
const lastGen = srcT as IrType & { kind: "generator" };
if (!fromGenChunk(lowerer, lastGen.yieldT)) {
lowerer.noLowering(
`a pipeline stage yielding '${lowerer.fmt(lastGen.yieldT)}'`,
stageArgs[end - 1]!,
"stage values written to a stream are strings or Buffers",
);
}
const helpers = fromGenHelpers(lowerer, lastGen, loc);
if (!helpers)
lowerer.noLowering("a pipeline stage with this generator type", stageArgs[end - 1]!);
const streams = [...before, ...after];
return {
args: [
numLit(streams.length, loc),
numLit(before.length, loc),
numLit(stages.length, loc),
...streams,
...stages,
helpers![0]!,
helpers![1]!,
helpers![2]!,
],
last: streams[streams.length - 1]!.type,
};
}
/** An async generator stage, lowered with its `source` parameter typed
* as what the pipeline passes (a stream or the previous stage's
* generator). The parameter may only head `for await` loops. */
function lowerStageFunction(
lowerer: Lowerer,
fn: ts.FunctionExpression,
srcT: IrType,
): { closure: IrExpr; out: IrType & { kind: "generator" } } {
const param = fn.parameters[0];
const symbol =
param && ts.isIdentifier(param.name)
? lowerer.checker.getSymbolAtLocation(param.name)
: undefined;
if (param && !symbol) {
lowerer.noLowering(
"pipeline stage parameters that are patterns",
param,
"name the source parameter",
);
}
if (symbol && fn.body) {
const visit = (node: ts.Node): void => {
if (
ts.isIdentifier(node) &&
node !== param!.name &&
lowerer.checker.getSymbolAtLocation(node) === symbol
) {
const parent = node.parent;
if (
!(
parent &&
ts.isForOfStatement(parent) &&
parent.awaitModifier &&
parent.expression === node
)
) {
lowerer.noLowering(
"a pipeline stage's source used outside a for await loop head",
node,
"iterate the source with for await (const chunk of source)",
);
}
}
ts.forEachChild(node, visit);
};
visit(fn.body);
}
if (symbol) lowerer.stageSourceTypes.set(symbol, srcT);
try {
const closure = lowerer.lowerExpr(fn);
const out = closure.type.kind === "func" ? closure.type.ret : null;
if (out?.kind !== "generator" || !out.async) {
lowerer.noLowering("a pipeline stage that does not return an async generator", fn);
}
return { closure, out: out as IrType & { kind: "generator" } };
} finally {
if (symbol) lowerer.stageSourceTypes.delete(symbol);
}
}
/** `stream.compose(...streams)` over stream instances: a Duplex writing
* to the first stream and reading from the last (scr_stream_compose). The
* classes fix which positions can be readable and writable; a stream in
packages/compiler/src/frontend/lowering/lowerer.ts
+15
View File
@@ -1801,6 +1801,11 @@ export class Lowerer {
* prepass. This includes explicit annotations: unchecked TS annotations do
* not prove that an array property read produced a value. */
readonly runtimeOptionalBindingTypes = new Map<ts.Symbol, IrType>();
/** A pipeline() function stage's `source` parameter: the IR type of what
* the pipeline passes it (the previous stage as an async iterable), set
* while the stage lowers. Parameter typing and `for await` heads read
* it before the checker's `AsyncIterable<T>`. */
readonly stageSourceTypes = new Map<ts.Symbol, IrType>();
/** Set when a match result lowers to execGroups/matchGroups, the inputs
* of regex span folding. */
regexGroupResults = false;
@@ -9029,6 +9034,16 @@ export class Lowerer {
return paramShape(this, param);
}
/** The pipeline stage source type bound to `node` (a parameter name or
* a reference to it), or null. */
stageSourceType(node: ts.Node): IrType | null {
if (this.stageSourceTypes.size === 0) return null;
while (ts.isParenthesizedExpression(node)) node = node.expression;
if (!ts.isIdentifier(node)) return null;
const symbol = this.checker.getSymbolAtLocation(node);
return (symbol && this.stageSourceTypes.get(symbol)) ?? null;
}
checkDefaultParamBodyType(param: ts.ParameterDeclaration, bodyType: IrType): void {
return checkDefaultParamBodyType(this, param, bodyType);
}
packages/compiler/src/ir/builtin-effects.ts
+3
View File
@@ -245,6 +245,9 @@ export const MAY_THROW_LIB_FNS: ReadonlySet<IrLibFn> = new Set([
"stream.compose",
"sp.finished",
"sp.pipeline",
"sp.pipelineStages",
"stream.pipelineStages",
"stream.pipelineStagesDyn",
"sc.text",
"sc.json",
"sc.buffer",
packages/compiler/src/ir/builtin-signatures.ts
+7
View File
@@ -3207,6 +3207,8 @@ export const LIB_FN_SIGS = defineLibFnSignatures({
"stream.finishedDyn": { argTypes: [null, DYN], result: VOID },
"stream.pipeline": { argTypes: [F64], result: VOID },
"stream.pipelineDyn": { argTypes: [F64], result: VOID },
"stream.pipelineStages": { argTypes: [F64, F64, F64], result: VOID },
"stream.pipelineStagesDyn": { argTypes: [F64, F64, F64], result: VOID },
/** stream.compose(count, s1..sn): the Duplex over a pipeline of the
* streams (+1). MAY THROW (ERR_INVALID_ARG_VALUE for a stream that is
* no longer readable or writable). */
@@ -3237,6 +3239,11 @@ export const LIB_FN_SIGS = defineLibFnSignatures({
"stream.finishedOpts": { argTypes: [null, F64, F64, BOOL, null], result: VOID },
"stream.finishedOptsSignal": { argTypes: [null, F64, F64, BOOL, DYN, null], result: VOID },
"sp.pipeline": { argTypes: [F64], result: { kind: "promise", inner: VOID } },
/** pipeline with a run of async generator stages: [n, at, k, s1..sn,
* stage1..stagek, next, read, return] (+ the callback for the stream.*
* forms): the user's stage functions, then the frontend's helpers over
* the last stage's generator (scr_stream_pipeline_stages). */
"sp.pipelineStages": { argTypes: [F64, F64, F64], result: { kind: "promise", inner: VOID } },
/** node:stream/consumers — the promise consumers over the readable
* machinery: (s) → a pending promise settled at the terminal point
* with the accumulated result (sc.text: the utf8 decode, sc.json: the
packages/compiler/src/ir/runtime-features.ts
+2
View File
@@ -221,6 +221,8 @@ function scanRuntimeFeatures(mod: IrModule, stopAt?: keyof RuntimeFeatures): Run
fn === "stream.errored" ||
fn === "sp.finished" ||
fn === "sp.pipeline" ||
fn === "sp.pipelineStages" ||
fn.startsWith("stream.pipelineStages") ||
fn === "console.bind" ||
fn === "stream.addAbortSignal" ||
fn === "stream.addAbortSignalFn" ||
packages/compiler/src/ir/validate.ts
+27
View File
@@ -3961,6 +3961,9 @@ function validateFunction(
e.fn === "stream.pipeline" ||
e.fn === "stream.pipelineDyn" ||
e.fn === "stream.compose" ||
e.fn === "stream.pipelineStages" ||
e.fn === "stream.pipelineStagesDyn" ||
e.fn === "sp.pipelineStages" ||
e.fn === "sp.pipeline" ||
e.fn === "sp.pipelineOpts" ||
e.fn === "sp.pipelineOptsSignal" ||
@@ -5222,6 +5225,8 @@ function validateFunction(
e.fn === "stream.pipeline" ||
e.fn === "stream.pipelineDyn" ||
e.fn === "stream.compose" ||
e.fn === "stream.pipelineStages" ||
e.fn === "stream.pipelineStagesDyn" ||
e.fn === "stream.addAbortSignal" ||
e.fn === "stream.addAbortSignalFn" ||
e.fn === "stream.finishedOpts" ||
@@ -5328,6 +5333,28 @@ function validateFunction(
if (!isStreamObject(e.type)) err(`libCall ${e.fn} must return a stream`, e.loc);
return true;
}
if (e.fn === "stream.pipelineStages" || e.fn === "stream.pipelineStagesDyn") {
// (n, at, k, s1..sn, stage1..stagek, next, read, return, cb)
const [n, at, k] = e.args.slice(0, 3).map((a) => (a?.kind === "numLit" ? a.value : -1));
if (n! < 2 || at! < 1 || at! >= n! || k! < 1 || e.args.length !== 3 + n! + k! + 3 + 1) {
err(`libCall ${e.fn} count/arity mismatch`, e.loc);
return true;
}
for (let i = 3; i < 3 + n!; i++) {
if (!isStreamObject(e.args[i]?.type)) {
err(`libCall ${e.fn} stage ${i - 3} must be a stream-hierarchy object`, e.loc);
}
}
for (let i = 3 + n!; i < e.args.length - 1; i++) {
if (e.args[i]?.type.kind !== "func")
err(`libCall ${e.fn} helper ${i} must be a func`, e.loc);
}
const cbK = e.args[e.args.length - 1]?.type.kind;
if (e.fn === "stream.pipelineStages" ? cbK !== "func" : cbK !== "dyn") {
err(`libCall ${e.fn} callback has the wrong kind`, e.loc);
}
return true;
}
if (e.fn === "stream.pipeline" || e.fn === "stream.pipelineDyn") {
// (count, s1..sn, cb) → the destination's type.
const count = e.args[0];
packages/compiler/test/ts7/baselines/order-parity/tests/corpus/5425-pipeline-generator-stage.ts.json
+6
View File
@@ -0,0 +1,6 @@
{
"order": [
"<repo>/tests/corpus/5425-pipeline-generator-stage.ts"
],
"diags": []
}
packages/compiler/test/ts7/baselines/order-parity/tests/corpus/5426-pipeline-generator-errors.ts.json
+6
View File
@@ -0,0 +1,6 @@
{
"order": [
"<repo>/tests/corpus/5426-pipeline-generator-errors.ts"
],
"diags": []
}
packages/compiler/test/ts7/baselines/order-parity/tests/corpus/5427-pipeline-generator-backpressure.ts.json
+6
View File
@@ -0,0 +1,6 @@
{
"order": [
"<repo>/tests/corpus/5427-pipeline-generator-backpressure.ts"
],
"diags": []
}
packages/compiler/test/ts7/baselines/order-parity/tests/corpus/5428-pipeline-generator-stages.cjs.json
+6
View File
@@ -0,0 +1,6 @@
{
"order": [
"<repo>/tests/corpus/5428-pipeline-generator-stages.cjs"
],
"diags": []
}
packages/compiler/test/ts7/baselines/order-parity/tests/corpus/5429-pipeline-generator-buffers.ts.json
+6
View File
@@ -0,0 +1,6 @@
{
"order": [
"<repo>/tests/corpus/5429-pipeline-generator-buffers.ts"
],
"diags": []
}
packages/runtime/src/scr_stream.c
+368 -11
View File
@@ -2992,9 +2992,310 @@ static void scr_pipe_throw(const char *code, const char *msg, int kind) {
scr_throw_obj(e, &scr_error_retain_v, &scr_error_release_v, scr_error_trace_arg());
}
/* The generators' done marker (Readable.from's and the pipeline stages'
* helpers answer it): a process-lived object compared by identity. */
static SCR_TL ScrDyn *scr_stream_done_marker = NULL;
static void scr_stream_done_marker_free(void) {
scr_dyn_release(scr_stream_done_marker);
scr_stream_done_marker = NULL;
}
ScrDyn *scr_stream_from_done(void) {
if (!scr_stream_done_marker) {
scr_stream_done_marker = scr_dyn_new_obj();
scr_atexit(scr_stream_done_marker_free);
}
return scr_dyn_retain(scr_stream_done_marker);
}
/* ── pipeline function stages (pipeline.js: makeAsyncIterable, pumpToNode) ──
* A run of async generator functions between two node streams. Each
* stage is called while the pipeline is set up, in pipeline order:
* makeAsyncIterable hands the first one the previous stream itself (a
* Readable is async iterable) and each later one the previous stage's
* generator. The last generator is pumped into the stream after the run
* with pumpToNode. The stages and helpers use the closure ABI (the
* closure first; the callee owns its parameters). */
typedef ScrGen *(*ScrPipeFirstFn)(ScrClosure *, ScrStream *);
typedef ScrGen *(*ScrPipeStageFn)(ScrClosure *, ScrGen *);
typedef ScrPromise *(*ScrPipeNextFn)(ScrClosure *, ScrGen *);
typedef ScrDyn *(*ScrPipeReadFn)(ScrClosure *, void *);
typedef struct {
size_t at; /* streams[0..at-1] come before the stages */
size_t count; /* the function stages */
ScrClosure **stages; /* (stream or async generator) => async generator */
ScrClosure *next, *read, *ret; /* the last generator's helpers */
} ScrPipeStages;
enum { SCR_PUMP_DRAIN0, SCR_PUMP_WRITE, SCR_PUMP_END };
typedef struct {
size_t rc;
ScrStreamPipeCtx *ctx; /* +1 */
ScrGen *gen; /* +1 */
ScrClosure *next, *read, *ret; /* +1 */
ScrStream *w; /* +1 */
bool end;
ScrError *error; /* +1: what resume() recorded, or NULL */
ScrPromise *waiter; /* +1: the pending wait(), or NULL */
int stage; /* what the pending wait() resumes */
ScrClosure *drain_l; /* +1 */
ScrClosure *eos_cleanup; /* +1 */
ScrError *fail; /* +1: the loop's error while return() runs */
} ScrPump;
static void *scr_pump_retain_v(void *p) {
((ScrPump *)p)->rc++;
return p;
}
static void scr_pump_release_v(void *p) {
ScrPump *pm = p;
if (--pm->rc > 0) return;
scr_stream_pipectx_release_v(pm->ctx);
scr_gen_release(pm->gen);
scr_closure_release(pm->next);
scr_closure_release(pm->read);
scr_closure_release(pm->ret);
scr_stream_release(pm->w);
if (pm->error) scr_error_release(pm->error);
if (pm->waiter) scr_promise_release(pm->waiter);
if (pm->drain_l) scr_closure_release(pm->drain_l);
if (pm->eos_cleanup) scr_closure_release(pm->eos_cleanup);
if (pm->fail) scr_error_release(pm->fail);
free(pm);
}
static ScrClosure *scr_pump_closure(void *fn, ScrPump *pm) {
ScrClosure *c = scr_closure_new(fn, 1);
c->caps[0] = scr_box_new_obj(&scr_pump_retain_v, &scr_pump_release_v, NULL);
scr_box_set_ref(c->caps[0], scr_pump_retain_v(pm));
return c;
}
/* A rejection as a pipeline error: Error values as themselves. +1. */
static ScrError *scr_pump_caught(void) {
ScrCaught *c = scr_exc_take();
ScrDyn *d = scr_caught_to_dyn(c);
scr_caught_release(c);
ScrError *e = d ? scr_errdyn_err_of(d) : NULL;
if (d) scr_dyn_release(d);
if (e) return e;
return scr_stream_mkerr("ERR_INTERNAL_ASSERTION",
"a pipeline stage rejection that is not an Error object is not supported yet");
}
static void scr_pump_loop(ScrPump *pm);
static void scr_pump_wait(ScrPump *pm, int stage);
static void scr_pump_fail_in_loop(ScrPump *pm, ScrError *err);
/* finish(err) and the finally block: the eos cleanup, then the 'drain'
* listener. Moves err. */
static void scr_pump_done(ScrPump *pm, ScrError *err) {
ScrStreamPipeCtx *ctx = pm->ctx;
scr_pipe_finish_impl(ctx, err, --ctx->finish_count == 0);
if (err) scr_error_release(err);
if (pm->eos_cleanup) {
ScrClosure *c = pm->eos_cleanup;
pm->eos_cleanup = NULL;
((void (*)(ScrClosure *))c->fn)(c);
scr_closure_release(c);
}
if (pm->drain_l) {
ScrStr *n = scr_str_new("drain", 5);
scr_emitter_release(scr_emitter_off((ScrEmitter *)pm->w, n, pm->drain_l));
scr_str_release(n);
scr_closure_release(pm->drain_l);
pm->drain_l = NULL;
}
}
/* resume(err): the 'drain' listener and the eos callback. */
static void scr_pump_resume(ScrPump *pm, ScrError *err) {
if (err && !pm->error) pm->error = scr_error_retain(err);
ScrPromise *w = pm->waiter;
if (!w) return;
pm->waiter = NULL;
if (pm->error) {
scr_throw_obj(scr_error_retain(pm->error), &scr_error_retain_v, &scr_error_release_v,
scr_error_trace_arg());
scr_promise_reject_pending(w);
} else {
scr_promise_fulfill_void(w);
}
scr_promise_release(w);
}
static void scr_pump_drain_l(ScrClosure *c) {
ScrPump *pm = scr_box_get_ref(c->caps[0]);
scr_pump_resume(pm, NULL);
scr_pump_release_v(pm);
}
static void scr_pump_eos(ScrClosure *c, ScrStream *s, ScrError *err) {
(void)s;
ScrPump *pm = scr_box_get_ref(c->caps[0]);
scr_pump_resume(pm, err);
scr_pump_release_v(pm);
}
/* `await p` for the pump: kind 0 a wait(), 1 the generator's next(),
* 2 its return(). */
typedef struct {
ScrPump *pm; /* +1 */
ScrPromise *p; /* +1 */
int kind;
int stage;
} ScrPumpAwait;
static void scr_pump_after_wait(ScrPump *pm, int stage, bool rejected);
static void scr_pump_after_next(ScrPump *pm, ScrDyn *v);
static void scr_pump_after_return(ScrPump *pm);
static void scr_pump_await_entry(ScrFiber *fiber, void *arg) {
(void)fiber;
ScrPumpAwait a = *(ScrPumpAwait *)arg;
free(arg);
if (a.kind == 1) {
void *rec = scr_await_ref(a.p);
ScrDyn *v = scr_exc_pending() ? NULL : ((ScrPipeReadFn)a.pm->read->fn)(a.pm->read, rec);
scr_pump_after_next(a.pm, v);
} else {
scr_await_void(a.p);
if (a.kind == 2) {
if (scr_exc_pending()) scr_exc_clear(); /* the loop's error wins */
scr_pump_after_return(a.pm);
} else {
scr_pump_after_wait(a.pm, a.stage, scr_exc_pending());
}
}
if (scr_exc_pending()) scr_exc_clear();
scr_promise_release(a.p);
scr_pump_release_v(a.pm);
}
static void scr_pump_await(ScrPump *pm, ScrPromise *p /*borrowed*/, int kind, int stage) {
ScrPumpAwait *a = malloc(sizeof *a);
if (!a) scr_stream_oom();
a->pm = scr_pump_retain_v(pm);
a->p = scr_promise_retain(p);
a->kind = kind;
a->stage = stage;
scr_promise_release(scr_async_spawn_after(p, &scr_pump_await_entry, a));
}
/* `await wait()`: rejects at once with a recorded error, else waits for
* resume(). */
static void scr_pump_wait(ScrPump *pm, int stage) {
ScrPromise *p = scr_promise_new();
if (pm->error) {
scr_throw_obj(scr_error_retain(pm->error), &scr_error_retain_v, &scr_error_release_v,
scr_error_trace_arg());
scr_promise_reject_pending(p);
} else {
pm->waiter = scr_promise_retain(p);
}
scr_pump_await(pm, p, 0, stage);
scr_promise_release(p);
}
/* `for await (const chunk of iterable)`: one next() per pass. */
static void scr_pump_loop(ScrPump *pm) {
ScrPromise *p = ((ScrPipeNextFn)pm->next->fn)(pm->next, scr_gen_retain(pm->gen));
scr_pump_await(pm, p, 1, 0);
scr_promise_release(p);
}
static void scr_pump_after_next(ScrPump *pm, ScrDyn *v) {
if (!v) {
/* the generator threw: catch (err) */
scr_pump_done(pm, scr_pump_caught());
return;
}
if (v == scr_stream_done_marker) {
scr_dyn_release(v);
if (pm->end) {
scr_stream_release(scr_stream_end(pm->w, NULL, NULL, NULL));
if (scr_exc_pending()) {
scr_pump_done(pm, scr_pump_caught());
return;
}
scr_pump_wait(pm, SCR_PUMP_END);
return;
}
scr_pump_done(pm, NULL);
return;
}
bool ok = scr_stream_write_dyn(pm->w, v, NULL);
scr_dyn_release(v);
if (scr_exc_pending()) {
scr_pump_fail_in_loop(pm, scr_pump_caught());
return;
}
if (!ok) {
scr_pump_wait(pm, SCR_PUMP_WRITE);
return;
}
scr_pump_loop(pm);
}
/* An error inside the loop body: for-await's AsyncIteratorClose awaits
* the generator's return() before the catch block runs. Moves err. */
static void scr_pump_fail_in_loop(ScrPump *pm, ScrError *err) {
pm->fail = err;
ScrPromise *p = ((ScrPipeNextFn)pm->ret->fn)(pm->ret, scr_gen_retain(pm->gen));
scr_pump_await(pm, p, 2, 0);
scr_promise_release(p);
}
static void scr_pump_after_return(ScrPump *pm) {
ScrError *err = pm->fail;
pm->fail = NULL;
scr_pump_done(pm, err);
}
static void scr_pump_after_wait(ScrPump *pm, int stage, bool rejected) {
ScrError *err = rejected ? scr_pump_caught() : NULL;
if (stage == SCR_PUMP_WRITE) {
if (err) scr_pump_fail_in_loop(pm, err);
else scr_pump_loop(pm);
} else if (stage == SCR_PUMP_DRAIN0) {
if (err) scr_pump_done(pm, err);
else scr_pump_loop(pm);
} else {
scr_pump_done(pm, err);
}
}
/* pumpToNode(gen, w, finish, { end }): starts at once, as the async
* function does. The helpers are borrowed. */
static void scr_pump_start(ScrStreamPipeCtx *ctx, ScrGen *gen /*moves*/, const ScrPipeStages *fs,
ScrStream *w, bool end) {
ScrPump *pm = calloc(1, sizeof *pm);
if (!pm) scr_stream_oom();
pm->rc = 1;
pm->ctx = scr_stream_pipectx_retain_v(ctx);
pm->gen = gen;
pm->next = scr_closure_retain(fs->next);
pm->read = scr_closure_retain(fs->read);
pm->ret = scr_closure_retain(fs->ret);
pm->w = scr_stream_retain(w);
pm->end = end;
pm->drain_l = scr_pump_closure((void *)&scr_pump_drain_l, pm);
scr_pipe_on(w, "drain", scr_closure_retain(pm->drain_l), scr_ee_inv_fixed0, false);
pm->eos_cleanup = scr_pipe_eos(w, scr_pump_closure((void *)&scr_pump_eos, pm), &scr_pump_eos, false,
w->st->has_w);
if (w->st->has_w && w->st->w.need_drain) scr_pump_wait(pm, SCR_PUMP_DRAIN0);
else scr_pump_loop(pm);
scr_pump_release_v(pm);
}
static ScrStream *scr_stream_pipeline_ex(size_t n, ScrStream **streams, ScrClosure *cb,
ScrStreamErrInv inv, bool end_opt, ScrDyn *signal,
ScrAbortAddFn add) {
ScrAbortAddFn add, const ScrPipeStages *fs) {
ScrStreamPipeCtx *ctx = calloc(1, sizeof *ctx);
if (!ctx) scr_stream_oom();
ctx->rc = 1;
@@ -3026,14 +3327,35 @@ static ScrStream *scr_stream_pipeline_ex(size_t n, ScrStream **streams, ScrClosu
ctx->dispose = add(signal, listener);
scr_dyn_release(listener);
}
/* pipelineImpl's loop over the stages: the n streams, with fs->count
* function stages before streams[fs->at] */
size_t stages = n + (fs ? fs->count : 0);
size_t fn_first = fs ? fs->at : stages, fn_end = fs ? fs->at + fs->count : stages;
ScrStream *ret = NULL;
for (size_t i = 0; i < n; i++) {
ScrStream *s = streams[i];
bool reading = i + 1 < n;
bool writing = i > 0;
ScrGen *ret_gen = NULL; /* +1: ret when the previous stage was a function */
size_t i = 0; /* the next stream */
for (size_t j = 0; j < stages; j++) {
bool reading = j + 1 < stages;
bool end = reading || end_opt;
bool last = i + 1 == n;
if (reading && (streams[i + 1]->st->close_emitted || streams[i + 1]->st->destroyed)) {
bool last = j + 1 == stages;
if (j >= fn_first && j < fn_end) {
/* ret = stage(makeAsyncIterable(ret)) */
ScrClosure *stage = fs->stages[j - fn_first];
ScrGen *src = ret_gen;
ret_gen = src ? ((ScrPipeStageFn)stage->fn)(stage, src)
: ((ScrPipeFirstFn)stage->fn)(stage, scr_stream_retain(ret));
if (scr_exc_pending()) {
scr_gen_release(ret_gen);
scr_stream_pipectx_drop(ctx);
scr_stream_pipectx_release_v(ctx);
return NULL;
}
continue;
}
ScrStream *s = streams[i];
bool writing = j > 0;
if (reading && !(j + 1 >= fn_first && j + 1 < fn_end) &&
(streams[i + 1]->st->close_emitted || streams[i + 1]->st->destroyed)) {
scr_pipe_throw("ERR_STREAM_UNABLE_TO_PIPE", "Cannot pipe to a closed or destroyed stream",
SCR_ERR_ERROR);
scr_stream_pipectx_drop(ctx);
@@ -3054,7 +3376,13 @@ static ScrStream *scr_stream_pipeline_ex(size_t n, ScrStream **streams, ScrClosu
scr_pipe_on(s, "error", scr_closure_retain(on_error), scr_ee_inv_fixed1, false);
if (last_readable) ctx->last_error_l = on_error;
else scr_closure_release(on_error);
if (i > 0) {
if (ret_gen) {
/* an async iterable before a stream: pumpToNode(ret, s, finish, { end }) */
ctx->finish_count++;
ScrGen *g = ret_gen;
ret_gen = NULL;
scr_pump_start(ctx, g, fs, s, end);
} else if (j > 0) {
if (!ret->st->has_r) {
static const char msg[] =
"The \"val\" argument must be an instance of Readable, Iterable, AsyncIterable, "
@@ -3084,15 +3412,33 @@ static ScrStream *scr_stream_pipeline_ex(size_t n, ScrStream **streams, ScrClosu
else scr_closure_release(cleanup);
}
ret = s;
i++;
}
if (aborted) scr_next_tick(scr_pipe_closure((void *)&scr_pipe_abort_tick, ctx, 0));
scr_stream_pipectx_release_v(ctx); /* the listeners and the live list hold it */
return scr_stream_retain(ret);
}
/* pipeline(...) with a run of async generator stages before streams[at]:
* the helpers are borrowed (see ScrPipeStages). */
ScrStream *scr_stream_pipeline_stages(double n, ScrStream **streams, double at, double count,
ScrClosure **stages, ScrClosure *next, ScrClosure *read,
ScrClosure *ret, ScrClosure *cb /*moves*/, ScrStreamErrInv inv) {
ScrPipeStages fs = {(size_t)at, (size_t)count, stages, next, read, ret};
return scr_stream_pipeline_ex((size_t)n, streams, cb, inv, true, NULL, NULL, &fs);
}
ScrStream *scr_stream_pipeline_stages_dyn(double n, ScrStream **streams, double at, double count,
ScrClosure **stages, ScrClosure *next, ScrClosure *read,
ScrClosure *ret, ScrDyn *cb) {
ScrClosure *c = scr_stream_dynopt_clo(cb, (void *)&scr_stream_finished_dyn_inv);
return scr_stream_pipeline_stages(n, streams, at, count, stages, next, read, ret, c,
&scr_stream_finished_dyn_inv);
}
ScrStream *scr_stream_pipeline(double n_d, ScrStream **streams /*borrowed*/,
ScrClosure *cb /*moves*/, ScrStreamErrInv inv) {
return scr_stream_pipeline_ex((size_t)n_d, streams, cb, inv, true, NULL, NULL);
return scr_stream_pipeline_ex((size_t)n_d, streams, cb, inv, true, NULL, NULL, NULL);
}
/* The dyn-valued twins: the callback rides as a checked-dynamic value
@@ -3360,7 +3706,7 @@ ScrStream *scr_stream_compose(double n_d, ScrStream **streams) {
c->tail = scr_stream_retain(streams[n - 1]);
c->onclose = SCR_COMPOSE_RUNNING;
ScrStream *tail = scr_stream_pipeline_ex(n, streams, scr_compose_closure((void *)&scr_compose_finished, c),
&scr_compose_finished, true, NULL, NULL);
&scr_compose_finished, true, NULL, NULL, NULL);
if (!tail) {
scr_compose_release_v(c);
return NULL;
@@ -3444,6 +3790,17 @@ ScrPromise *scr_sp_pipeline(double n, ScrStream **streams) {
return p;
}
ScrPromise *scr_sp_pipeline_stages(double n, ScrStream **streams, double at, double count,
ScrClosure **stages, ScrClosure *next, ScrClosure *read,
ScrClosure *ret) {
ScrPromise *p = scr_promise_new();
ScrStream *dst = scr_stream_pipeline_stages(n, streams, at, count, stages, next, read, ret,
scr_sp_watcher(p), &scr_sp_settle_inv);
if (dst) scr_stream_release(dst);
else scr_promise_reject_pending(p);
return p;
}
/* pipeline(...streams, { signal, end }): end false leaves the destination
* open (the promise settles once the stages before it finish); signal is
* NULL or borrowed — a value that is not an AbortSignal rejects with
@@ -3465,7 +3822,7 @@ ScrPromise *scr_sp_pipeline_opts(double n, ScrStream **streams, bool end, ScrDyn
scr_dyn_release(aborted);
}
ScrStream *dst = scr_stream_pipeline_ex((size_t)n, streams, scr_sp_watcher(p),
&scr_sp_settle_inv, end, signal, add);
&scr_sp_settle_inv, end, signal, add, NULL);
if (dst) scr_stream_release(dst);
else scr_promise_reject_pending(p);
return p;
packages/runtime/src/scr_stream.h
+17 -1
View File
@@ -171,6 +171,22 @@ ScrStream *scr_stream_pipeline(double n, ScrStream **streams, ScrClosure *cb, Sc
* mustCall wrappers; borrowed). */
ScrClosure *scr_stream_finished_dyn(ScrStream *s, struct ScrDyn *cb);
ScrStream *scr_stream_pipeline_dyn(double n, ScrStream **streams, struct ScrDyn *cb);
/* pipeline(s1.., stage1..stagek, ..sn[, cb]) with a run of async generator
* function stages before streams[at] (pipeline.js): the first stage gets
* streams[at-1] itself, each later one the previous stage's generator,
* and the last generator is pumped into streams[at] (pumpToNode) through
* its helpers: next/ret answer the resume's promise, read maps its
* IteratorResult to the value or scr_stream_from_done. Streams and
* closures borrowed; cb moves. */
ScrStream *scr_stream_pipeline_stages(double n, ScrStream **streams, double at, double count,
ScrClosure **stages, ScrClosure *next, ScrClosure *read,
ScrClosure *ret, ScrClosure *cb, ScrStreamErrInv inv);
ScrStream *scr_stream_pipeline_stages_dyn(double n, ScrStream **streams, double at, double count,
ScrClosure **stages, ScrClosure *next, ScrClosure *read,
ScrClosure *ret, struct ScrDyn *cb);
struct ScrPromise *scr_sp_pipeline_stages(double n, ScrStream **streams, double at, double count,
ScrClosure **stages, ScrClosure *next, ScrClosure *read,
ScrClosure *ret);
/* stream.compose(s1..sn): a pipeline over the streams and a Duplex that
* writes to the first and reads from the last (compose.js). One stream
* answers itself (Duplex.from of a duplex). +1, or NULL with
@@ -293,7 +309,7 @@ ScrStream *scr_stream_from_gen(struct ScrGen *g, ScrClosure *next, ScrClosure *r
ScrStream *scr_stream_from_async_gen(struct ScrGen *g, ScrClosure *next, ScrClosure *read,
ScrClosure *ret, ScrClosure *thr, double flags,
double hwm);
ScrDyn *scr_stream_from_done(void); /* +1 */
ScrDyn *scr_stream_from_done(void); /* +1, the process-lived marker (scr_stream.c) */
ScrStream *scr_stream_pause(ScrStream *s); /* recv +1 */
ScrStream *scr_stream_resume(ScrStream *s); /* recv +1 */
bool scr_stream_is_paused(ScrStream *s);
packages/runtime/src/scr_stream_ops.c
+1 -5
View File
@@ -78,7 +78,7 @@ static void op_sentinels_init(void) {
if (op_k_empty) return;
op_k_empty = scr_dyn_new_obj();
op_k_eof = scr_dyn_new_obj();
op_k_done = scr_dyn_new_obj();
op_k_done = scr_stream_from_done(); /* shared with pumpToNode (scr_stream.c) */
scr_atexit(op_sentinels_free);
}
@@ -1726,10 +1726,6 @@ ScrStream *scr_stream_from_async_gen(ScrGen *g, ScrClosure *next, ScrClosure *re
return op_from_compiled(FROM_ASYNC, g, next, read, ret, thr, flags, hwm);
}
ScrDyn *scr_stream_from_done(void) {
op_sentinels_init();
return scr_dyn_retain(op_k_done);
}
/* ── the promise operators ─────────────────────────────────────────── */
tests/corpus/5425-pipeline-generator-stage.ts
+35
View File
@@ -0,0 +1,35 @@
// pipeline() with an async generator stage between two streams: the stage
// iterates the source stream itself, and its values reach the destination
// on the same microtasks as in Node.
import { Readable, Writable, pipeline } from "node:stream";
const log: string[] = [];
let k = 0;
const chain = (): void => {
if (k++ < 40) {
log.push(`m${k}`);
Promise.resolve().then(chain);
}
};
const src = Readable.from(["a", "b"]);
const sink = new Writable({
write(c: Buffer, _e, cb) {
log.push(`write ${c.toString()}`);
cb();
},
});
Promise.resolve().then(chain);
pipeline(
src,
async function* (source: AsyncIterable<string>) {
for await (const c of source) {
log.push(`stage ${c}`);
yield c.toUpperCase();
}
},
sink,
(err?: Error | null) => {
log.push(`done ${err ? err.message : "ok"}`);
console.log(log.join(" "));
},
);
tests/corpus/5426-pipeline-generator-errors.ts
+118
View File
@@ -0,0 +1,118 @@
// pipeline() generator stages: two chained stages, the promise form with a
// slow sink, a stage that throws, a destination that fails (the stage is
// returned and its finally runs), and a source that errors.
import { PassThrough, Readable, Writable, pipeline } from "node:stream";
import { pipeline as pipelineP } from "node:stream/promises";
const log: string[] = [];
let hop = 0;
const chain = (): void => {
if (hop++ < 60) {
log.push(`h${hop}`);
Promise.resolve().then(chain);
}
};
const collect = (label: string, hwm = 16384): Writable =>
new Writable({
highWaterMark: hwm,
write(c: Buffer, _e, cb) {
log.push(`${label} write ${c.toString()}`);
if (hwm === 1) setTimeout(cb, 1);
else cb();
},
});
Promise.resolve().then(chain);
// two chained stages
await new Promise<void>((resolve) =>
pipeline(
Readable.from(["a", "b", "c"]),
async function* (source: AsyncIterable<string>) {
for await (const c of source) yield c + "1";
},
async function* (source: AsyncIterable<string>) {
for await (const c of source) {
log.push(`second ${c}`);
yield c + "2";
}
},
collect("chained"),
(err?: Error | null) => {
log.push(`chained done ${err ? err.message : "ok"}`);
resolve();
},
),
);
// the promise form, with backpressure into a slow sink
await pipelineP(
Readable.from(["x", "y", "z"]),
async function* (source: AsyncIterable<string>) {
for await (const c of source) yield c.repeat(3);
},
collect("slow", 1),
);
log.push("promise form done");
// a stage that throws
await new Promise<void>((resolve) =>
pipeline(
Readable.from(["p", "q"]),
async function* (source: AsyncIterable<string>) {
try {
for await (const c of source) {
if (c === "q") throw new Error("stage failed");
yield c;
}
} finally {
log.push("thrower finally");
}
},
collect("thrower"),
(err?: Error | null) => {
log.push(`thrower done ${err ? err.message : "ok"}`);
resolve();
},
),
);
// a destination that fails: the stage's generator is returned
const failing = new Writable({
write(_c: Buffer, _e, cb) {
cb(new Error("sink failed"));
},
});
await new Promise<void>((resolve) =>
pipeline(
Readable.from(["u", "v", "w"]),
async function* (source: AsyncIterable<string>) {
try {
for await (const c of source) yield c;
} finally {
log.push("returned stage finally");
}
},
failing,
(err?: Error | null) => {
log.push(`sink done ${err ? err.message : "ok"}`);
resolve();
},
),
);
// a source that errors
const bad = new Readable({
read() {
this.destroy(new Error("source failed"));
},
});
await new Promise<void>((resolve) =>
pipeline(
bad,
async function* (source: AsyncIterable<Buffer>) {
for await (const c of source) yield c;
},
new PassThrough(),
(err?: Error | null) => {
log.push(`source done ${err ? err.message : "ok"}`);
resolve();
},
),
);
console.log(log.join("\n"));
tests/corpus/5427-pipeline-generator-backpressure.ts
+42
View File
@@ -0,0 +1,42 @@
// pipeline() pumps a generator stage into a highWaterMark: 1 sink: each
// write waits for drain, timed against microtasks and ticks.
import { Readable, Writable } from "node:stream";
import { pipeline } from "node:stream/promises";
const log: string[] = [];
let hop = 0;
const chain = (): void => {
if (hop++ < 40) {
log.push(`h${hop}`);
Promise.resolve().then(chain);
}
};
let tick = 0;
const ticks = (): void => {
if (tick++ < 10) {
log.push(`t${tick}`);
process.nextTick(ticks);
}
};
const sink = new Writable({
highWaterMark: 1,
write(c: Buffer, _e, cb) {
log.push(`write ${c.toString()} ${sink.writableLength}`);
Promise.resolve().then(() => cb());
},
});
sink.on("drain", () => log.push("drain"));
Promise.resolve().then(chain);
process.nextTick(ticks);
await pipeline(
Readable.from(["1", "2", "3", "4"]),
async function* (source: AsyncIterable<string>) {
for await (const c of source) {
log.push(`stage ${c}`);
yield c + c;
}
},
sink,
);
log.push("done");
console.log(log.join(" "));
tests/corpus/5428-pipeline-generator-stages.cjs
+28
View File
@@ -0,0 +1,28 @@
// @no-node-shims
// pipeline() generator stages from CommonJS: untyped stages over Buffer
// chunks, the promise and callback forms, and the returned destination.
'use strict';
const { Readable, PassThrough, Writable, pipeline } = require('node:stream');
const { pipeline: pipelineP } = require('node:stream/promises');
async function main() {
const seen = [];
const sink = new Writable({ write(c, e, cb) { seen.push(String(c)); cb(); } });
await pipelineP(
Readable.from([Buffer.from('ab'), Buffer.from('cd')]),
async function* (source) {
for await (const chunk of source) yield String(chunk).toUpperCase();
},
async function* (source) {
for await (const chunk of source) yield chunk + '|';
},
sink,
);
console.log('seen', seen.join(''));
const out = new PassThrough();
const ret = pipeline(Readable.from(['x']), async function* (s) { for await (const c of s) yield c + c; }, out, (err) => {
console.log('cb', err === undefined ? 'undefined' : err);
});
console.log('returns destination', ret === out);
out.on('data', (c) => console.log('out', String(c)));
}
main();
tests/corpus/5429-pipeline-generator-buffers.ts
+43
View File
@@ -0,0 +1,43 @@
// pipeline() with a stage typed AsyncIterable<Buffer> between streams, more
// streams after it, and the callback form's return value.
import { PassThrough, Readable, Transform, Writable, pipeline } from "node:stream";
const log: string[] = [];
const upper = new Transform({
transform(c: Buffer, _e, cb) {
cb(null, c.toString().toUpperCase());
},
});
const sink = new Writable({
write(c: Buffer, _e, cb) {
log.push(`sink ${c.toString()}`);
cb();
},
});
const source = new Readable({
read() {
this.push(Buffer.from("hello "));
this.push(Buffer.from("world"));
this.push(null);
},
});
const ret = pipeline(
source,
new PassThrough(),
async function* (chunks: AsyncIterable<Buffer>) {
let total = 0;
for await (const chunk of chunks) {
total += chunk.length;
log.push(`stage ${chunk.length} ${chunk.toString("hex")}`);
yield chunk.subarray(0, 3);
}
log.push(`stage total ${total}`);
},
upper,
sink,
(err?: Error | null) => {
log.push(`done ${err ? err.message : "ok"} ${sink.writableFinished}`);
console.log(log.join("\n"));
},
);
console.log("returns the destination", ret === sink);