stream: finished on closed OutgoingMessage

finished should invoke callback on closed OutgoingMessage the
same way as for regular streams.

Fixes: #34301

PR-URL: #34313
Fixes: #34274
Reviewed-By: Matteo Collina <matteo.collina@gmail.com>
Reviewed-By: Luigi Pinca <luigipinca@gmail.com>

Author: ronag

Diff for: lib/_http_client.js

@@ -49,7 +49,7 @@ const Agent = require('_http_agent');
 const { Buffer } = require('buffer');
 const { defaultTriggerAsyncIdScope } = require('internal/async_hooks');
 const { URL, urlToOptions, searchParamsSymbol } = require('internal/url');
-const { kOutHeaders, kNeedDrain, kClosed } = require('internal/http');
+const { kOutHeaders, kNeedDrain } = require('internal/http');
 const { connResetException, codes } = require('internal/errors');
 const {
   ERR_HTTP_HEADERS_SENT,
@@ -387,7 +387,7 @@ function _destroy(req, socket, err) {
     if (err) {
       req.emit('error', err);
     }
-    req[kClosed] = true;
+    req._closed = true;
     req.emit('close');
   }
 }
@@ -430,7 +430,7 @@ function socketCloseListener() {
         res.emit('error', connResetException('aborted'));
       }
     }
-    req[kClosed] = true;
+    req._closed = true;
     req.emit('close');
     if (!res.aborted && res.readable) {
       res.on('end', function() {
@@ -450,7 +450,7 @@ function socketCloseListener() {
       req.socket._hadError = true;
       req.emit('error', connResetException('socket hang up'));
     }
-    req[kClosed] = true;
+    req._closed = true;
     req.emit('close');
   }
 
@@ -555,7 +555,7 @@ function socketOnData(d) {
 
       req.emit(eventName, res, socket, bodyHead);
       req.destroyed = true;
-      req[kClosed] = true;
+      req._closed = true;
       req.emit('close');
     } else {
       // Requested Upgrade or used CONNECT method, but have no handler.
@@ -727,7 +727,7 @@ function requestOnPrefinish() {
 }
 
 function emitFreeNT(req) {
-  req[kClosed] = true;
+  req._closed = true;
   req.emit('close');
   if (req.res) {
     req.res.emit('close');

Diff for: lib/_http_outgoing.js

@@ -36,7 +36,7 @@ const assert = require('internal/assert');
 const EE = require('events');
 const Stream = require('stream');
 const internalUtil = require('internal/util');
-const { kOutHeaders, utcDate, kNeedDrain, kClosed } = require('internal/http');
+const { kOutHeaders, utcDate, kNeedDrain } = require('internal/http');
 const { Buffer } = require('buffer');
 const common = require('_http_common');
 const checkIsHttpToken = common._checkIsHttpToken;
@@ -117,7 +117,7 @@ function OutgoingMessage() {
   this.finished = false;
   this._headerSent = false;
   this[kCorked] = 0;
-  this[kClosed] = false;
+  this._closed = false;
 
   this.socket = null;
   this._header = null;
@@ -664,7 +664,7 @@ function onError(msg, err, callback) {
 
 function emitErrorNt(msg, err, callback) {
   callback(err);
-  if (typeof msg.emit === 'function' && !msg[kClosed]) {
+  if (typeof msg.emit === 'function' && !msg._closed) {
     msg.emit('error', err);
   }
 }

Diff for: lib/_http_server.js

@@ -49,7 +49,6 @@ const { OutgoingMessage } = require('_http_outgoing');
 const {
   kOutHeaders,
   kNeedDrain,
-  kClosed,
   emitStatistics
 } = require('internal/http');
 const {
@@ -216,7 +215,7 @@ function onServerResponseClose() {
   // Fortunately, that requires only a single if check. :-)
   if (this._httpMessage) {
     this._httpMessage.destroyed = true;
-    this._httpMessage[kClosed] = true;
+    this._httpMessage._closed = true;
     this._httpMessage.emit('close');
   }
 }
@@ -733,7 +732,7 @@ function resOnFinish(req, res, socket, state, server) {
 
 function emitCloseNT(self) {
   self.destroyed = true;
-  self[kClosed] = true;
+  self._closed = true;
   self.emit('close');
 }
 

Diff for: lib/internal/http.js

@@ -51,7 +51,6 @@ function emitStatistics(statistics) {
 module.exports = {
   kOutHeaders: Symbol('kOutHeaders'),
   kNeedDrain: Symbol('kNeedDrain'),
-  kClosed: Symbol('kClosed'),
   nowDate,
   utcDate,
   emitStatistics

Diff for: lib/internal/streams/end-of-stream.js

@@ -147,7 +147,8 @@ function eos(stream, options, callback) {
   if (options.error !== false) stream.on('error', onerror);
   stream.on('close', onclose);
 
-  const closed = (
+  // _closed is for OutgoingMessage which is not a proper Writable.
+  const closed = (!wState && !rState && stream._closed === true) || (
     (wState && wState.closed) ||
     (rState && rState.closed) ||
     (wState && wState.errorEmitted) ||

Diff for: test/parallel/test-stream-finished.js

@@ -13,6 +13,7 @@ const assert = require('assert');
 const EE = require('events');
 const fs = require('fs');
 const { promisify } = require('util');
+const http = require('http');
 
 {
   const rs = new Readable({
@@ -480,3 +481,21 @@ testClosed((opts) => new Writable({ write() {}, ...opts }));
     finished(p, common.mustNotCall());
   }));
 }
+
+{
+  const server = http.createServer((req, res) => {
+    res.on('close', () => {
+      finished(res, common.mustCall(() => {
+        server.close();
+      }));
+    });
+    res.end();
+  })
+  .listen(0, function() {
+    http.request({
+      method: 'GET',
+      port: this.address().port
+    }).end()
+      .on('response', common.mustCall());
+  });
+}