inspector,http: support http body tracking

Signed-off-by: GrinZero <774933704@qq.com>

Author: GrinZero

doc/api/diagnostics_channel.md

@@ -1579,13 +1579,35 @@ Unlike `http.client.request.start`, this event is emitted before the request has
 
 Emitted when client starts a request.
 
+##### Event: `'http.client.request.bodyChunkSent'`
+
+* `request` {http.ClientRequest}
+* `chunk` {Buffer|string|Uint8Array}
+* `encoding` {string|null|undefined}
+
+Emitted when a client request body chunk is sent.
+
+##### Event: `'http.client.request.bodySent'`
+
+* `request` {http.ClientRequest}
+
+Emitted when a client request body with at least one chunk has been sent.
+
 ##### Event: `'http.client.request.error'`
 
 * `request` {http.ClientRequest}
 * `error` {Error}
 
 Emitted when an error occurs during a client request.
 
+##### Event: `'http.client.response.bodyChunkReceived'`
+
+* `request` {http.ClientRequest}
+* `response` {http.IncomingMessage}
+* `chunk` {Buffer|Uint8Array}
+
+Emitted when raw bytes of a client response body chunk are received.
+
 ##### Event: `'http.client.response.finish'`
 
 * `request` {http.ClientRequest}

lib/_http_client.js

