stream: add diagnostics for readable stream chunk additions

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

Author: RaisinTen

doc/api/diagnostics_channel.md

@@ -1400,6 +1400,16 @@ added: v16.18.0
 
 Emitted when a new process 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.
+
 ##### Event: `'execve'`
 
 * `execPath` {string}

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.

test/parallel/test-diagnostics-channel-stream-readable-add-chunk.js

@@ -0,0 +1,57 @@
+'use strict';
+
+const common = require('../common');
+if (!common.hasCrypto)
+  common.skip('missing crypto');
+
+// This test ensures that the built-in Stream diagnostics channels are reporting
+// the diagnostics messages for the 'stream.readable.addChunk' channel when
+// chunks are added to the readable stream by:
+// - readable.push()
+// - readable.unshift()
+
+const assert = require('assert');
+const dc = require('diagnostics_channel');
+const { Readable } = require('stream');
+
+const values = [
+  'hello',
+  'world',
+  '🌍',
+];
+
+const objects = [
+  { msg: values[0] },
+  { msg: values[1] },
+  { msg: values[2] },
+];
+
+let i = 0;
+dc.subscribe('stream.readable.addChunk', common.mustCall(({ stream, chunk }) => {
+  if (i < values.length) {
+    assert.strictEqual(stream, readable);
+    assert.strictEqual(chunk.toString(), values[i]);
+  } else {
+    assert.strictEqual(stream, objReadable);
+    assert.strictEqual(chunk, objects[i - 3]);
+  }
+  ++i;
+}, values.length * 2));
+
+const readable = new Readable({
+  read() {}
+});
+
+readable.push(values[0]);
+readable.push(values[1]);
+readable.unshift(values[2]);
+readable.push(null);
+
+const objReadable = new Readable({
+  objectMode: true,
+  read() {}
+});
+objReadable.push(objects[0]);
+objReadable.push(objects[1]);
+objReadable.unshift(objects[2]);
+objReadable.push(null);