diff --git a/lib/internal/streams/iter/push.js b/lib/internal/streams/iter/push.js index 23adc5edc1cd3e..e5ea217461840f 100644 --- a/lib/internal/streams/iter/push.js +++ b/lib/internal/streams/iter/push.js @@ -668,7 +668,7 @@ function createReadable(queue) { }, async throw(error) { queue.consumerThrow(error); - return { __proto__: null, done: true, value: undefined }; + throw error; }, }; }, diff --git a/test/parallel/test-stream-iter-push-writer.js b/test/parallel/test-stream-iter-push-writer.js index e617a8a5309dae..cf61ede0ab86cb 100644 --- a/test/parallel/test-stream-iter-push-writer.js +++ b/test/parallel/test-stream-iter-push-writer.js @@ -147,10 +147,16 @@ async function testOndrainRejectsOnConsumerThrow() { // Consumer throws via iterator.throw() before draining enough // to clear backpressure. The drain should reject. const iter = readable[Symbol.asyncIterator](); - await iter.throw(new Error('consumer error')); + const err = new Error('consumer error'); + const drainRejects = assert.rejects(drainPromise, (e) => e === err); + const pendingWriteRejects = pendingWrite.catch(() => {}); + await assert.rejects( + () => iter.throw(err), + (e) => e === err, + ); - await assert.rejects(drainPromise, /consumer error/); - await pendingWrite.catch(() => {}); // Ignore write rejection + await drainRejects; + await pendingWriteRejects; // Ignore write rejection } async function testWritev() { @@ -281,7 +287,11 @@ async function testConsumerThrowRejectsWrites() { writer.writeSync('a'); const iter = readable[Symbol.asyncIterator](); - await iter.throw(new Error('consumer boom')); + const err = new Error('consumer boom'); + await assert.rejects( + () => iter.throw(err), + (e) => e === err, + ); // Subsequent async writes should reject with the consumer's error await assert.rejects( @@ -290,6 +300,18 @@ async function testConsumerThrowRejectsWrites() { ); } +async function testConsumerThrowRejectsWithThrownError() { + const { readable } = push(); + + const iter = readable[Symbol.asyncIterator](); + const err = new Error('boom'); + + await assert.rejects( + () => iter.throw(err), + (e) => e === err, + ); +} + // end() resolves a pending read as done:true async function testEndResolvesPendingRead() { const { writer, readable } = push(); @@ -351,14 +373,16 @@ async function testConsumerThrowRejectsPendingRead() { await new Promise(setImmediate); const err = new Error('consumer read boom'); - const throwResult = await iter.throw(err); - assert.strictEqual(throwResult.value, undefined); - assert.strictEqual(throwResult.done, true); - - await assert.rejects( + const readRejects = assert.rejects( () => readPromise, (e) => e === err, ); + await assert.rejects( + () => iter.throw(err), + (e) => e === err, + ); + + await readRejects; } // end() while writes are pending rejects those writes @@ -504,6 +528,7 @@ Promise.all([ testWriteUint8Array(), testOndrainWaitsForDrain(), testConsumerThrowRejectsWrites(), + testConsumerThrowRejectsWithThrownError(), testEndResolvesPendingRead(), testFailRejectsPendingRead(), testFailRejectsFutureReadWithFalsyReason(),