From 95b2f81716dc644ec5a58309cf6d08ff71c81ca9 Mon Sep 17 00:00:00 2001 From: Malte Ubl Date: Tue, 20 Oct 2026 03:24:08 -0700 Subject: [PATCH] Pipe into untyped destinations by unwrapping boxed native streams Co-Authored-By: Claude Opus 5.5 --- .../builtin-apis/streams-and-readline.mdx | 1 + .../src/frontend/lowering/lower-stream.ts | 53 +++++++++++++++++++ .../6467-node-stream-untyped-sources/main.ts | 11 +++- 3 files changed, 63 insertions(+), 2 deletions(-) diff --git a/docs/content/limitations/builtin-apis/streams-and-readline.mdx b/docs/content/limitations/builtin-apis/streams-and-readline.mdx index 860daa5d8..042fd1e50 100644 --- a/docs/content/limitations/builtin-apis/streams-and-readline.mdx +++ b/docs/content/limitations/builtin-apis/streams-and-readline.mdx @@ -1,6 +1,7 @@ - **Streams and readline:** Stream chunks are Buffers or strings. `NodeJS.ReadableStream` and `NodeJS.WritableStream`, and `stream.Readable` and `stream.Writable` under `@types/node`, accept every stream: when a program passes file, zlib, or `node:stream` streams through them, child-process pipes and `process.stdout` and `process.stderr` join those streams as `Readable` and `Writable` instances. `readable.pipe(process.stdout)` and `readable.pipe(process.stderr)` compile and, as in Node, do not end the process stream. + A `pipeline` stage that arrives as an untyped value must be a native stream at run time; other values throw a not-supported error. `Readable.from` accepts arrays, Sets, Map and Set iterators, generators, async generators, and class iterables and async iterables whose values are strings or Buffers, and in JavaScript any iterable or async iterable of them; a stream that mixes strings and Buffers is unsupported. In JavaScript, a sync iterator value with a `then` method that is not a promise, and an object whose `[Symbol.asyncIterator]` or `[Symbol.iterator]` method is a generator method, are unsupported. Object-mode streams of other values (including `Readable.from` over a `Map`, whose values are entry arrays) are unsupported. diff --git a/packages/compiler/src/frontend/lowering/lower-stream.ts b/packages/compiler/src/frontend/lowering/lower-stream.ts index 400fd79ac..51c86fe39 100644 --- a/packages/compiler/src/frontend/lowering/lower-stream.ts +++ b/packages/compiler/src/frontend/lowering/lower-stream.ts @@ -3371,6 +3371,58 @@ function pipeIntoStdio( }; } +/** `readable.pipe(untyped)`: a checked-dynamic destination unwraps a boxed + * native stream at run time (stream.unboxDyn; other values throw the + * not-supported Error), and pipe() answers the destination value itself. */ +function pipeIntoDyn( + lowerer: Lowerer, + call: ts.CallExpression, + access: ts.PropertyAccessExpression, + target: IrExpr, +): IrExpr { + const loc = locOf(call); + const receiverValue = lowerer.lowerExpr(access.expression); + const source = lowerer.declareHiddenLocal("%pipeSource", receiverValue.type); + const destination = lowerer.declareHiddenLocal("%pipeDestination", target.type); + const stmts: IrStmt[] = [ + { kind: "varDecl", localId: source.id, init: receiverValue, loc }, + { kind: "varDecl", localId: destination.id, init: target, loc }, + ]; + const end = pipeEndOption(lowerer, call.arguments.nodes[1], loc); + const duplex: IrType = { kind: "object", className: "%Duplex" }; + stmts.push({ + kind: "exprStmt", + expr: { + kind: "libCall", + fn: "readable.pipe", + args: [ + { kind: "varRef", localId: source.id, type: receiverValue.type, loc }, + { + kind: "libCall", + fn: "stream.unboxDyn", + args: [ + { kind: "varRef", localId: destination.id, type: target.type, loc }, + strLit("pipe destinations other than native streams are not supported yet", loc), + ], + type: duplex, + loc, + }, + end, + ], + type: duplex, + loc, + }, + loc, + }); + return { + kind: "seqExpr", + stmts, + result: { kind: "varRef", localId: destination.id, type: target.type, loc }, + type: target.type, + loc, + }; +} + export function lowerStreamMethodCall( lowerer: Lowerer, call: ts.CallExpression, @@ -3622,6 +3674,7 @@ export function lowerStreamMethodCall( } const target = lowerer.lowerExpr(args[0]!); if (target.type.kind === "procStream") return pipeIntoStdio(lowerer, call, access, target); + if (target.type.kind === "dyn") return pipeIntoDyn(lowerer, call, access, target); const dst = target; const dstInfo = dst.type.kind === "object" ? lowerer.classes.get(dst.type.className) : undefined; diff --git a/tests/corpus/6467-node-stream-untyped-sources/main.ts b/tests/corpus/6467-node-stream-untyped-sources/main.ts index bab7e8cc7..5cb3a628d 100644 --- a/tests/corpus/6467-node-stream-untyped-sources/main.ts +++ b/tests/corpus/6467-node-stream-untyped-sources/main.ts @@ -1,7 +1,7 @@ // Streams that arrive untyped meet NodeJS stream interface slots: an // untyped JavaScript function's stream returned as NodeJS.ReadableStream, -// a checked cast from 'unknown', pipeline() over untyped source and -// destination stages, and 'for await' over a NodeJS.ReadableStream +// a checked cast from 'unknown', pipe() and pipeline() into untyped +// destinations, and 'for await' over a NodeJS.ReadableStream // parameter fed by a file stream and a child's stdout. import { spawn } from "node:child_process"; import * as fs from "node:fs"; @@ -37,6 +37,13 @@ async function main(): Promise { const child = spawn("sh", ["-c", "printf 'child text\\n'"], { stdio: ["ignore", "pipe", "inherit"] }); if (child.stdout) await collect("child", child.stdout); + await new Promise((resolve) => { + const piped = fs.createReadStream(plain); + piped.on("end", () => resolve()); + piped.pipe(collector()); + }); + console.log("pipe", JSON.stringify(joined())); + const sink = collector(); await pipeline(openUntyped(packed), zlib.createGunzip(), sink); console.log("pipeline", JSON.stringify(joined()));