http: add highWaterMark opt in http.createServer

doc/api/http.md

@@ -3273,6 +3273,8 @@ changes:
   * `uniqueHeaders` {Array} A list of response headers that should be sent only
     once. If the header's value is an array, the items will be joined
     using `; `.
+  * `highWaterMark` {number} Optionally overrides the default `highWaterMark`
+    value of both `IncomingMessage` and `ServerResponse`.
 
 * `requestListener` {Function}
 

lib/_http_incoming.js

@@ -55,7 +55,7 @@ function IncomingMessage(socket) {
 
   if (socket) {
     streamOptions = {
-      highWaterMark: socket.readableHighWaterMark,
+      highWaterMark: socket.server?.highWaterMark ?? socket.readableHighWaterMark,
     };
   }
 

lib/_http_outgoing.js

@@ -1171,6 +1171,7 @@ function(err, event) {
 };
 
 module.exports = {
+  kHighWaterMark,
   kUniqueHeaders,
   parseUniqueHeadersOption,
   validateHeaderName,

lib/_http_server.js

@@ -49,6 +49,7 @@ const {
 } = require('_http_common');
 const { ConnectionsList } = internalBinding('http_parser');
 const {
+  kHighWaterMark,
   kUniqueHeaders,
   parseUniqueHeadersOption,
   OutgoingMessage,
@@ -86,6 +87,7 @@ const {
   validateBoolean,
   validateLinkHeaderValue,
   validateObject,
+  validateNumber,
 } = require('internal/validators');
 const Buffer = require('buffer').Buffer;
 const { setInterval, clearInterval } = require('timers');
@@ -105,6 +107,7 @@ const {
   startPerf,
   stopPerf,
 } = require('internal/perf/observe');
+const { getDefaultHighWaterMark } = require('internal/streams/state');
 
 const STATUS_CODES = {
   100: 'Continue',                   // RFC 7231 6.2.1
@@ -484,6 +487,14 @@ function storeHTTPOptions(options) {
     validateBoolean(joinDuplicateHeaders, 'options.joinDuplicateHeaders');
   }
   this.joinDuplicateHeaders = joinDuplicateHeaders;
+
+  const highWaterMark = options.highWaterMark;
+  if (highWaterMark !== undefined) {
+    validateInteger(highWaterMark, 'highWaterMark', 0);
+    this.highWaterMark = highWaterMark;
+  } else {
+    this.highWaterMark = getDefaultHighWaterMark();
+  }
 }
 
 function setupConnectionsTracking(server) {
@@ -1027,6 +1038,9 @@ function parserOnIncoming(server, socket, state, req, keepAlive) {
 
   res.shouldKeepAlive = keepAlive;
   res[kUniqueHeaders] = server[kUniqueHeaders];
+  if (server.highWaterMark) {
+    res[kHighWaterMark] = server.highWaterMark;
+  }
 
   if (onRequestStartChannel.hasSubscribers) {
     onRequestStartChannel.publish({

lib/http.js

@@ -54,6 +54,7 @@ let maxHeaderSize;
  *   maxHeaderSize?: number;
  *   requireHostHeader?: boolean;
  *   joinDuplicateHeaders?: boolean;
+ *   highWaterMark?: number;
  *   }} [opts]
  * @param {Function} [requestListener]
  * @returns {Server}

test/parallel/test-http-server-options-highwatermark.js

@@ -0,0 +1,28 @@
+// Flags: --expose-internals
+'use strict';
+const common = require('../common');
+const assert = require('assert');
+const http = require('http');
+const { kHighWaterMark } = require('_http_outgoing');
+
+const { getDefaultHighWaterMark } = require('internal/streams/state');
+
+const server = http.createServer({
+  highWaterMark: getDefaultHighWaterMark() * 2,
+}, common.mustCall((req, res) => {
+  assert.strictEqual(req._readableState.highWaterMark, getDefaultHighWaterMark() * 2);
+  assert.strictEqual(res[kHighWaterMark], getDefaultHighWaterMark() * 2);
+  res.statusCode = 200;
+  res.end();
+}));
+
+server.listen(0, common.mustCall(() => {
+  http.get({
+    port: server.address().port,
+  }, (res) => {
+    assert.strictEqual(res.statusCode, 200);
+    res.resume().on('end', common.mustCall(() => {
+      server.close();
+    }));
+  });
+}));