diff --git a/packages/compiler/src/frontend/lowering/lower-stream.ts b/packages/compiler/src/frontend/lowering/lower-stream.ts index 4d729f639..44027f65c 100644 --- a/packages/compiler/src/frontend/lowering/lower-stream.ts +++ b/packages/compiler/src/frontend/lowering/lower-stream.ts @@ -3876,6 +3876,44 @@ export function lowerStreamMethodCall( loc, }; } + // Checked-dynamic slots: unpipe() also unpipes the JavaScript + // destinations stream.pipeDyn registered (stream.unpipeDyn with an + // undefined destination), as Node's unpipe() empties every pipe. + if (lowerer.dynNodeStreams) { + const source = lowerer.declareHiddenLocal("%unpipeSource", receiver.type); + const sourceRef = varRef(source.id, receiver.type, loc); + return { + kind: "seqExpr", + stmts: [ + { kind: "varDecl", localId: source.id, init: receiver, loc }, + { + kind: "exprStmt", + expr: { + kind: "libCall", + fn: "readable.unpipe", + args: [sourceRef], + type: receiver.type, + loc, + }, + loc, + }, + { + kind: "exprStmt", + expr: { + kind: "libCall", + fn: "stream.unpipeDyn", + args: [sourceRef, dynUndefinedExpr(loc)], + type: VOID, + loc, + }, + loc, + }, + ], + result: sourceRef, + type: receiver.type, + loc, + }; + } return { kind: "libCall", fn: "readable.unpipe", args: [receiver], type: receiver.type, loc }; } diff --git a/packages/runtime/src/scr_stream_dyn.c b/packages/runtime/src/scr_stream_dyn.c index d230bf22e..0859c4e07 100644 --- a/packages/runtime/src/scr_stream_dyn.c +++ b/packages/runtime/src/scr_stream_dyn.c @@ -365,13 +365,25 @@ out: /* readable.unpipe(destination) for a destination that is not a native * stream: the pipe scr_stream_pipe_dyn registered from src into it * unpipes, as Node's unpipe does for a piped destination; nothing happens - * when there is none. Borrowed. */ + * when there is none. An undefined destination (unpipe() without one) + * unpipes every JavaScript destination of src, oldest pipe first. + * Borrowed. */ void scr_stream_unpipe_dyn(ScrStream *src, ScrDyn *dest) { - for (ScrDynPipe *p = scr_dyn_pipe_live; p; p = p->live_next) { - if (p->src != src || p->unpiped || p->cleaned || !p->dest || !scr_dyn_strict_eq(p->dest, dest)) continue; - ScrDynPipe *held = scr_dyn_pipe_retain_v(p); + bool all = dest == NULL || dest->kind == SCR_DYN_UNDEF; + for (;;) { + /* The live list is newest first; the oldest matching pipe goes next. + * Unpiping runs JavaScript ('unpipe' listeners), so each round walks + * the list again. */ + ScrDynPipe *match = NULL; + for (ScrDynPipe *p = scr_dyn_pipe_live; p; p = p->live_next) { + if (p->src != src || p->unpiped || p->cleaned || !p->dest) continue; + if (!all && !scr_dyn_strict_eq(p->dest, dest)) continue; + match = p; + } + if (!match) return; + ScrDynPipe *held = scr_dyn_pipe_retain_v(match); scr_dyn_pipe_unpipe(held); scr_dyn_pipe_release_v(held); - return; + if (!all || scr_exc_pending()) return; } } diff --git a/tests/corpus/5819-node-stream-dyn-direct-pipe/main.ts b/tests/corpus/5819-node-stream-dyn-direct-pipe/main.ts index c1feb8eb5..a72328d3f 100644 --- a/tests/corpus/5819-node-stream-dyn-direct-pipe/main.ts +++ b/tests/corpus/5819-node-stream-dyn-direct-pipe/main.ts @@ -1,7 +1,8 @@ // A native Readable piped straight into a JavaScript sink, the only // untyped stream flow in the program; unpipe of a NodeJS.WritableStream -// that holds a native Writable or a JavaScript sink; and JavaScript -// destinations whose prependListener or writableNeedDrain getter throws. +// that holds a native Writable or a JavaScript sink; unpipe() of every +// JavaScript sink; and JavaScript destinations whose prependListener or +// writableNeedDrain getter throws. import { Readable, Writable } from "node:stream"; import { plainSink, throwingSink } from "./sources.js"; @@ -37,6 +38,14 @@ async function main(): Promise { await new Promise((resolve) => setTimeout(resolve, 10)); console.log("second paused", second.isPaused()); + const third = new Readable({ read() {} }); + third.pipe(plainSink("first of two")); + third.pipe(plainSink("second of two")); + third.unpipe(); + third.push("after unpipe()"); + await new Promise((resolve) => setTimeout(resolve, 10)); + console.log("third paused", third.isPaused(), third.listenerCount("data")); + for (const member of ["prependListener", "writableNeedDrain"]) { try { new Readable({ read() {} }).pipe(throwingSink(member));