diff --git a/lib/internal/webstreams/readablestream.js b/lib/internal/webstreams/readablestream.js index dfe1124b3e93be..5603d2bfd8500f 100644 --- 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);