http2: add diagnostics channels for client stream request body

These would allow inspection of HTTP/2 client stream request bodies.

Signed-off-by: Darshan Sen <[email protected]>

Author: RaisinTen

doc/api/diagnostics_channel.md

@@ -1243,6 +1243,19 @@ Emitted when an error occurs during the processing of a stream on the client.
 
 Emitted when a stream is received on the client.
 
+##### Event: `'http2.client.stream.bodyChunkSent'`
+
+* `stream` {ClientHttp2Stream}
+* `chunk` {Buffer}
+
+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}

lib/internal/http2/core.js

@@ -186,10 +186,14 @@ const { UV_EOF } = internalBinding('uv');
 const { StreamPipe } = internalBinding('stream_pipe');
 const { _connectionListener: httpConnectionListener } = http;
 
+const { Buffer } = require('buffer');
+
 const dc = require('diagnostics_channel');
 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');
@@ -2300,6 +2304,34 @@ class Http2Stream extends Duplex {
       req = writeGeneric(this, data, encoding, writeCallback);
 
     trackWriteState(this, req.bytes);
+
+    if (this.session[kType] === NGHTTP2_SESSION_CLIENT && onClientStreamBodyChunkSentChannel.hasSubscribers) {
+      let chunk;
+
+      if (ArrayIsArray(data)) {
+        const buffers = [];
+        for (let i = 0; i < data.length; ++i) {
+          if (typeof data[i] === 'object') {
+            if (typeof data[i].chunk === 'string') {
+              buffers.push(Buffer.from(data[i].chunk, data[i].encoding));
+            } else {
+              buffers.push(data[i].chunk);
+            }
+          }
+        }
+
+        chunk = Buffer.concat(buffers);
+      } else if (typeof data === 'string') {
+        chunk = Buffer.from(data);
+      } else {
+        chunk = data;
+      }
+
+      onClientStreamBodyChunkSentChannel.publish({
+        stream: this,
+        chunk,
+      });
+    }
   }
 
   _write(data, encoding, cb) {
@@ -2317,6 +2349,10 @@ class Http2Stream extends Duplex {
     }
     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) {

test/parallel/test-diagnostics-channel-http2-client-stream-body.js

@@ -0,0 +1,60 @@
+'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.
+
+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, chunk }) => {
+  // 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(Buffer.isBuffer(chunk), true);
+  assert.strictEqual(chunk.toString(), 'foobarbaz');
+
+  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.write(Buffer.from('bar'));
+  stream.write(new TextEncoder().encode('baz'));
+  stream.end();
+
+  stream.on('response', common.mustCall(() => {
+    client.close();
+    server.close();
+  }));
+}, 1));