@@ -63,6 +63,8 @@ const { URL, urlToHttpOptions, isURL } = require('internal/url');
 const {
   kOutHeaders,
   kNeedDrain,
+  kOnOutgoingBodyChunkSent,
+  kOnOutgoingBodySent,
   isTraceHTTPEnabled,
   traceBegin,
   traceEnd,
@@ -102,6 +104,10 @@ const onClientRequestCreatedChannel = dc.channel('http.client.request.created');
 const onClientRequestStartChannel = dc.channel('http.client.request.start');
 const onClientRequestErrorChannel = dc.channel('http.client.request.error');
 const onClientResponseFinishChannel = dc.channel('http.client.response.finish');
+const onClientRequestBodyChunkSentChannel =
+  dc.channel('http.client.request.bodyChunkSent');
+const onClientRequestBodySentChannel =
+  dc.channel('http.client.request.bodySent');
 
 function emitErrorEvent(request, error) {
   if (onClientRequestErrorChannel.hasSubscribers) {
@@ -672,6 +678,25 @@ ClientRequest.prototype._implicitHeader = function _implicitHeader() {
                     this[kOutHeaders]);
 };
 
+ClientRequest.prototype[kOnOutgoingBodyChunkSent] =
+  function onOutgoingBodyChunkSent(chunk, encoding) {
+    if (onClientRequestBodyChunkSentChannel.hasSubscribers) {
+      onClientRequestBodyChunkSentChannel.publish({
+        request: this,
+        chunk,
+        encoding,
+      });
+    }
+  };
+
+ClientRequest.prototype[kOnOutgoingBodySent] = function onOutgoingBodySent() {
+  if (onClientRequestBodySentChannel.hasSubscribers) {
+    onClientRequestBodySentChannel.publish({
+      request: this,
+    });
+  }
+};
+
 ClientRequest.prototype.abort = function abort() {
   if (this.aborted) {
     return;

lib/_http_common.js

@@ -27,6 +27,7 @@ const {
   Uint8Array,
 } = primordials;
 const { setImmediate } = require('timers');
+const dc = require('diagnostics_channel');
 
 const { methods, allMethods, HTTPParser } = internalBinding('http_parser');
 const { getOptionValue } = require('internal/options');
@@ -42,6 +43,7 @@ const {
 
 const kIncomingMessage = Symbol('IncomingMessage');
 const kSkipPendingData = Symbol('SkipPendingData');
+const kClientRequest = Symbol('kClientRequest');
 const kOnMessageBegin = HTTPParser.kOnMessageBegin | 0;
 const kOnHeaders = HTTPParser.kOnHeaders | 0;
 const kOnHeadersComplete = HTTPParser.kOnHeadersComplete | 0;
@@ -50,6 +52,9 @@ const kOnMessageComplete = HTTPParser.kOnMessageComplete | 0;
 const kOnExecute = HTTPParser.kOnExecute | 0;
 const kOnTimeout = HTTPParser.kOnTimeout | 0;
 
+const onClientResponseBodyChunkReceivedChannel =
+  dc.channel('http.client.response.bodyChunkReceived');
+
 const MAX_HEADER_PAIRS = 2000;
 
 // Only called in the slow case where slow means
@@ -120,6 +125,7 @@ function parserOnHeadersComplete(versionMajor, versionMinor, headers, method,
     // client only
     incoming.statusCode = statusCode;
     incoming.statusMessage = statusMessage;
+    incoming[kClientRequest] = socket?._httpMessage;
   }
 
   return parser.onIncoming(incoming, shouldKeepAlive);
@@ -134,6 +140,15 @@ function parserOnBody(b) {
 
   // Pretend this was the result of a stream._read call.
   if (!stream._dumped) {
+    const request = stream[kClientRequest];
+    if (request !== undefined &&
+        onClientResponseBodyChunkReceivedChannel.hasSubscribers) {
+      onClientResponseBodyChunkReceivedChannel.publish({
+        request,
+        response: stream,
+        chunk: b,
+      });
+    }
     const ret = stream.push(b);
     if (!ret)
       readStop(this.socket);

lib/_http_outgoing.js

@@ -38,7 +38,13 @@ const { getDefaultHighWaterMark } = require('internal/streams/state');
 const assert = require('internal/assert');
 const EE = require('events');
 const Stream = require('stream');
-const { kOutHeaders, utcDate, kNeedDrain } = require('internal/http');
+const {
+  kOnOutgoingBodyChunkSent,
+  kOnOutgoingBodySent,
+  kOutHeaders,
+  utcDate,
+  kNeedDrain,
+} = require('internal/http');
 const { Buffer } = require('buffer');
 const {
   _checkIsHttpToken: checkIsHttpToken,
@@ -87,6 +93,7 @@ const kBytesWritten = Symbol('kBytesWritten');
 const kErrored = Symbol('errored');
 const kHighWaterMark = Symbol('kHighWaterMark');
 const kRejectNonStandardBodyWrites = Symbol('kRejectNonStandardBodyWrites');
+const kHasOutgoingBodyChunks = Symbol('kHasOutgoingBodyChunks');
 
 const nop = () => {};
 
@@ -155,6 +162,7 @@ function OutgoingMessage(options) {
   this[kErrored] = null;
   this[kHighWaterMark] = options?.highWaterMark ?? getDefaultHighWaterMark();
   this[kRejectNonStandardBodyWrites] = options?.rejectNonStandardBodyWrites ?? false;
+  this[kHasOutgoingBodyChunks] = false;
 }
 ObjectSetPrototypeOf(OutgoingMessage.prototype, Stream.prototype);
 ObjectSetPrototypeOf(OutgoingMessage, Stream);
@@ -1002,6 +1010,11 @@ function write_(msg, chunk, encoding, callback, fromEnd) {
     process.nextTick(connectionCorkNT, msg.socket);
   }
 
+  if (chunk.length !== 0) {
+    msg[kHasOutgoingBodyChunks] = true;
+    msg[kOnOutgoingBodyChunkSent]?.(chunk, encoding);
+  }
+
   let ret;
   if (msg.chunkedEncoding && chunk.length !== 0) {
     len ??= typeof chunk === 'string' ? Buffer.byteLength(chunk, encoding) : chunk.byteLength;
@@ -1151,6 +1164,10 @@ OutgoingMessage.prototype.end = function end(chunk, encoding, callback) {
 
   this.finished = true;
 
+  if (this[kHasOutgoingBodyChunks]) {
+    this[kOnOutgoingBodySent]?.();
+  }
+
   // There is the first message on the outgoing queue, and we've sent
   // everything to the socket.
   debug('outgoing message end.');

lib/internal/http.js

@@ -270,6 +270,8 @@ function getGlobalAgent(proxyEnv, Agent) {
 }
 
 module.exports = {
+  kOnOutgoingBodyChunkSent: Symbol('kOnOutgoingBodyChunkSent'),
+  kOnOutgoingBodySent: Symbol('kOnOutgoingBodySent'),
   kOutHeaders: Symbol('kOutHeaders'),
   kNeedDrain: Symbol('kNeedDrain'),
   kProxyConfig: Symbol('kProxyConfig'),