diff --git a/packages/compiler/test/ts7/baselines/order-parity/tests/corpus/6564-constructor-function-stream-late-hooks.cjs.json b/packages/compiler/test/ts7/baselines/order-parity/tests/corpus/6564-constructor-function-stream-late-hooks.cjs.json new file mode 100644 index 000000000..52d8e55ff --- /dev/null +++ b/packages/compiler/test/ts7/baselines/order-parity/tests/corpus/6564-constructor-function-stream-late-hooks.cjs.json @@ -0,0 +1,6 @@ +{ + "order": [ + "/tests/corpus/6564-constructor-function-stream-late-hooks.cjs" + ], + "diags": [] +} diff --git a/packages/runtime/src/scr_dyn_builtin_stream.c b/packages/runtime/src/scr_dyn_builtin_stream.c index d9ac6bc50..1926b0622 100644 --- a/packages/runtime/src/scr_dyn_builtin_stream.c +++ b/packages/runtime/src/scr_dyn_builtin_stream.c @@ -204,10 +204,10 @@ ScrDyn *scr_dyn_builtin_stream_object(ScrStream *s) { } /* A stream as a listener argument: its constructor-function object, or a - * live view of another runtime stream. +1. */ + * live view of another runtime stream whose pipe and unpipe take those + * objects (scr_dyn_builtin_stream_view). +1. */ static ScrDyn *bcs_stream_value(ScrStream *s) { - ScrDyn *object = scr_dyn_builtin_stream_object(s); - return object ? object : scr_stream_dyn_view(s); + return scr_dyn_builtin_stream_view(s); } /* The receiver's stream for a prototype method, borrowed; throws Node's @@ -398,12 +398,35 @@ static void bcs_plain_call(ScrClosure *cb, ScrStream *s, const char *method, int scr_dyn_release(object); } +static bool bcs_is_function(ScrDyn *object, const char *name, bool *failed); + +/* prefinish reads `stream._final` when the writable side finishes, so a + * method assigned after the constructor runs; without one the finish goes + * on as Node's does when `typeof stream._final !== 'function'`. */ static void bcs_final_inv(ScrClosure *cb, ScrStream *s) { - bcs_plain_call(cb, s, "_final", BCS_DONE_FINAL); + ScrDyn *object = scr_box_get_ref(cb->caps[0]); + bool failed = false; + bool present = bcs_is_function(object, "_final", &failed); + scr_dyn_release(object); + if (failed) return; + if (present) bcs_plain_call(cb, s, "_final", BCS_DONE_FINAL); + else scr_stream_final_done(s, NULL); } +/* Transform's final reads `this._flush` the same way: without one it ends + * the readable side. */ static void bcs_flush_inv(ScrClosure *cb, ScrStream *s) { - bcs_plain_call(cb, s, "_flush", BCS_DONE_FLUSH); + ScrDyn *object = scr_box_get_ref(cb->caps[0]); + bool failed = false; + bool present = bcs_is_function(object, "_flush", &failed); + if (!failed && bcs_is_function(object, "_final", &failed)) { + bcs_unsupported("_final on a constructor-function Transform is not supported yet"); + failed = true; + } + scr_dyn_release(object); + if (failed) return; + if (present) bcs_plain_call(cb, s, "_flush", BCS_DONE_FLUSH); + else scr_stream_flush_done(s, NULL, NULL, NULL); } static ScrDyn *bcs_default_method(int kind, const char *name); @@ -897,23 +920,14 @@ static ScrDyn *bcs_construct(int kind, ScrDyn *self, ScrDyn *const *args, size_t scr_stream_set_read(s, bcs_hook(self, (void *)bcs_read_inv), bcs_read_inv); if (kind == BCS_WRITABLE || kind == BCS_DUPLEX) { scr_stream_set_write(s, bcs_hook(self, (void *)bcs_write_inv), bcs_write_inv); - if (bcs_is_function(self, "_final", &failed)) scr_stream_set_final(s, bcs_hook(self, (void *)bcs_final_inv), bcs_final_inv); - else scr_stream_set_final(s, NULL, NULL); + /* _final is read when the writable side finishes (bcs_final_inv). */ + scr_stream_set_final(s, bcs_hook(self, (void *)bcs_final_inv), bcs_final_inv); } if (kind == BCS_TRANSFORM || kind == BCS_PASSTHROUGH) { - bool identity = false; - if (kind == BCS_PASSTHROUGH) { - ScrDyn *own = bcs_get(self, "_transform"); - ScrDyn *fallback = bcs_default_method(BCS_PASSTHROUGH, "_transform"); - identity = own && fallback && scr_dyn_strict_eq(own, fallback); - if (own) scr_dyn_release(own); - else failed = true; - if (fallback) scr_dyn_release(fallback); - } - if (identity) scr_stream_set_transform(s, NULL, NULL); - else scr_stream_set_transform(s, bcs_hook(self, (void *)bcs_transform_inv), bcs_transform_inv); - if (bcs_is_function(self, "_flush", &failed)) scr_stream_set_flush(s, bcs_hook(self, (void *)bcs_flush_inv), bcs_flush_inv); - else scr_stream_set_flush(s, NULL, NULL); + /* _transform is read per chunk (a PassThrough's default passes the + * chunk on) and _flush when the writable side finishes. */ + scr_stream_set_transform(s, bcs_hook(self, (void *)bcs_transform_inv), bcs_transform_inv); + scr_stream_set_flush(s, bcs_hook(self, (void *)bcs_flush_inv), bcs_flush_inv); if (!failed && bcs_is_function(self, "_final", &failed)) { bcs_unsupported("_final on a constructor-function Transform is not supported yet"); failed = true; diff --git a/tests/corpus/6564-constructor-function-stream-late-hooks.cjs b/tests/corpus/6564-constructor-function-stream-late-hooks.cjs new file mode 100644 index 000000000..22a8ec2f4 --- /dev/null +++ b/tests/corpus/6564-constructor-function-stream-late-hooks.cjs @@ -0,0 +1,117 @@ +// Stream hooks a constructor function assigns after the base constructor +// runs, and unpipe from the destination's 'pipe' listener. +const { Writable, Transform, PassThrough } = require("stream"); +const util = require("util"); + +function Sink() { + Writable.call(this); + this.chunks = []; + this._final = function (cb) { + console.log("_final", this instanceof Sink, this.chunks.join(",")); + setImmediate(cb); + }; +} +util.inherits(Sink, Writable); +Sink.prototype._write = function (chunk, enc, cb) { + this.chunks.push(String(chunk)); + cb(); +}; + +const sink = new Sink(); +sink.on("prefinish", () => console.log("sink prefinish")); +sink.on("finish", () => { + console.log("sink finish"); + flushing(); +}); +sink.write("a"); +sink.end("b"); + +function Summer() { + Transform.call(this); + this.total = 0; + this._flush = function (cb) { + console.log("_flush", this.total); + cb(null, "total " + this.total); + }; +} +util.inherits(Summer, Transform); +Summer.prototype._transform = function (chunk, enc, cb) { + this.total += Number(String(chunk)); + cb(null, "seen " + chunk + "\n"); +}; + +function flushing() { + const summer = new Summer(); + const out = []; + summer.on("data", (chunk) => out.push(String(chunk))); + summer.on("end", () => { + console.log("summer end", JSON.stringify(out)); + passing(); + }); + summer.write("2"); + summer.write("3"); + summer.end(); +} + +function Upper() { + PassThrough.call(this); + this._transform = function (chunk, enc, cb) { + cb(null, String(chunk).toUpperCase()); + }; +} +util.inherits(Upper, PassThrough); + +function passing() { + const upper = new Upper(); + const out = []; + upper.on("data", (chunk) => out.push(String(chunk))); + upper.on("end", () => { + console.log("upper end", JSON.stringify(out)); + plain(); + }); + upper.write("x"); + upper.end("y"); +} + +// Without late hooks the defaults still apply. +function Plain() { + PassThrough.call(this); +} +util.inherits(Plain, PassThrough); + +function plain() { + const p = new Plain(); + const out = []; + p.on("data", (chunk) => out.push(String(chunk))); + p.on("finish", () => console.log("plain finish")); + p.on("end", () => { + console.log("plain end", JSON.stringify(out)); + unpiping(); + }); + p.end("same"); +} + +function Collector() { + Writable.call(this); + this.chunks = []; +} +util.inherits(Collector, Writable); +Collector.prototype._write = function (chunk, enc, cb) { + this.chunks.push(String(chunk)); + cb(); +}; + +function unpiping() { + const source = new PassThrough(); + const collector = new Collector(); + collector.on("pipe", function (src) { + console.log("pipe", typeof src.unpipe, this === collector); + src.unpipe(this); + }); + collector.on("unpipe", (src) => { + console.log("unpipe", typeof src.pipe); + source.write("after"); + setImmediate(() => console.log("collected", JSON.stringify(collector.chunks), source.readableFlowing)); + }); + source.pipe(collector); +} diff --git a/tests/harness/corpus-costs/6564-constructor-function-stream-late-hooks.cjs.json b/tests/harness/corpus-costs/6564-constructor-function-stream-late-hooks.cjs.json new file mode 100644 index 000000000..e873baf78 --- /dev/null +++ b/tests/harness/corpus-costs/6564-constructor-function-stream-late-hooks.cjs.json @@ -0,0 +1 @@ +{ "plain": 2750, "san": 3200 }