inspector,http: support builtin http request bodies

doc/api/diagnostics_channel.md

@@ -1576,13 +1576,36 @@ 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 chunk of the client request body is being sent.
+
+##### Event: `'http.client.request.bodySent'`
+
+* `request` {http.ClientRequest}
+
+Emitted after the client request body has been fully sent, if a request body
+was written.
+
 ##### 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}
+
+Emitted when a chunk of the client response body is received.
+
 ##### Event: `'http.client.response.finish'`
 
 * `request` {http.ClientRequest}

lib/_http_client.js

@@ -60,6 +60,7 @@ const { Buffer } = require('buffer');
 const { defaultTriggerAsyncIdScope } = require('internal/async_hooks');
 const { URL, urlToHttpOptions, isURL } = require('internal/url');
 const {
+  kIsClientRequest,
   kOutHeaders,
   kNeedDrain,
   isTraceHTTPEnabled,
@@ -192,6 +193,7 @@ function rewriteForProxiedHttp(req, reqOptions) {
 
 function ClientRequest(input, options, cb) {
   OutgoingMessage.call(this);
+  this[kIsClientRequest] = true;
 
   if (typeof input === 'string') {
     const urlStr = input;

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');
@@ -50,6 +51,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 +124,7 @@ function parserOnHeadersComplete(versionMajor, versionMinor, headers, method,
     // client only
     incoming.statusCode = statusCode;
     incoming.statusMessage = statusMessage;
+    incoming.req = socket?._httpMessage;
   }
 
   return parser.onIncoming(incoming, shouldKeepAlive);
@@ -134,6 +139,13 @@ function parserOnBody(b) {
 
   // Pretend this was the result of a stream._read call.
   if (!stream._dumped) {
+    if (stream.req && onClientResponseBodyChunkReceivedChannel.hasSubscribers) {
+      onClientResponseBodyChunkReceivedChannel.publish({
+        request: stream.req,
+        response: stream,
+        chunk: b,
+      });
+    }
     const ret = stream.push(b);
     if (!ret)
       readStop(this.socket);

lib/_http_outgoing.js

@@ -36,9 +36,10 @@ const {
 
 const { getDefaultHighWaterMark } = require('internal/streams/state');
 const assert = require('internal/assert');
+const dc = require('diagnostics_channel');
 const EE = require('events');
 const Stream = require('stream');
-const { kOutHeaders, utcDate, kNeedDrain } = require('internal/http');
+const { kIsClientRequest, kOutHeaders, utcDate, kNeedDrain } = require('internal/http');
 const { Buffer } = require('buffer');
 const {
   _checkIsHttpToken: checkIsHttpToken,
@@ -86,6 +87,12 @@ const kBytesWritten = Symbol('kBytesWritten');
 const kErrored = Symbol('errored');
 const kHighWaterMark = Symbol('kHighWaterMark');
 const kRejectNonStandardBodyWrites = Symbol('kRejectNonStandardBodyWrites');
+const kClientRequestBodyChunksWritten = Symbol('kClientRequestBodyChunksWritten');
+
+const onClientRequestBodyChunkSentChannel =
+  dc.channel('http.client.request.bodyChunkSent');
+const onClientRequestBodySentChannel =
+  dc.channel('http.client.request.bodySent');
 
 const nop = () => {};
 
@@ -950,6 +957,17 @@ function write_(msg, chunk, encoding, callback, fromEnd) {
     }
   }
 
+  if (msg[kIsClientRequest]) {
+    msg[kClientRequestBodyChunksWritten] = true;
+    if (onClientRequestBodyChunkSentChannel.hasSubscribers) {
+      onClientRequestBodyChunkSentChannel.publish({
+        request: msg,
+        chunk,
+        encoding,
+      });
+    }
+  }
+
   if (!fromEnd && msg.socket && !msg.socket.writableCorked) {
     msg.socket.cork();
     process.nextTick(connectionCorkNT, msg.socket);
@@ -1103,6 +1121,12 @@ OutgoingMessage.prototype.end = function end(chunk, encoding, callback) {
 
   this.finished = true;
 
+  if (this[kIsClientRequest] &&
+      this[kClientRequestBodyChunksWritten] &&
+      onClientRequestBodySentChannel.hasSubscribers) {
+    onClientRequestBodySentChannel.publish({ request: this });
+  }
+
   // 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

@@ -262,6 +262,7 @@ function getGlobalAgent(proxyEnv, Agent) {
 }
 
 module.exports = {
+  kIsClientRequest: Symbol('kIsClientRequest'),
   kOutHeaders: Symbol('kOutHeaders'),
   kNeedDrain: Symbol('kNeedDrain'),
   kProxyConfig: Symbol('kProxyConfig'),

lib/internal/inspector/network_http.js

@@ -2,7 +2,9 @@
 
 const {
   ArrayIsArray,
+  ArrayPrototypePush,
   DateNow,
+  MathMax,
   ObjectEntries,
   String,
   StringPrototypeStartsWith,
@@ -17,10 +19,13 @@ const {
   registerDiagnosticChannels,
   sniffMimeType,
 } = require('internal/inspector/network');
+const { getStructuredStack } = require('internal/util');
 const { Network } = require('inspector');
-const EventEmitter = require('events');
+const { Buffer } = require('buffer');
 
 const kRequestUrl = Symbol('kRequestUrl');
+const kRequestWillBeSent = Symbol('kRequestWillBeSent');
+const kInitiator = Symbol('kInitiator');
 
 function isAbsoluteURLPath(path) {
   return typeof path === 'string' &&
@@ -67,37 +72,81 @@ const convertHeaderObject = (headers = {}) => {
   return [dict, host, charset, mimeType];
 };
 
+function createInitiator() {
+  const callSites = getStructuredStack();
+  const callFrames = [];
+  for (let i = 0; i < callSites.length; i++) {
+    const callSite = callSites[i];
+    ArrayPrototypePush(callFrames, {
+      functionName: callSite.getFunctionName() ?? callSite.getMethodName() ?? '',
+      scriptId: '',
+      url: callSite.getScriptNameOrSourceURL() ?? callSite.getFileName() ?? '',
+      lineNumber: MathMax((callSite.getLineNumber() ?? 1) - 1, 0),
+      columnNumber: MathMax((callSite.getColumnNumber() ?? 1) - 1, 0),
+    });
+  }
+  return {
+    type: 'script',
+    stack: { callFrames },
+  };
+}
+
 /**
- * When a client request is created, emit Network.requestWillBeSent event.
- * https://chromedevtools.github.io/devtools-protocol/1-3/Network/#event-requestWillBeSent
+ * When a client request is created, assign its inspector request id.
  * @param {{ request: import('http').ClientRequest }} event
  */
 function onClientRequestCreated({ request }) {
   request[kInspectorRequestId] = getNextRequestId();
+  // Ensure that the stack obtained here is the one created at the time of actual construction.
+  request[kInitiator] = createInitiator();
+}
+
+/**
+ * Emit Network.requestWillBeSent once the request body state is known.
+ * @param {import('http').ClientRequest} request
+ * @param {boolean} hasPostData
+ */
+function emitRequestWillBeSent(request, hasPostData) {
+  if (request[kRequestWillBeSent] ||
+      typeof request[kInspectorRequestId] !== 'string') {
+    return;
+  }
 
   const { 0: headers, 1: host, 2: charset } = convertHeaderObject(request.getHeaders());
   const url = getRequestURL(request, host);
   request[kRequestUrl] = url;
+  request[kRequestWillBeSent] = true;
 
   Network.requestWillBeSent({
     requestId: request[kInspectorRequestId],
     timestamp: getMonotonicTime(),
     wallTime: DateNow(),
     charset,
+    initiator: request[kInitiator],
     request: {
       url,
       method: request.method,
       headers,
+      hasPostData,
     },
   });
 }
 
+/**
+ * When a client request starts without a body, emit Network.requestWillBeSent.
+ * @param {{ request: import('http').ClientRequest }} event
+ */
+function onClientRequestStart({ request }) {
+  emitRequestWillBeSent(request, false);
+}
+
 /**
  * When a client request errors, emit Network.loadingFailed event.
  * https://chromedevtools.github.io/devtools-protocol/1-3/Network/#event-loadingFailed
  * @param {{ request: import('http').ClientRequest, error: any }} event
  */
 function onClientRequestError({ request, error }) {
+  emitRequestWillBeSent(request, false);
   if (typeof request[kInspectorRequestId] !== 'string') {
     return;
   }
@@ -109,12 +158,73 @@ function onClientRequestError({ request, error }) {
   });
 }
 
+/**
+ * When a chunk of the request body is being sent, cache it until
+ * `getRequestPostData` request.
+ * https://chromedevtools.github.io/devtools-protocol/1-3/Network/#method-getRequestPostData
+ * @param {{ request: import('http').ClientRequest, chunk: Uint8Array | string, encoding?: string }} event
+ */
+function onClientRequestBodyChunkSent({ request, chunk, encoding }) {
+  if (typeof request[kInspectorRequestId] !== 'string') {
+    return;
+  }
+
+  emitRequestWillBeSent(request, true);
+
+  const buffer = typeof chunk === 'string' ? Buffer.from(chunk, encoding) : Buffer.from(chunk);
+  Network.dataSent({
+    requestId: request[kInspectorRequestId],
+    timestamp: getMonotonicTime(),
+    dataLength: buffer.byteLength,
+    data: buffer,
+  });
+}
+
+/**
+ * Mark a request body as fully sent.
+ * @param {{ request: import('http').ClientRequest }} event
+ */
+function onClientRequestBodySent({ request }) {
+  if (typeof request[kInspectorRequestId] !== 'string') {
+    return;
+  }
+
+  Network.dataSent({
+    requestId: request[kInspectorRequestId],
+    timestamp: getMonotonicTime(),
+    dataLength: 0,
+    data: Buffer.alloc(0),
+    finished: true,
+  });
+}
+
+/**
+ * When a chunk of the response body is received, cache the raw bytes until
+ * `getResponseBody` request.
+ * https://chromedevtools.github.io/devtools-protocol/1-3/Network/#method-getResponseBody
+ * @param {{ request: import('http').ClientRequest, chunk: Uint8Array }} event
+ */
+function onClientResponseBodyChunkReceived({ request, chunk }) {
+  if (typeof request[kInspectorRequestId] !== 'string') {
+    return;
+  }
+
+  Network.dataReceived({
+    requestId: request[kInspectorRequestId],
+    timestamp: getMonotonicTime(),
+    dataLength: chunk.byteLength,
+    encodedDataLength: chunk.byteLength,
+    data: chunk,
+  });
+}
+
 /**
  * When response headers are received, emit Network.responseReceived event.
  * https://chromedevtools.github.io/devtools-protocol/1-3/Network/#event-responseReceived
  * @param {{ request: import('http').ClientRequest, error: any }} event
  */
 function onClientResponseFinish({ request, response }) {
+  emitRequestWillBeSent(request, false);
   if (typeof request[kInspectorRequestId] !== 'string') {
     return;
   }
@@ -135,17 +245,6 @@ function onClientResponseFinish({ request, response }) {
     },
   });
 
-  // Unlike response.on('data', ...), this does not put the stream into flowing mode.
-  EventEmitter.prototype.on.call(response, 'data', (chunk) => {
-    Network.dataReceived({
-      requestId: request[kInspectorRequestId],
-      timestamp: getMonotonicTime(),
-      dataLength: chunk.byteLength,
-      encodedDataLength: chunk.byteLength,
-      data: chunk,
-    });
-  });
-
   // Wait until the response body is consumed by user code.
   response.once('end', () => {
     Network.loadingFinished({
@@ -157,6 +256,10 @@ function onClientResponseFinish({ request, response }) {
 
 module.exports = registerDiagnosticChannels([
   ['http.client.request.created', onClientRequestCreated],
+  ['http.client.request.start', onClientRequestStart],
+  ['http.client.request.bodyChunkSent', onClientRequestBodyChunkSent],
+  ['http.client.request.bodySent', onClientRequestBodySent],
   ['http.client.request.error', onClientRequestError],
+  ['http.client.response.bodyChunkReceived', onClientResponseBodyChunkReceived],
   ['http.client.response.finish', onClientResponseFinish],
 ]);

node.gyp

@@ -1418,6 +1418,7 @@
             # TODO(legendecas): make node_inspector.gypi a dependable target.
             '<(SHARED_INTERMEDIATE_DIR)', # for inspector
             '<(SHARED_INTERMEDIATE_DIR)/src', # for inspector
+            '<(SHARED_INTERMEDIATE_DIR)/inspector-generated-output-root/include',
           ],
           'dependencies': [
             'deps/inspector_protocol/inspector_protocol.gyp:crdtp',