diff --git a/doc/api/diagnostics_channel.md b/doc/api/diagnostics_channel.md index e5ace9abcfdd..7861e1ec7670 100644 --- a/doc/api/diagnostics_channel.md +++ b/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} diff --git a/lib/_http_common.js b/lib/_http_common.js index d5e7bdedee39..7af044d33d59 100644 --- a/lib/_http_common.js +++ b/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); diff --git a/lib/internal/inspector/network_http.js b/lib/internal/inspector/network_http.js index 1bb08762bdb4..e385805b710e 100644 --- a/lib/internal/inspector/network_http.js +++ b/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], ]); diff --git a/test/parallel/test-diagnostics-channel-http-client-response-body.js b/test/parallel/test-diagnostics-channel-http-client-response-body.js new file mode 100644 index 000000000000..2d5d7758e56f --- /dev/null +++ b/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()); + })); + })); +})); diff --git a/test/parallel/test-inspector-network-http.js b/test/parallel/test-inspector-network-http.js index 88d717d83c89..a3ba7cca1e1a 100644 --- a/test/parallel/test-inspector-network-http.js +++ b/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'); })); }