stream: improve readable webstream pipeTo

rluvaton · ruyadorno · commit b29d9270108c · 2023-09-27T23:10:21.000-04:00
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>
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,
@@ -1354,7 +1353,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
@@ -1371,9 +1372,9 @@ function readableStreamPipeTo(
 }
 
 async function waitForCurrentWrite(){
- const write = currentWrite;
+ const write = state.currentWrite;
 await write;
- if (write !== currentWrite)
+ if (write !== state.currentWrite)
 await waitForCurrentWrite();
 }
 
@@ -1464,20 +1465,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(){
@@ -1539,6 +1534,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);

-Original file line number
+Diff line change
  ObjectCreate,
  ObjectDefineProperties,
  ObjectSetPrototypeOf,
 - Promise,
  PromisePrototypeThen,
  PromiseResolve,
  PromiseReject,
 constpromise=createDeferredPromise();
 -letcurrentWrite=PromiseResolve();
 +conststate={
 +currentWrite: PromiseResolve(),
 +};
 // The error here can be undefined. The rejected arg
 // tells us that the promise must be rejected even
+}
 asyncfunctionwaitForCurrentWrite(){
 -constwrite=currentWrite;
 +constwrite=state.currentWrite;
 awaitwrite;
 -if(write!==currentWrite)
 +if(write!==state.currentWrite)
 awaitwaitForCurrentWrite();
+}
 asyncfunctionstep(){
 if(shuttingDown)
 returntrue;
++
 awaitwriter[kState].ready.promise;
 -returnnewPromise((resolve,reject)=>{
 -readableStreamDefaultReaderRead(
 -reader,
 -{
 -[kChunk](chunk){
 -currentWrite=writableStreamDefaultWriterWrite(writer,chunk);
 -setPromiseHandled(currentWrite);
 -resolve(false);
 -},
 -[kClose]: ()=>resolve(true),
 -[kError]: reject,
 -});
 -});
++
 +constpromise=createDeferredPromise();
 +// eslint-disable-next-line no-use-before-define
 +readableStreamDefaultReaderRead(reader,newPipeToReadableStreamReadRequest(writer,state,promise));
++
 +returnpromise.promise;
+}
 asyncfunctionrun(){
 returnpromise.promise;
+}
 +classPipeToReadableStreamReadRequest{
 +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);
 +}
 +}
++
 functionreadableStreamTee(stream,cloneForBranch2){
 if(isReadableByteStreamController(stream[kState].controller)){
 returnreadableByteStreamTee(stream);