stream: add diagnostics channel for tracking chunks added to a readable stream

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

Author: RaisinTen

doc/api/diagnostics_channel.md

@@ -1399,6 +1399,17 @@ Emitted when [`net.Server.listen()`][] is returning an error.
 
 Emitted when a new UDP socket is created.
 
+#### Stream
+
+> Stability: 1 - Experimental
+
+##### Event: `'stream.readable.addChunk'`
+
+* `stream` {stream.Readable}
+* `chunk` {Buffer}
+
+Emitted when a chunk is added to a readable stream.
+
 #### Process
 
 > Stability: 1 - Experimental

lib/internal/inspector/network_http2.js

@@ -180,22 +180,26 @@ function onClientStreamFinish({ stream, headers }) {
       charset,
     },
   });
+}
+
+/**
+ * When a chunk of the response body has been received, cache it until `getResponseBody` request
+ * https://chromedevtools.github.io/devtools-protocol/1-3/Network/#method-getResponseBody or
+ * stream it with `streamResourceContent` request.
+ * https://chromedevtools.github.io/devtools-protocol/tot/Network/#method-streamResourceContent
+ * @param {{ stream: import('http2').ClientHttp2Stream, chunk: Buffer }} event
+ */
+function onStreamReadableAddChunk({ stream, chunk }) {
+  if (typeof stream[kInspectorRequestId] !== 'string') {
+    return;
+  }
 
-  stream.on('data', (chunk) => {
-    /**
-     * When a chunk of the response body has been received, cache it until `getResponseBody` request
-     * https://chromedevtools.github.io/devtools-protocol/1-3/Network/#method-getResponseBody or
-     * stream it with `streamResourceContent` request.
-     * https://chromedevtools.github.io/devtools-protocol/tot/Network/#method-streamResourceContent
-     */
-
-    Network.dataReceived({
-      requestId: stream[kInspectorRequestId],
-      timestamp: getMonotonicTime(),
-      dataLength: chunk.byteLength,
-      encodedDataLength: chunk.byteLength,
-      data: chunk,
-    });
+  Network.dataReceived({
+    requestId: stream[kInspectorRequestId],
+    timestamp: getMonotonicTime(),
+    dataLength: chunk.byteLength,
+    encodedDataLength: chunk.byteLength,
+    data: chunk,
   });
 }
 
@@ -228,4 +232,5 @@ module.exports = registerDiagnosticChannels([
   ['http2.client.stream.close', onClientStreamClose],
   ['http2.client.stream.bodyChunkSent', onClientStreamBodyChunkSent],
   ['http2.client.stream.bodySent', onClientStreamBodySent],
+  ['stream.readable.addChunk', onStreamReadableAddChunk],
 ]);

lib/internal/streams/readable.js

@@ -92,6 +92,9 @@ const FastBuffer = Buffer[SymbolSpecies];
 const { StringDecoder } = require('string_decoder');
 const from = require('internal/streams/from');
 
+const dc = require('diagnostics_channel');
+const onAddChunkChannel = dc.channel('stream.readable.addChunk');
+
 ObjectSetPrototypeOf(Readable.prototype, Stream.prototype);
 ObjectSetPrototypeOf(Readable, Stream);
 const nop = () => {};
@@ -546,6 +549,13 @@ function canPushMore(state) {
 }
 
 function addChunk(stream, state, chunk, addToFront) {
+  if (onAddChunkChannel.hasSubscribers) {
+    onAddChunkChannel.publish({
+      stream,
+      chunk,
+    });
+  }
+
   if ((state[kState] & (kFlowing | kSync | kDataListening)) === (kFlowing | kDataListening) && state.length === 0) {
     // Use the guard to avoid creating `Set()` repeatedly
     // when we have multiple pipes.