Await the values of untyped sync pipeline sources

An untyped pipeline source that is a sync iterable passed each value on
as it came, so a promise reached the next stage as the promise: a
byte-mode destination refused it and an object-mode one received it.
for await over a sync iterator, which Node's pipeline pumps it with,
awaits each value. The source's generator now awaits a value that is
an object (a promise included) before yielding it, and a rejection
closes the source, as Node's async-from-sync iterator does.
This commit is contained in:
Chris Tate
2026-10-24 16:10:00 -05:00
parent 55740d094a
commit 530bd9534f
2 changed files with 69 additions and 2 deletions
@@ -3947,11 +3947,13 @@ function lowerDynIterableSource(
const syncId = frame.local("sync", BOOL, false);
const nextId = frame.local("next", DYN, false);
const stepId = frame.local("step", DYN, true);
const valueId = frame.local("value", DYN, true);
const activeId = frame.local("active", BOOL, true);
const closeId = frame.local("close", DYN, false);
const it = (): IrExpr => varRef(itId, DYN, loc);
const sync = (): IrExpr => varRef(syncId, BOOL, loc);
const step = (): IrExpr => varRef(stepId, DYN, loc);
const value = (): IrExpr => varRef(valueId, DYN, loc);
const get = (obj: IrExpr, name: string): IrExpr => ({
kind: "dynKeyGet",
value: obj,
@@ -4011,6 +4013,7 @@ function lowerDynIterableSource(
},
{ kind: "varDecl", localId: nextId, init: get(it(), "next"), loc },
{ kind: "varDecl", localId: stepId, init: dynUndefinedExpr(loc), loc },
{ kind: "varDecl", localId: valueId, init: dynUndefinedExpr(loc), loc },
{ kind: "varDecl", localId: activeId, init: boolLit(false, loc), loc },
{
kind: "tryCatch",
@@ -4055,9 +4058,48 @@ function lowerDynIterableSource(
loc,
},
setActive(true),
{ kind: "assign", localId: valueId, value: get(step(), "value"), loc },
// for await over a sync iterator awaits each value
// (CreateAsyncFromSyncIterator): a promise or a thenable
// passes on its result, and a rejection closes the iterator
{
kind: "if",
cond: {
kind: "logical",
op: "&&",
left: sync(),
right: {
kind: "logical",
op: "||",
left: { kind: "dynTest", test: "promise", value: value(), type: BOOL, loc },
right: {
kind: "logical",
op: "&&",
left: {
kind: "dynTest",
test: "nullish",
negated: true,
value: value(),
type: BOOL,
loc,
},
right: { kind: "dynTest", test: "object", value: value(), type: BOOL, loc },
type: BOOL,
loc,
},
type: BOOL,
loc,
},
type: BOOL,
loc,
},
then: [{ kind: "assign", localId: valueId, value: awaitDyn(value()), loc }],
else_: null,
loc,
},
{
kind: "exprStmt",
expr: { kind: "yieldExpr", value: get(step(), "value"), type: VOID, loc },
expr: { kind: "yieldExpr", value: value(), type: VOID, loc },
loc,
},
],
@@ -1,6 +1,7 @@
// pipeline() over untyped iterable sources: a destination that fails closes
// the source, backpressure waits for drain, an abort signal stops the pump,
// and async generator and Map sources reach object-mode destinations.
// async generator and Map sources reach object-mode destinations, and a
// sync source's promise values are awaited.
import { pipeline, Writable, PassThrough } from "node:stream";
import { pipeline as pipelineP } from "node:stream/promises";
@@ -88,3 +89,27 @@ await new Promise((resolve) => {
(err) => { console.log("map done", err); resolve(undefined); },
);
});
// sync sources whose values are promises: each value is awaited, as
// for await over a sync iterator does, and a rejection closes the source
await pipelineP(
pick([Promise.resolve("p1"), "p2", new Promise((r) => setTimeout(() => r("p3"), 1))]),
new Writable({ write(c, _e, cb) { console.log("byte write", String(c)); cb(); } }),
);
function* promised() {
try {
yield Promise.resolve({ n: 1 });
yield Promise.reject(new Error("value rejected"));
yield "never";
} finally {
console.log("promised finally");
}
}
try {
await pipelineP(
pick(promised()),
new Writable({ objectMode: true, write(c, _e, cb) { console.log("object write", JSON.stringify(c)); cb(); } }),
);
} catch (err) {
console.log("promised rejected", err.message);
}