fix some more

src/js/internal/streams/duplexify.js

@@ -21,6 +21,7 @@ const Readable = require("internal/streams/readable");
 const Writable = require("internal/streams/writable");
 const { createDeferredPromise } = require("../../node/util");
 const from = require("internal/streams/from");
+const { AbortError } = require("../../node/events");
 
 const isBlob =
   typeof Blob !== "undefined"
@@ -55,7 +56,6 @@ class Duplexify extends Duplex {
 }
 
 export default function duplexify(body, name) {
-  console.log(1);
   if (isDuplexNodeStream(body)) {
     return body;
   }
@@ -80,8 +80,6 @@ export default function duplexify(body, name) {
     return _duplexify({ writable: Writable.fromWeb(body) });
   }
 
-  console.log(2);
-
   if (typeof body === "function") {
     const { value, write, final, destroy } = fromAsyncGen(body);
 
@@ -139,15 +137,10 @@ export default function duplexify(body, name) {
     throw new ERR_INVALID_RETURN_VALUE("Iterable, AsyncIterable or AsyncFunction", name, value);
   }
 
-  console.log(3, isBlob);
-
   if (isBlob(body)) {
-    console.log(3.01);
     return duplexify(body.arrayBuffer());
   }
 
-  console.log(3.1);
-
   if (isIterable(body)) {
     return from(Duplexify, body, {
       // TODO (ronag): highWaterMark?
@@ -156,14 +149,10 @@ export default function duplexify(body, name) {
     });
   }
 
-  console.log(3.2);
-
   if (isReadableStream(body?.readable) && isWritableStream(body?.writable)) {
     return Duplexify.fromWeb(body);
   }
 
-  console.log(3.3);
-
   if (typeof body?.writable === "object" || typeof body?.readable === "object") {
     const readable = body?.readable
       ? isReadableNodeStream(body?.readable)
@@ -180,8 +169,6 @@ export default function duplexify(body, name) {
     return _duplexify({ readable, writable });
   }
 
-  console.log(4);
-
   const then = body?.then;
   if (typeof then === "function") {
     let d;
@@ -207,8 +194,6 @@ export default function duplexify(body, name) {
     }));
   }
 
-  console.log(5);
-
   throw new ERR_INVALID_ARG_TYPE(
     name,
     [

src/js/internal/streams/end-of-stream.js

@@ -127,10 +127,10 @@ function eos(stream, options, callback) {
     }
 
     if (readable && !readableFinished && isReadableNodeStream(stream, true)) {
-      if (!isReadableFinished(stream, false)) return callback.$call(stream, new ERR_STREAM_PREMATURE_CLOSE());
+      if (!isReadableFinished(stream, false)) return callback.$call(stream, new Error("ERR_STREAM_PREMATURE_CLOSE()"));
     }
     if (writable && !writableFinished) {
-      if (!isWritableFinished(stream, false)) return callback.$call(stream, new ERR_STREAM_PREMATURE_CLOSE());
+      if (!isWritableFinished(stream, false)) return callback.$call(stream, new Error("ERR_STREAM_PREMATURE_CLOSE()"));
     }
 
     callback.$call(stream);

src/js/internal/streams/writable.js

@@ -540,8 +540,6 @@ Writable.prototype.setDefaultEncoding = function setDefaultEncoding(encoding) {
 // in the queue, and wait our turn.  Otherwise, call _write
 // If we return false, then we need a drain event, so set that flag.
 function writeOrBuffer(stream, state, chunk, encoding, callback) {
-  console.info(state);
-
   const len = (state[kState] & kObjectMode) !== 0 ? 1 : chunk.length;
 
   state.length += len;

src/js/node/util.ts

@@ -297,7 +297,7 @@ function once(callback, { preserveReturnValue = false } = kEmptyObject) {
   return function (...args) {
     if (called) return returnValue;
     called = true;
-    const result = Reflect.$apply(callback, this, args);
+    const result = callback.$apply(this, args);
     returnValue = preserveReturnValue ? result : undefined;
     return result;
   };