stream: use internal streams for TransformStream

nodejs · nodejs-github-bot · Dec 26, 2023 · Dec 15, 2023 · Dec 4, 2023 · Dec 4, 2023
commit 3846bfed16a387bc62a077e6ffefb06112a00dff
diff --git a/lib/internal/webstreams/readablestream.js b/lib/internal/webstreams/readablestream.js
@@ -3391,4 +3391,6 @@ module.exports = {
   readableByteStreamControllerPullSteps,
   setupReadableByteStreamController,
   setupReadableByteStreamControllerFromSource,
+  createReadableStream,
+  createReadableByteStream,
 };
diff --git a/lib/internal/webstreams/transformstream.js b/lib/internal/webstreams/transformstream.js
@@ -51,6 +51,7 @@ const {
 
 const {
   ReadableStream,
+  createReadableStream,
   readableStreamDefaultControllerCanCloseOrEnqueue,
   readableStreamDefaultControllerClose,
   readableStreamDefaultControllerEnqueue,
@@ -61,6 +62,7 @@ const {
 
 const {
   WritableStream,
+  createWritableStream,
   writableStreamDefaultControllerErrorIfNeeded,
 } = require('internal/webstreams/writablestream');
 
@@ -360,36 +362,24 @@ function initializeTransformStream(
   readableHighWaterMark,
   readableSizeAlgorithm) {
 
-  const writable = new WritableStream({
-    __proto__: null,
-    start() { return startPromise.promise; },
-    write(chunk) {
-      return transformStreamDefaultSinkWriteAlgorithm(stream, chunk);
-    },
-    abort(reason) {
-      return transformStreamDefaultSinkAbortAlgorithm(stream, reason);
-    },
-    close() {
-      return transformStreamDefaultSinkCloseAlgorithm(stream);
-    },
-  }, {
-    highWaterMark: writableHighWaterMark,
-    size: writableSizeAlgorithm,
-  });
+  const startAlgorithm = () => startPromise.promise;
 
-  const readable = new ReadableStream({
-    __proto__: null,
-    start() { return startPromise.promise; },
-    pull() {
-      return transformStreamDefaultSourcePullAlgorithm(stream);
-    },
-    cancel(reason) {
-      return transformStreamDefaultSourceCancelAlgorithm(stream, reason);
-    },
-  }, {
-    highWaterMark: readableHighWaterMark,
-    size: readableSizeAlgorithm,
-  });
+  const writable = createWritableStream(
+    startAlgorithm,
+    (chunk) => transformStreamDefaultSinkWriteAlgorithm(stream, chunk),
+    () => transformStreamDefaultSinkCloseAlgorithm(stream),
+    (reason) => transformStreamDefaultSinkAbortAlgorithm(stream, reason),
+    writableHighWaterMark,
+    writableSizeAlgorithm,
+  );
+
+  const readable = createReadableStream(
+      startAlgorithm,
+    () => transformStreamDefaultSourcePullAlgorithm(stream),
+    (reason) => transformStreamDefaultSourceCancelAlgorithm(stream, reason),
+    readableHighWaterMark,
+    readableSizeAlgorithm,
+  );
 
   stream[kState] = {
     readable,

diff --git a/lib/internal/webstreams/writablestream.js b/lib/internal/webstreams/writablestream.js
@@ -581,6 +581,74 @@ ObjectDefineProperties(WritableStreamDefaultController.prototype, {
   [SymbolToStringTag]: getNonWritablePropertyDescriptor(WritableStreamDefaultController.name),
 });
 
+function InternalWritableStream(start, write, close, abort, highWaterMark, size) {
+  markTransferMode(this, false, true);
+  this[kType] = 'WritableStream';
+  this[kState] = {
+    close: createDeferredPromise(),
+    closeRequest: {
+      promise: undefined,
+      resolve: undefined,
+      reject: undefined,
+    },
+    inFlightWriteRequest: {
+      promise: undefined,
+      resolve: undefined,
+      reject: undefined,
+    },
+    inFlightCloseRequest: {
+      promise: undefined,
+      resolve: undefined,
+      reject: undefined,
+    },
+    pendingAbortRequest: {
+      abort: {
+        promise: undefined,
+        resolve: undefined,
+        reject: undefined,
+      },
+      reason: undefined,
+      wasAlreadyErroring: false,
+    },
+    backpressure: false,
+    controller: undefined,
+    state: 'writable',
+    storedError: undefined,
+    writeRequests: [],
+    writer: undefined,
+    transfer: {
+      readable: undefined,
+      port1: undefined,
+      port2: undefined,
+      promise: undefined,
+    },
+  };
+  this[kIsClosedPromise] = createDeferredPromise();
+
+  const controller = new WritableStreamDefaultController(kSkipThrow);
+  setupWritableStreamDefaultController(
+    this,
+    controller,
+    start,
+    write,
+    close,
+    abort,
+    highWaterMark,
+    size
+  )
+}
+
+ObjectSetPrototypeOf(InternalWritableStream.prototype, WritableStream.prototype);
+ObjectSetPrototypeOf(InternalWritableStream, WritableStream);
+
+function createWritableStream(start, write, close, abort, highWaterMark, size) {
+  const stream = new InternalWritableStream(start, write, close, abort, highWaterMark, size);
+
+  // For spec compliance the InternalWritableStream must be a WritableStream
+  stream.constructor = WritableStream;
+  return stream;
+}
+
 const isWritableStream =
   isBrandCheck('WritableStream');
 const isWritableStreamDefaultWriter =
@@ -1360,4 +1428,5 @@ module.exports = {
   writableStreamDefaultControllerAdvanceQueueIfNeeded,
   setupWritableStreamDefaultControllerFromSink,
   setupWritableStreamDefaultController,
+  createWritableStream,
 };