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: #34102
Reviewed-By: Matteo Collina <matteo.collina@gmail.com>
Reviewed-By: Luigi Pinca <luigipinca@gmail.com>
Reviewed-By: Anna Henningsen <anna@addaleax.net>

Author: ronag

lib/_stream_readable.js

@@ -66,7 +66,6 @@ ObjectSetPrototypeOf(Readable.prototype, Stream.prototype);
 ObjectSetPrototypeOf(Readable,Stream);
 
 const{ errorOrDestroy }=destroyImpl;
-constkProxyEvents=['error','close','destroy','pause','resume'];
 
 functionprependListener(emitter,event,fn){
 // Sadly this is not cacheable as some libraries bundle their own
@@ -1034,10 +1033,29 @@ Readable.prototype.wrap = function(stream) {
 }
 }
 
-// Proxy certain important events.
-for(constkProxyEventofkProxyEvents){
-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.

test/parallel/test-stream2-readable-wrap-error.js

@@ -0,0 +1,32 @@
+'use strict';
+constcommon=require('../common');
+constassert=require('assert');
+
+constReadable=require('_stream_readable');
+constEE=require('events').EventEmitter;
+
+constoldStream=newEE();
+oldStream.pause=()=>{};
+oldStream.resume=()=>{};
+
+{
+constr=newReadable({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',newError());
+}
+
+{
+constr=newReadable({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',newError());
+}