http: add captureRejection support to OutgoingMessage

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/_http_outgoing.js

@@ -32,6 +32,7 @@ const {
 
 const{ getDefaultHighWaterMark }=require('internal/streams/state');
 constassert=require('internal/assert');
+constEE=require('events');
 constStream=require('stream');
 constinternalUtil=require('internal/util');
 const{ kOutHeaders, utcDate, kNeedDrain }=require('internal/http');
@@ -884,6 +885,11 @@ OutgoingMessage.prototype.pipe = function pipe() {
 this.emit('error',newERR_STREAM_CANNOT_PIPE());
 };
 
+OutgoingMessage.prototype[EE.captureRejectionSymbol]=
+function(err,event){
+this.destroy(err);
+};
+
 module.exports={
  OutgoingMessage
 };

test/parallel/test-http-outgoing-message-capture-rejection.js

@@ -0,0 +1,89 @@
+'use strict';
+
+constcommon=require('../common');
+constassert=require('assert');
+constevents=require('events');
+const{ createServer, request }=require('http');
+
+events.captureRejections=true;
+
+{
+constserver=createServer(common.mustCall((req,res)=>{
+const_err=newError('kaboom');
+res.on('drain',common.mustCall(async()=>{
+throw_err;
+}));
+
+res.socket.on('error',common.mustCall((err)=>{
+assert.strictEqual(err,_err);
+}));
+
+// Write until there is space in the buffer
+while(res.write('hello')){}
+}));
+
+server.listen(0,common.mustCall(()=>{
+constreq=request({
+method: 'GET',
+host: server.address().host,
+port: server.address().port
+});
+
+req.end();
+
+req.on('response',common.mustCall((res)=>{
+res.on('aborted',common.mustCall());
+res.resume();
+server.close();
+}));
+}));
+}
+
+{
+let_res;
+letshouldEnd=false;
+// Not using mustCall here, because it is OS-dependant.
+constserver=createServer((req,res)=>{
+// So that we cleanly stop
+_res=res;
+
+if(shouldEnd){
+res.end();
+}
+});
+
+server.listen(0,common.mustCall(()=>{
+const_err=newError('kaboom');
+
+constreq=request({
+method: 'POST',
+host: server.address().host,
+port: server.address().port
+});
+
+req.on('response',common.mustNotCall((res)=>{
+// So that we cleanly stop
+res.resume();
+server.close();
+}));
+
+req.on('error',common.mustCall((err)=>{
+server.close();
+// On some variants of Windows, this can happen before
+// the server has received the request.
+if(_res){
+_res.end();
+}else{
+shouldEnd=true;
+}
+assert.strictEqual(err,_err);
+}));
+
+req.on('drain',common.mustCall(async()=>{
+throw_err;
+}));
+
+// Write until there is space in the buffer
+while(req.write('hello')){}
+}));
+}