inspector: capture HTTP body before decoding

doc/api/diagnostics_channel.md

@@ -1586,6 +1586,15 @@ Emitted when client starts a request.
 
 Emitted when an error occurs during a client request.
 
+##### Event: `'http.client.response.bodyChunk'`
+
+* `request` {http.ClientRequest}
+* `response` {http.IncomingMessage}
+* `chunk` {Buffer}
+
+Emitted before each response body chunk is added to the `IncomingMessage`
+readable buffer.
+
 ##### Event: `'http.client.response.finish'`
 
 * `request` {http.ClientRequest}

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,8 @@ const kOnMessageComplete = HTTPParser.kOnMessageComplete | 0;
 const kOnExecute = HTTPParser.kOnExecute | 0;
 const kOnTimeout = HTTPParser.kOnTimeout | 0;
 
+const responseBodyChannel = dc.channel('http.client.response.bodyChunk');
+
 const MAX_HEADER_PAIRS = 2000;
 
 // Only called in the slow case where slow means
@@ -134,6 +137,14 @@ function parserOnBody(b) {
 
   // Pretend this was the result of a stream._read call.
   if (!stream._dumped) {
+    // Response parsers retain their ClientRequest in `outgoing`.
+    if (responseBodyChannel.hasSubscribers && this.outgoing !== null) {
+      responseBodyChannel.publish({
+        request: this.outgoing,
+        response: stream,
+        chunk: b,
+      });
+    }
     const ret = stream.push(b);
     if (!ret)
       readStop(this.socket);

lib/internal/inspector/network_http.js

@@ -18,7 +18,6 @@ const {
   sniffMimeType,
 } = require('internal/inspector/network');
 const { Network } = require('inspector');
-const EventEmitter = require('events');
 const { kEmptyObject } = require('internal/util');
 
 const kRequestUrl = Symbol('kRequestUrl');
@@ -110,6 +109,20 @@ function onClientRequestError({ request, error }) {
   });
 }
 
+function onResponseBody({ request, chunk }) {
+  if (request[kInspectorRequestId] === undefined) {
+    return;
+  }
+
+  Network.dataReceived({
+    requestId: request[kInspectorRequestId],
+    timestamp: getMonotonicTime(),
+    dataLength: chunk.length,
+    encodedDataLength: chunk.length,
+    data: chunk,
+  });
+}
+
 /**
  * When response headers are received, emit Network.responseReceived event.
  * https://chromedevtools.github.io/devtools-protocol/1-3/Network/#event-responseReceived
@@ -136,17 +149,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({
@@ -159,5 +161,6 @@ function onClientResponseFinish({ request, response }) {
 module.exports = registerDiagnosticChannels([
   ['http.client.request.created', onClientRequestCreated],
   ['http.client.request.error', onClientRequestError],
+  ['http.client.response.bodyChunk', onResponseBody],
   ['http.client.response.finish', onClientResponseFinish],
 ]);

test/parallel/test-diagnostics-channel-http-client-response-body.js

@@ -0,0 +1,43 @@
+'use strict';
+
+const common = require('../common');
+const assert = require('node:assert');
+const dc = require('node:diagnostics_channel');
+const http = require('node:http');
+
+const body = Buffer.from('A\u{1F642}B');
+const raw = [];
+const decoded = [];
+let clientRequest;
+let clientResponse;
+
+dc.subscribe('http.client.response.bodyChunk', common.mustCallAtLeast((message) => {
+  const { request, response, chunk } = message;
+  assert.strictEqual(request, clientRequest);
+  assert.strictEqual(response, clientResponse);
+  assert.strictEqual(response.req, request);
+  assert(Buffer.isBuffer(chunk));
+  raw.push(Buffer.from(chunk));
+}));
+
+const server = http.createServer(common.mustCall((request, response) => {
+  request.resume();
+  response.write(body.subarray(0, 3));
+  setImmediate(() => response.end(body.subarray(3)));
+}));
+
+server.listen(0, common.mustCall(() => {
+  clientRequest = http.get({ port: server.address().port }, common.mustCall((response) => {
+    clientResponse = response;
+    response.setEncoding('utf8');
+    response.on('data', common.mustCallAtLeast((chunk) => {
+      assert.strictEqual(typeof chunk, 'string');
+      decoded.push(chunk);
+    }));
+    response.on('end', common.mustCall(() => {
+      assert.deepStrictEqual(Buffer.concat(raw), body);
+      assert.strictEqual(decoded.join(''), body.toString());
+      server.close(common.mustCall());
+    }));
+  }));
+}));

test/parallel/test-inspector-network-http.js

@@ -179,18 +179,20 @@ function verifyHttpResponse(response) {
 
   // Verifies that the inspector does not put the response into flowing mode.
   assert.strictEqual(response.readableFlowing, null);
+  response.setEncoding('utf8');
+  assert.strictEqual(response.readableFlowing, null);
   // Verifies that the data listener may be added at a later time, and it can
   // still observe the data in full.
   queueMicrotask(common.mustCall(() => {
-    response.on('data', (chunk) => {
+    response.on('data', common.mustCallAtLeast((chunk) => {
+      assert.strictEqual(typeof chunk, 'string');
       chunks.push(chunk);
-    });
+    }));
     assert.strictEqual(response.readableFlowing, true);
   }));
 
   response.on('end', common.mustCall(() => {
-    const body = Buffer.concat(chunks).toString();
-    assert.strictEqual(body, '\nhello world\n');
+    assert.strictEqual(chunks.join(''), '\nhello world\n');
   }));
 }