mirror of
https://github.com/nodejs/node.git
synced 2026-11-09 15:11:11 -05:00
stream: destroy wrapped streams on error
Stream should be destroyed and update state accordingly when the wrapped stream emits error. Does some additional cleanup with future TODOs that might be worth looking into. PR-URL: https://github.com/nodejs/node/pull/34102 Reviewed-By: Matteo Collina <matteo.collina@gmail.com> Reviewed-By: Luigi Pinca <luigipinca@gmail.com> Reviewed-By: Anna Henningsen <anna@addaleax.net>
This commit is contained in:
1 parent
204f20f2d1
commit
8f4b4f272e
2 files changed
+55
-5
No files matched your search
+23
-5
@@ -66,7 +66,6 @@ ObjectSetPrototypeOf(Readable.prototype, Stream.prototype);
|
||||
ObjectSetPrototypeOf(Readable, Stream);
|
||||
|
||||
const { errorOrDestroy } = destroyImpl;
|
||||
const kProxyEvents = ['error', 'close', 'destroy', 'pause', 'resume'];
|
||||
|
||||
function prependListener(emitter, event, fn) {
|
||||
// Sadly this is not cacheable as some libraries bundle their own
|
||||
@@ -1051,10 +1050,29 @@ Readable.prototype.wrap = function(stream) {
|
||||
}
|
||||
}
|
||||
|
||||
// Proxy certain important events.
|
||||
for (const kProxyEvent of kProxyEvents) {
|
||||
stream.on(kProxyEvent, this.emit.bind(this, kProxyEvent));
|
||||
}
|
||||
stream.on('error', (err) => {
|
||||
errorOrDestroy(this, err);
|
||||
});
|
||||
|
||||
stream.on('close', () => {
|
||||
// TODO(ronag): Update readable state?
|
||||
this.emit('close');
|
||||
});
|
||||
|
||||
stream.on('destroy', () => {
|
||||
// TODO(ronag): this.destroy()?
|
||||
this.emit('destroy');
|
||||
});
|
||||
|
||||
stream.on('pause', () => {
|
||||
// TODO(ronag): this.pause()?
|
||||
this.emit('pause');
|
||||
});
|
||||
|
||||
stream.on('resume', () => {
|
||||
// TODO(ronag): this.resume()?
|
||||
this.emit('resume');
|
||||
});
|
||||
|
||||
// When we try to consume some more bytes, simply unpause the
|
||||
// underlying stream.
|
||||
|
||||
@@ -0,0 +1,32 @@
|
||||
'use strict';
|
||||
const common = require('../common');
|
||||
const assert = require('assert');
|
||||
|
||||
const Readable = require('_stream_readable');
|
||||
const EE = require('events').EventEmitter;
|
||||
|
||||
const oldStream = new EE();
|
||||
oldStream.pause = () => {};
|
||||
oldStream.resume = () => {};
|
||||
|
||||
{
|
||||
const r = new Readable({ autoDestroy: true })
|
||||
.wrap(oldStream)
|
||||
.on('error', common.mustCall(() => {
|
||||
assert.strictEqual(r._readableState.errorEmitted, true);
|
||||
assert.strictEqual(r._readableState.errored, true);
|
||||
assert.strictEqual(r.destroyed, true);
|
||||
}));
|
||||
oldStream.emit('error', new Error());
|
||||
}
|
||||
|
||||
{
|
||||
const r = new Readable({ autoDestroy: false })
|
||||
.wrap(oldStream)
|
||||
.on('error', common.mustCall(() => {
|
||||
assert.strictEqual(r._readableState.errorEmitted, true);
|
||||
assert.strictEqual(r._readableState.errored, true);
|
||||
assert.strictEqual(r.destroyed, false);
|
||||
}));
|
||||
oldStream.emit('error', new Error());
|
||||
}
|
||||
Reference in new issue
Block a user