fix(stream): Fix Transform class constructor fn (#1583)

* fix(stream): fix ReadableFromWeb class, fix Transform

* test(stream): add tests for .call on stream constructors

2 files changed, 76 insertions, 4 deletions

diff --git a/src/bun.js/streams.exports.js b/src/bun.js/streams.exports.js
index 1582cb756..cc6325923 100644
--- a/src/bun.js/streams.exports.js
+++ b/src/bun.js/streams.exports.js
@@ -2659,8 +2659,8 @@ var require_readable = __commonJS({
       if (ev === "data") {
         state.readableListening = this.listenerCount("readable") > 0;
         if (state.flowing !== false) {
+          debug("in flowing mode!", this.__id);
           this.resume();
-          debug("in flowing mode!");
         } else {
           debug("in readable mode!", this.__id);
         }
@@ -2739,6 +2739,7 @@ var require_readable = __commonJS({
       }
 
       async _read() {
+        debug("ReadableFromWeb _read()", this.__id);
         var stream = this.#stream,
           reader = this.#reader;
         if (stream) {
@@ -2782,7 +2783,7 @@ var require_readable = __commonJS({
                 return;
               }
             }
-          } while (this._readableState.flowing && !this.#closed);
+          } while (!this.#closed);
         } catch (e) {
           deferredError = e;
         } finally {
@@ -2838,12 +2839,32 @@ var require_readable = __commonJS({
       }
 
       validateObject(options, "options");
-      const { highWaterMark, encoding, objectMode = false, signal } = options;
+      const {
+        highWaterMark,
+        encoding,
+        objectMode = false,
+        signal,
+        // native = true,
+      } = options;
 
       if (encoding !== undefined && !Buffer.isEncoding(encoding))
         throw new ERR_INVALID_ARG_VALUE(encoding, "options.encoding");
       validateBoolean(objectMode, "options.objectMode");
 
+      // validateBoolean(native, "options.native");
+
+      // if (!native) {
+      //   return new ReadableFromWeb(
+      //     {
+      //       highWaterMark,
+      //       encoding,
+      //       objectMode,
+      //       signal,
+      //     },
+      //     readableStream,
+      //   );
+      // }
+
       const nativeStream = getNativeReadableStream(
         Readable,
         readableStream,
@@ -3141,6 +3162,7 @@ var require_readable = __commonJS({
         state.errored ||
         !state.constructed
       ) {
+        debug("state.constructed?", state.constructed, this.__id);
         doRead = false;
         debug("reading, ended or constructing", doRead, this.__id);
       } else if (doRead) {
@@ -4966,15 +4988,18 @@ var require_transform = __commonJS({
     var Duplex = require_duplex();
     function Transform(options) {
       if (!(this instanceof Transform)) return new Transform(options);
+      Duplex.call(this, options);
+
       this._readableState.sync = false;
       this[kCallback] = null;
+
       if (options) {
         if (typeof options.transform === "function")
           this._transform = options.transform;
         if (typeof options.flush === "function") this._flush = options.flush;
       }
+
       this.on("prefinish", prefinish);
-      Duplex.call(this, options);
     }
     ObjectSetPrototypeOf(Transform.prototype, Duplex.prototype);
     ObjectSetPrototypeOf(Transform, Duplex);
diff --git a/test/bun.js/node-stream.test.js b/test/bun.js/node-stream.test.js
new file mode 100644
index 000000000..8024ab562
--- /dev/null
+++ b/test/bun.js/node-stream.test.js
@@ -0,0 +1,47 @@
+import { expect, describe, it } from "bun:test";
+import { Duplex, Transform, PassThrough } from "node:stream";
+
+describe("Duplex", () => {
+  it("should allow subclasses to be derived via .call() on class", () => {
+    function Subclass(opts) {
+      if (!(this instanceof Subclass)) return new Subclass(opts);
+      Duplex.call(this, opts);
+    }
+
+    Object.setPrototypeOf(Subclass.prototype, Duplex.prototype);
+    Object.setPrototypeOf(Subclass, Duplex);
+
+    const subclass = new Subclass();
+    expect(subclass instanceof Duplex).toBe(true);
+  });
+});
+
+describe("Transform", () => {
+  it("should allow subclasses to be derived via .call() on class", () => {
+    function Subclass(opts) {
+      if (!(this instanceof Subclass)) return new Subclass(opts);
+      Transform.call(this, opts);
+    }
+
+    Object.setPrototypeOf(Subclass.prototype, Transform.prototype);
+    Object.setPrototypeOf(Subclass, Transform);
+
+    const subclass = new Subclass();
+    expect(subclass instanceof Transform).toBe(true);
+  });
+});
+
+describe("PassThrough", () => {
+  it("should allow subclasses to be derived via .call() on class", () => {
+    function Subclass(opts) {
+      if (!(this instanceof Subclass)) return new Subclass(opts);
+      PassThrough.call(this, opts);
+    }
+
+    Object.setPrototypeOf(Subclass.prototype, PassThrough.prototype);
+    Object.setPrototypeOf(Subclass, PassThrough);
+
+    const subclass = new Subclass();
+    expect(subclass instanceof PassThrough).toBe(true);
+  });
+});