Read constructor-function stream hooks when they run

A constructor-function stream now reads `_final` and `_flush` when its
writable side finishes, and `_transform` for each chunk, as Node does. A hook
assigned after `Writable.call(this)` or `Transform.call(this)` runs, and an
override of a PassThrough's `_transform` replaces the pass-through. Without
`_final` or `_flush` the finish continues as Node's does.

A runtime stream passed to a constructor-function stream's listeners, such
as the source of a 'pipe' event, is the view whose pipe and unpipe accept
constructor-function streams, so `src.unpipe(this)` in a 'pipe' listener
unpipes instead of throwing.

packages/compiler/test/ts7/baselines/order-parity/tests/corpus/6564-constructor-function-stream-late-hooks.cjs.json

@@ -0,0 +1,6 @@
+{
+ "order": [
+  "<repo>/tests/corpus/6564-constructor-function-stream-late-hooks.cjs"
+ ],
+ "diags": []
+}

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;

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);
+}

tests/harness/corpus-costs/6564-constructor-function-stream-late-hooks.cjs.json

@@ -0,0 +1 @@
+{ "plain": 2750, "san": 3200 }