stream: improve readable webstream pipeTo

PR-URL: #49690 Reviewed-By: Matteo Collina <matteo.collina@gmail.com> Reviewed-By: Yagiz Nizipli <yagiz@nizipli.com> Reviewed-By: Moshe Atlow <moshe@atlow.co.il>
nodejs · Sep 19, 2023 · 346abdd · 346abdd
1 parent 1149d6b
commit 346abdd
Showing 1 changed file with 34 additions and 17 deletions.
diff --git a/lib/internal/webstreams/readablestream.js b/lib/internal/webstreams/readablestream.js
@@ -14,7 +14,6 @@ const {
   ObjectCreate,
   ObjectDefineProperties,
   ObjectSetPrototypeOf,
-  Promise,
   PromisePrototypeThen,
   PromiseResolve,
   PromiseReject,
@@ -1351,7 +1350,9 @@ function readableStreamPipeTo(
 
   const promise = createDeferredPromise();
 
-  let currentWrite = PromiseResolve();
+  const state = {
+    currentWrite: PromiseResolve(),
+  };
 
   // The error here can be undefined. The rejected arg
   // tells us that the promise must be rejected even
@@ -1368,9 +1369,9 @@ function readableStreamPipeTo(
   }
 
   async function waitForCurrentWrite() {
-    const write = currentWrite;
+    const write = state.currentWrite;
     await write;
-    if (write !== currentWrite)
+    if (write !== state.currentWrite)
       await waitForCurrentWrite();
   }
 
@@ -1461,20 +1462,14 @@ function readableStreamPipeTo(
   async function step() {
     if (shuttingDown)
       return true;
+
     await writer[kState].ready.promise;
-    return new Promise((resolve, reject) => {
-      readableStreamDefaultReaderRead(
-        reader,
-        {
-          [kChunk](chunk) {
-            currentWrite = writableStreamDefaultWriterWrite(writer, chunk);
-            setPromiseHandled(currentWrite);
-            resolve(false);
-          },
-          [kClose]: () => resolve(true),
-          [kError]: reject,
-        });
-    });
+
+    const promise = createDeferredPromise();
+    // eslint-disable-next-line no-use-before-define
+    readableStreamDefaultReaderRead(reader, new PipeToReadableStreamReadRequest(writer, state, promise));
+
+    return promise.promise;
   }
 
   async function run() {
@@ -1536,6 +1531,28 @@ function readableStreamPipeTo(
   return promise.promise;
 }
 
+class PipeToReadableStreamReadRequest {
+  constructor(writer, state, promise) {
+    this.writer = writer;
+    this.state = state;
+    this.promise = promise;
+  }
+
+  [kChunk](chunk) {
+    this.state.currentWrite = writableStreamDefaultWriterWrite(this.writer, chunk);
+    setPromiseHandled(this.state.currentWrite);
+    this.promise.resolve(false);
+  }
+
+  [kClose]() {
+    this.promise.resolve(true);
+  }
+
+  [kError](error) {
+    this.promise.reject(error);
+  }
+}
+
 function readableStreamTee(stream, cloneForBranch2) {
   if (isReadableByteStreamController(stream[kState].controller)) {
     return readableByteStreamTee(stream);