http2: add diagnostics channels for client stream request body · nodejs/node@a38e2f5

Emitted when a stream is received on the client.

##### Event: `'http2.client.stream.bodyChunkSent'`

* `stream` {ClientHttp2Stream}

* `writev` {boolean}

* `data` {Buffer | string | Buffer\[] | Object\[]}

* `chunk` {Buffer|string}

* `encoding` {string}

* `encoding` {string}

Emitted when a chunk of the client stream body is being sent.

##### Event: `'http2.client.stream.bodySent'`

* `stream` {ClientHttp2Stream}

Emitted after the client stream body has been fully sent.

##### Event: `'http2.client.stream.close'`

* `stream` {ClientHttp2Stream}

const onClientStreamCreatedChannel = dc.channel('http2.client.stream.created');

const onClientStreamStartChannel = dc.channel('http2.client.stream.start');

const onClientStreamErrorChannel = dc.channel('http2.client.stream.error');

const onClientStreamBodyChunkSentChannel = dc.channel('http2.client.stream.bodyChunkSent');

const onClientStreamBodySentChannel = dc.channel('http2.client.stream.bodySent');

const onClientStreamFinishChannel = dc.channel('http2.client.stream.finish');

const onClientStreamCloseChannel = dc.channel('http2.client.stream.close');

const onServerStreamCreatedChannel = dc.channel('http2.server.stream.created');

req = writeGeneric(this, data, encoding, writeCallback);

trackWriteState(this, req.bytes);

if (this.session[kType] === NGHTTP2_SESSION_CLIENT && onClientStreamBodyChunkSentChannel.hasSubscribers) {

onClientStreamBodyChunkSentChannel.publish({

stream: this,

writev,

data,

encoding,

});

}

}

_write(data, encoding, cb) {

}

debugStreamObj(this, 'shutting down writable on _final');

ReflectApply(shutdownWritable, this, [cb]);

if (this.session[kType] === NGHTTP2_SESSION_CLIENT && onClientStreamBodySentChannel.hasSubscribers) {

onClientStreamBodySentChannel.publish({ stream: this });

}

}

