stream: add support for captureRejection option

PR-URL: #27867
Reviewed-By: Benjamin Gruenbaum <benjamingr@gmail.com>
Reviewed-By: James M Snell <jasnell@gmail.com>
Reviewed-By: Jeremiah Senkpiel <fishrock123@rocketmail.com>
Reviewed-By: Anna Henningsen <anna@addaleax.net>
Reviewed-By: Michaël Zasso <targos@protonmail.com>

Author: mcollina

lib/_stream_readable.js

@@ -190,7 +190,7 @@ function Readable(options) {
 this._destroy=options.destroy;
 }
 
-Stream.call(this);
+Stream.call(this,options);
 }
 
 ObjectDefineProperty(Readable.prototype,'destroyed',{
@@ -233,6 +233,14 @@ Readable.prototype._destroy = function(err, cb) {
 cb(err);
 };
 
+Readable.prototype[EE.captureRejectionSymbol]=function(err){
+// TODO(mcollina): remove the destroyed if once errorEmitted lands in
+// Readable.
+if(!this.destroyed){
+this.destroy(err);
+}
+};
+
 // Manually shove something into the read() buffer.
 // This returns true if the highWaterMark has not been hit yet,
 // similar to how Writable.write() returns true if you should

lib/_stream_writable.js

@@ -35,6 +35,7 @@ module.exports = Writable;
 Writable.WritableState=WritableState;
 
 constinternalUtil=require('internal/util');
+constEE=require('events');
 constStream=require('stream');
 const{ Buffer }=require('buffer');
 constdestroyImpl=require('internal/streams/destroy');
@@ -254,7 +255,7 @@ function Writable(options) {
 this._final=options.final;
 }
 
-Stream.call(this);
+Stream.call(this,options);
 }
 
 // Otherwise people can pipe Writable streams, which is just wrong.
@@ -804,3 +805,7 @@ Writable.prototype._undestroy = destroyImpl.undestroy;
 Writable.prototype._destroy=function(err,cb){
 cb(err);
 };
+
+Writable.prototype[EE.captureRejectionSymbol]=function(err){
+this.destroy(err);
+};

lib/internal/streams/legacy.js

@@ -6,8 +6,8 @@ const {
 
 constEE=require('events');
 
-functionStream(){
-EE.call(this);
+functionStream(opts){
+EE.call(this,opts);
 }
 ObjectSetPrototypeOf(Stream.prototype,EE.prototype);
 ObjectSetPrototypeOf(Stream,EE);

test/parallel/test-stream-catch-rejections.js

@@ -0,0 +1,52 @@
+'use strict';
+
+constcommon=require('../common');
+conststream=require('stream');
+constassert=require('assert');
+
+{
+constr=newstream.Readable({
+captureRejections: true,
+read(){
+this.push('hello');
+this.push('world');
+this.push(null);
+}
+});
+
+consterr=newError('kaboom');
+
+r.on('error',common.mustCall((_err)=>{
+assert.strictEqual(err,_err);
+assert.strictEqual(r.destroyed,true);
+}));
+
+r.on('data',async()=>{
+throwerr;
+});
+}
+
+{
+constw=newstream.Writable({
+captureRejections: true,
+highWaterMark: 1,
+write(chunk,enc,cb){
+cb();
+}
+});
+
+consterr=newError('kaboom');
+
+w.write('hello',()=>{
+w.write('world');
+});
+
+w.on('error',common.mustCall((_err)=>{
+assert.strictEqual(err,_err);
+assert.strictEqual(w.destroyed,true);
+}));
+
+w.on('drain',common.mustCall(async()=>{
+throwerr;
+},2));
+}