Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
22 changes: 22 additions & 0 deletions doc/api/diagnostics_channel.md
Original file line number Diff line number Diff line change
Expand Up @@ -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}
Expand Down
25 changes: 25 additions & 0 deletions lib/_http_client.js
Original file line number Diff line number Diff line change
Expand Up @@ -62,6 +62,8 @@ const { URL, urlToHttpOptions, isURL } = require('internal/url');
const {
kOutHeaders,
kNeedDrain,
kOnOutgoingBodyChunkSent,
kOnOutgoingBodySent,
isTraceHTTPEnabled,
traceBegin,
traceEnd,
Expand Down Expand Up @@ -101,6 +103,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) {
Expand Down Expand Up @@ -652,6 +658,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;
Expand Down
15 changes: 15 additions & 0 deletions lib/_http_common.js
Original file line number Diff line number Diff line change
Expand Up @@ -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');
Expand All @@ -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;
Expand All @@ -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
Expand Down Expand Up @@ -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);
Expand All @@ -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);
Expand Down
19 changes: 18 additions & 1 deletion lib/_http_outgoing.js
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down Expand Up @@ -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 = () => {};

Expand Down Expand Up @@ -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);
Expand Down Expand Up @@ -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;
Expand Down Expand Up @@ -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.');
Expand Down
2 changes: 2 additions & 0 deletions lib/internal/http.js
Original file line number Diff line number Diff line change
Expand Up @@ -268,6 +268,8 @@ function getGlobalAgent(proxyEnv, Agent) {
}

module.exports = {
kOnOutgoingBodyChunkSent: Symbol('kOnOutgoingBodyChunkSent'),
kOnOutgoingBodySent: Symbol('kOnOutgoingBodySent'),
kOutHeaders: Symbol('kOutHeaders'),
kNeedDrain: Symbol('kNeedDrain'),
kProxyConfig: Symbol('kProxyConfig'),
Expand Down
Loading
Loading