_read(nread) {

'use strict';

const common = require('../common');

if (!common.hasCrypto)

common.skip('missing crypto');

// This test ensures that the built-in HTTP/2 diagnostics channels are reporting

// the diagnostics messages for the 'http2.client.stream.bodyChunkSent' and

// 'http2.client.stream.bodySent' channels when ClientHttp2Streams bodies are

// being sent with multiple Buffers and strings.

const assert = require('assert');

const dc = require('diagnostics_channel');

const http2 = require('http2');

const { Duplex } = require('stream');

let bodyChunkSent = false;

dc.subscribe('http2.client.stream.bodyChunkSent', common.mustCall(({ stream, writev, data, encoding }) => {

// Since ClientHttp2Stream is not exported from any module, this just checks

// if the stream is an instance of Duplex.

assert.ok(stream instanceof Duplex);

assert.strictEqual(stream.constructor.name, 'ClientHttp2Stream');

assert.strictEqual(writev, true);

assert.ok(Array.isArray(data));

assert.strictEqual(data.length, 3);

assert.strictEqual(data[0].chunk, 'héllo');

assert.strictEqual(data[0].encoding, 'latin1');

assert.ok(Buffer.from('foo').equals(data[1].chunk));

assert.strictEqual(data[1].encoding, 'buffer');

assert.ok(Buffer.from('bar').equals(data[2].chunk));

assert.strictEqual(data[2].encoding, 'buffer');

assert.strictEqual(encoding, '');

bodyChunkSent = true;

}));

dc.subscribe('http2.client.stream.bodySent', common.mustCall(({ stream }) => {

// 'http2.client.stream.bodyChunkSent' must run first.

assert.ok(bodyChunkSent);

// Since ClientHttp2Stream is not exported from any module, this just checks

// if the stream is an instance of Duplex.

assert.ok(stream instanceof Duplex);

assert.strictEqual(stream.constructor.name, 'ClientHttp2Stream');

}));

const server = http2.createServer();

server.on('stream', common.mustCall((stream) => {

stream.respond({}, { endStream: true });

}));

server.listen(0, common.mustCall(() => {

const port = server.address().port;

const client = http2.connect(`http://localhost:${port}`);

const stream = client.request({ [http2.constants.HTTP2_HEADER_METHOD]: 'POST' });

stream.write('héllo', 'latin1');

stream.write(Buffer.from('foo'));

stream.write(new TextEncoder().encode('bar'));

stream.end();

stream.on('response', common.mustCall(() => {

client.close();

server.close();

}));

}, 1));

'use strict';

const common = require('../common');

if (!common.hasCrypto)

common.skip('missing crypto');

// This test ensures that the built-in HTTP/2 diagnostics channels are reporting

// the diagnostics messages for the 'http2.client.stream.bodyChunkSent' and

// 'http2.client.stream.bodySent' channels when ClientHttp2Streams bodies are

// being sent with multiple Buffers.

const assert = require('assert');

const dc = require('diagnostics_channel');

const http2 = require('http2');

const { Duplex } = require('stream');

let bodyChunkSent = false;

dc.subscribe('http2.client.stream.bodyChunkSent', common.mustCall(({ stream, writev, data, encoding }) => {

// Since ClientHttp2Stream is not exported from any module, this just checks

// if the stream is an instance of Duplex.

assert.ok(stream instanceof Duplex);

assert.strictEqual(stream.constructor.name, 'ClientHttp2Stream');

assert.strictEqual(writev, true);

assert.ok(Array.isArray(data));

assert.strictEqual(data.length, 2);

assert.ok(Buffer.from('foo').equals(data[0]));

assert.ok(Buffer.from('bar').equals(data[1]));

assert.strictEqual(encoding, '');

bodyChunkSent = true;

}));

dc.subscribe('http2.client.stream.bodySent', common.mustCall(({ stream }) => {

// 'http2.client.stream.bodyChunkSent' must run first.

assert.ok(bodyChunkSent);

// Since ClientHttp2Stream is not exported from any module, this just checks

// if the stream is an instance of Duplex.

assert.ok(stream instanceof Duplex);

assert.strictEqual(stream.constructor.name, 'ClientHttp2Stream');

}));

const server = http2.createServer();

server.on('stream', common.mustCall((stream) => {

stream.respond({}, { endStream: true });

}));

server.listen(0, common.mustCall(() => {

const port = server.address().port;

const client = http2.connect(`http://localhost:${port}`);

const stream = client.request({ [http2.constants.HTTP2_HEADER_METHOD]: 'POST' });

stream.write(Buffer.from('foo'));

stream.write(Buffer.from('bar'));

stream.end();

stream.on('response', common.mustCall(() => {

client.close();

server.close();

}));

}, 1));

'use strict';

const common = require('../common');

if (!common.hasCrypto)

common.skip('missing crypto');

// This test ensures that the built-in HTTP/2 diagnostics channels are reporting

// the diagnostics messages for the 'http2.client.stream.bodyChunkSent' and

// 'http2.client.stream.bodySent' channels when ClientHttp2Streams bodies are

// being sent with no chunks.

const assert = require('assert');

const dc = require('diagnostics_channel');

const http2 = require('http2');

const { Duplex } = require('stream');

dc.subscribe('http2.client.stream.bodyChunkSent', common.mustNotCall());

dc.subscribe('http2.client.stream.bodySent', common.mustCall(({ stream }) => {

// Since ClientHttp2Stream is not exported from any module, this just checks

// if the stream is an instance of Duplex.

assert.ok(stream instanceof Duplex);

assert.strictEqual(stream.constructor.name, 'ClientHttp2Stream');

}));

const server = http2.createServer();

server.on('stream', common.mustCall((stream) => {

stream.respond({}, { endStream: true });

}));

server.listen(0, common.mustCall(() => {

const port = server.address().port;

const client = http2.connect(`http://localhost:${port}`);

const stream = client.request({ [http2.constants.HTTP2_HEADER_METHOD]: 'POST' });

stream.end();

stream.on('response', common.mustCall(() => {

client.close();

server.close();

}));

}, 1));

'use strict';

const common = require('../common');

if (!common.hasCrypto)

common.skip('missing crypto');

// This test ensures that the built-in HTTP/2 diagnostics channels are reporting

// the diagnostics messages for the 'http2.client.stream.bodyChunkSent' and

// 'http2.client.stream.bodySent' channels when ClientHttp2Streams bodies are

// being sent with a single Buffer.

const assert = require('assert');

const dc = require('diagnostics_channel');

const http2 = require('http2');

const { Duplex } = require('stream');

let bodyChunkSent = false;

dc.subscribe('http2.client.stream.bodyChunkSent', common.mustCall(({ stream, writev, data, encoding }) => {

// Since ClientHttp2Stream is not exported from any module, this just checks

// if the stream is an instance of Duplex.

assert.ok(stream instanceof Duplex);

assert.strictEqual(stream.constructor.name, 'ClientHttp2Stream');

assert.strictEqual(writev, false);

assert.ok(Buffer.from('foo').equals(data));

assert.strictEqual(encoding, 'buffer');

bodyChunkSent = true;

}));

dc.subscribe('http2.client.stream.bodySent', common.mustCall(({ stream }) => {

// 'http2.client.stream.bodyChunkSent' must run first.

assert.ok(bodyChunkSent);

// Since ClientHttp2Stream is not exported from any module, this just checks

// if the stream is an instance of Duplex.

assert.ok(stream instanceof Duplex);

assert.strictEqual(stream.constructor.name, 'ClientHttp2Stream');

}));

const server = http2.createServer();

server.on('stream', common.mustCall((stream) => {

stream.respond({}, { endStream: true });

}));

server.listen(0, common.mustCall(() => {

const port = server.address().port;

const client = http2.connect(`http://localhost:${port}`);

const stream = client.request({ [http2.constants.HTTP2_HEADER_METHOD]: 'POST' });

stream.write(Buffer.from('foo'));

stream.end();

stream.on('response', common.mustCall(() => {

client.close();

server.close();

}));

}, 1));

'use strict';

const common = require('../common');

if (!common.hasCrypto)

common.skip('missing crypto');

// This test ensures that the built-in HTTP/2 diagnostics channels are reporting

// the diagnostics messages for the 'http2.client.stream.bodyChunkSent' and

// 'http2.client.stream.bodySent' channels when ClientHttp2Streams bodies are

// being sent with a single string.

const assert = require('assert');

const dc = require('diagnostics_channel');

const http2 = require('http2');

const { Duplex } = require('stream');

let bodyChunkSent = false;

dc.subscribe('http2.client.stream.bodyChunkSent', common.mustCall(({ stream, writev, data, encoding }) => {

// Since ClientHttp2Stream is not exported from any module, this just checks

// if the stream is an instance of Duplex.

assert.ok(stream instanceof Duplex);

assert.strictEqual(stream.constructor.name, 'ClientHttp2Stream');

assert.strictEqual(writev, false);

assert.strictEqual(data, 'foo');

assert.strictEqual(encoding, 'utf8');

bodyChunkSent = true;

}));

dc.subscribe('http2.client.stream.bodySent', common.mustCall(({ stream }) => {

// 'http2.client.stream.bodyChunkSent' must run first.

assert.ok(bodyChunkSent);

// Since ClientHttp2Stream is not exported from any module, this just checks

// if the stream is an instance of Duplex.

assert.ok(stream instanceof Duplex);

assert.strictEqual(stream.constructor.name, 'ClientHttp2Stream');

}));

const server = http2.createServer();

server.on('stream', common.mustCall((stream) => {

stream.respond({}, { endStream: true });

}));

server.listen(0, common.mustCall(() => {

const port = server.address().port;

const client = http2.connect(`http://localhost:${port}`);

const stream = client.request({ [http2.constants.HTTP2_HEADER_METHOD]: 'POST' });

stream.write('foo');

stream.end();

stream.on('response', common.mustCall(() => {

client.close();

server.close();

}));

}, 1));