Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion lib/internal/streams/iter/push.js
Original file line number Diff line number Diff line change
Expand Up @@ -668,7 +668,7 @@ function createReadable(queue) {
},
async throw(error) {
queue.consumerThrow(error);
return { __proto__: null, done: true, value: undefined };
throw error;
},
};
},
Expand Down
43 changes: 34 additions & 9 deletions test/parallel/test-stream-iter-push-writer.js
Original file line number Diff line number Diff line change
Expand Up @@ -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() {
Expand Down Expand Up @@ -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(
Expand All @@ -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();
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -504,6 +528,7 @@ Promise.all([
testWriteUint8Array(),
testOndrainWaitsForDrain(),
testConsumerThrowRejectsWrites(),
testConsumerThrowRejectsWithThrownError(),
testEndResolvesPendingRead(),
testFailRejectsPendingRead(),
testFailRejectsFutureReadWithFalsyReason(),
Expand Down
Loading