stream: make .destroy()
interact better with write queue
Make sure that it is safe to call the callback for `_write()` even in the presence of `.destroy()` calls during that write. In particular, letting the write queue continue processing would previously have thrown an exception, because processing writes after calling `.destroy()` is forbidden. One test had to be modified to account for the fact that callbacks for writes will now always be called, even when the stream is destroyed during the process. PR-URL: https://github.com/nodejs/node/pull/24062 Reviewed-By: Matteo Collina <matteo.collina@gmail.com> Reviewed-By: James M Snell <jasnell@gmail.com>
This commit is contained in:
parent
fb6c6692a8
commit
d3f02d0da3
@ -461,7 +461,7 @@ function onwrite(stream, er) {
|
|||||||
onwriteError(stream, state, sync, er, cb);
|
onwriteError(stream, state, sync, er, cb);
|
||||||
else {
|
else {
|
||||||
// Check if we're actually ready to finish, but don't emit yet
|
// Check if we're actually ready to finish, but don't emit yet
|
||||||
var finished = needFinish(state);
|
var finished = needFinish(state) || stream.destroyed;
|
||||||
|
|
||||||
if (!finished &&
|
if (!finished &&
|
||||||
!state.corked &&
|
!state.corked &&
|
||||||
|
59
test/parallel/test-stream-write-destroy.js
Normal file
59
test/parallel/test-stream-write-destroy.js
Normal file
@ -0,0 +1,59 @@
|
|||||||
|
'use strict';
|
||||||
|
require('../common');
|
||||||
|
const assert = require('assert');
|
||||||
|
const { Writable } = require('stream');
|
||||||
|
|
||||||
|
// Test interaction between calling .destroy() on a writable and pending
|
||||||
|
// writes.
|
||||||
|
|
||||||
|
for (const withPendingData of [ false, true ]) {
|
||||||
|
for (const useEnd of [ false, true ]) {
|
||||||
|
const callbacks = [];
|
||||||
|
|
||||||
|
const w = new Writable({
|
||||||
|
write(data, enc, cb) {
|
||||||
|
callbacks.push(cb);
|
||||||
|
},
|
||||||
|
// Effectively disable the HWM to observe 'drain' events more easily.
|
||||||
|
highWaterMark: 1
|
||||||
|
});
|
||||||
|
|
||||||
|
let chunksWritten = 0;
|
||||||
|
let drains = 0;
|
||||||
|
let finished = false;
|
||||||
|
w.on('drain', () => drains++);
|
||||||
|
w.on('finish', () => finished = true);
|
||||||
|
|
||||||
|
w.write('abc', () => chunksWritten++);
|
||||||
|
assert.strictEqual(chunksWritten, 0);
|
||||||
|
assert.strictEqual(drains, 0);
|
||||||
|
callbacks.shift()();
|
||||||
|
assert.strictEqual(chunksWritten, 1);
|
||||||
|
assert.strictEqual(drains, 1);
|
||||||
|
|
||||||
|
if (withPendingData) {
|
||||||
|
// Test 2 cases: There either is or is not data still in the write queue.
|
||||||
|
// (The second write will never actually get executed either way.)
|
||||||
|
w.write('def', () => chunksWritten++);
|
||||||
|
}
|
||||||
|
if (useEnd) {
|
||||||
|
// Again, test 2 cases: Either we indicate that we want to end the
|
||||||
|
// writable or not.
|
||||||
|
w.end('ghi', () => chunksWritten++);
|
||||||
|
} else {
|
||||||
|
w.write('ghi', () => chunksWritten++);
|
||||||
|
}
|
||||||
|
|
||||||
|
assert.strictEqual(chunksWritten, 1);
|
||||||
|
w.destroy();
|
||||||
|
assert.strictEqual(chunksWritten, 1);
|
||||||
|
callbacks.shift()();
|
||||||
|
assert.strictEqual(chunksWritten, 2);
|
||||||
|
assert.strictEqual(callbacks.length, 0);
|
||||||
|
assert.strictEqual(drains, 1);
|
||||||
|
|
||||||
|
// When we used `.end()`, we see the 'finished' event if and only if
|
||||||
|
// we actually finished processing the write queue.
|
||||||
|
assert.strictEqual(finished, !withPendingData && useEnd);
|
||||||
|
}
|
||||||
|
}
|
Loading…
x
Reference in New Issue
Block a user