mirror of
https://github.com/nodejs/node.git
synced 2026-11-09 15:11:11 -05:00
Matteo Collina
8aee69b912
A transform sink write arriving under backpressure parked the chunk together with a PromiseWithResolvers record whose promise was returned to the writable controller and later resolved with the perform-transform promise. The writable's write reactions already exist before the write algorithm runs, so the parked write can instead return the parked-result sentinel and have the continuation wire the perform-transform promise directly to those reactions, dropping the per-chunk promise record and the thenable adoption hop. Failures while erroring are delivered in a microtask, preserving the old rejection position. Signed-off-by: Matteo Collina <hello@matteocollina.com> PR-URL: https://github.com/nodejs/node/pull/65625 Reviewed-By: Vinícius Lourenço Claro Cardoso <contact@viniciusl.com.br> Reviewed-By: James M Snell <jasnell@gmail.com> Reviewed-By: Antoine du Hamel <duhamelantoine1995@gmail.com> Reviewed-By: Gürgün Dayıoğlu <hey@gurgun.day>
58 lines
1.8 KiB
JavaScript
58 lines
1.8 KiB
JavaScript
'use strict';
|
|
|
|
require('../common');
|
|
const { test } = require('node:test');
|
|
const assert = require('node:assert');
|
|
const { TransformStream } = require('stream/web');
|
|
const { setImmediate } = require('timers/promises');
|
|
|
|
// A transform sink write arriving while the stream has backpressure is
|
|
// parked until backpressure clears. These cases complete a parked write
|
|
// while the writable side is already erroring.
|
|
//
|
|
// The setImmediate() lets the start algorithm settle so the write below
|
|
// reaches the sink and parks (backpressure is set until the readable
|
|
// side pulls).
|
|
|
|
test('readable.cancel() rejects a parked write with the cancel reason', async () => {
|
|
const stream = new TransformStream();
|
|
const writer = stream.writable.getWriter();
|
|
await setImmediate();
|
|
|
|
const reason = new Error('cancelled');
|
|
const write = writer.write('parked');
|
|
await stream.readable.cancel(reason);
|
|
await assert.rejects(write, (err) => err === reason);
|
|
});
|
|
|
|
test('controller.error() rejects a parked write with the stored error', async () => {
|
|
let controller;
|
|
const stream = new TransformStream({
|
|
start(c) { controller = c; },
|
|
});
|
|
const writer = stream.writable.getWriter();
|
|
await setImmediate();
|
|
|
|
const reason = new Error('boom');
|
|
const write = writer.write('parked');
|
|
controller.error(reason);
|
|
await assert.rejects(write, (err) => err === reason);
|
|
await assert.rejects(writer.closed, (err) => err === reason);
|
|
});
|
|
|
|
test('controller.terminate() rejects a parked write', async () => {
|
|
let controller;
|
|
const stream = new TransformStream({
|
|
start(c) { controller = c; },
|
|
});
|
|
const writer = stream.writable.getWriter();
|
|
await setImmediate();
|
|
|
|
const write = writer.write('parked');
|
|
controller.terminate();
|
|
await assert.rejects(write, {
|
|
name: 'TypeError',
|
|
message: /terminated/,
|
|
});
|
|
});
|