stream: add diagnostics_channel event for completion

Signed-off-by: Bryan English <[email protected]>

Author: bengl

lib/internal/webstreams/readablestream.js

@@ -136,6 +136,14 @@ const { Buffer } = require('buffer');
 
 const assert = require('internal/assert');
 
+let streamDoneChannel;
+function getStreamDoneChannel() {
+  if (streamDoneChannel === undefined) {
+    streamDoneChannel = require('diagnostics_channel').channel('stream.web.done');
+  }
+  return streamDoneChannel;
+}
+
 const kCancel = Symbol('kCancel');
 const kClose = Symbol('kClose');
 const kChunk = Symbol('kChunk');
@@ -463,6 +471,9 @@ class ReadableStream {
     validateObject(options, 'options', kValidateObjectAllowObjectsAndNull);
     const preventCancel = !!(options?.preventCancel);
 
+    const stream = this;
+    const channel = getStreamDoneChannel();
+
     // eslint-disable-next-line no-use-before-define
     const reader = new ReadableStreamDefaultReader(this);
 
@@ -491,7 +502,7 @@ class ReadableStream {
       const promise = PromiseWithResolvers();
 
       // eslint-disable-next-line no-use-before-define
-      readableStreamDefaultReaderRead(reader, new ReadableStreamAsyncIteratorReadRequest(reader, state, promise));
+      readableStreamDefaultReaderRead(reader, new ReadableStreamAsyncIteratorReadRequest(reader, state, promise, stream, channel));
       return promise.promise;
     }
 
@@ -509,10 +520,16 @@ class ReadableStream {
         const result = readableStreamReaderGenericCancel(reader, value);
         readableStreamReaderGenericRelease(reader);
         await result;
+        if (channel.hasSubscribers) {
+          channel.publish({ stream });
+        }
         return { done: true, value }; // eslint-disable-line node-core/avoid-prototype-pollution
       }
 
       readableStreamReaderGenericRelease(reader);
+      if (channel.hasSubscribers) {
+        channel.publish({ stream });
+      }
       return { done: true, value }; // eslint-disable-line node-core/avoid-prototype-pollution
     }
 
@@ -760,10 +777,12 @@ function createReadableStreamBYOBRequest(controller, view) {
 }
 
 class ReadableStreamAsyncIteratorReadRequest {
-  constructor(reader, state, promise) {
+  constructor(reader, state, promise, stream, channel) {
     this.reader = reader;
     this.state = state;
     this.promise = promise;
+    this.stream = stream;
+    this.channel = channel;
   }
 
   [kChunk](chunk) {
@@ -775,6 +794,9 @@ class ReadableStreamAsyncIteratorReadRequest {
     this.state.current = undefined;
     this.state.done = true;
     readableStreamReaderGenericRelease(this.reader);
+    if (this.channel.hasSubscribers) {
+      this.channel.publish({ stream: this.stream });
+    }
     this.promise.resolve({ done: true, value: undefined });
   }
 
@@ -888,6 +910,13 @@ class ReadableStreamDefaultReader {
     // Slow path: create request and go through normal flow
     const readRequest = new DefaultReadRequest();
     readableStreamDefaultReaderRead(this, readRequest);
+    if (getStreamDoneChannel().hasSubscribers) {
+      const stream = this[kState].stream;
+      PromisePrototypeThen(readRequest.promise, ({ done }) => {
+        if (done)
+          getStreamDoneChannel().publish({ stream });
+      });
+    }
     return readRequest.promise;
   }
 
@@ -1028,6 +1057,13 @@ class ReadableStreamBYOBReader {
     }
     const readIntoRequest = new ReadIntoRequest();
     readableStreamBYOBReaderRead(this, view, min, readIntoRequest);
+    if (getStreamDoneChannel().hasSubscribers) {
+      const stream = this[kState].stream;
+      PromisePrototypeThen(readIntoRequest.promise, ({ done }) => {
+        if (done)
+          getStreamDoneChannel().publish({ stream });
+      });
+    }
     return readIntoRequest.promise;
   }
 

test/parallel/test-whatwg-webstreams-dc-events.mjs

@@ -0,0 +1,44 @@
+// Flags: --expose-internals
+import * as common from '../common/index.mjs';
+import assert from 'assert';
+
+import util from 'internal/webstreams/util';
+
+import { Readable } from 'stream';
+
+import * as dc from 'diagnostics_channel';
+
+{
+  const readable = Readable.toWeb(Readable.from([1]));
+
+  const channel = dc.channel('stream.web.done');
+  const subscriber = common.mustCall(({ stream }) => {
+    assert.strictEqual(readable, stream);
+    assert.strictEqual(readable[util.kState].state, 'closed');
+  });
+  channel.subscribe(subscriber);
+
+  const reader = readable.getReader();
+  let result;
+
+  while (!result?.done) {
+    result = await reader.read();
+  }
+
+  channel.unsubscribe(subscriber);
+}
+
+{
+  const readable = Readable.toWeb(Readable.from([1]));
+
+  const channel = dc.channel('stream.web.done');
+  const subscriber = common.mustCall(({ stream }) => {
+    assert.strictEqual(readable, stream);
+    assert.strictEqual(readable[util.kState].state, 'closed');
+  });
+  channel.subscribe(subscriber);
+
+  for await (const _ of readable) {}
+
+  channel.unsubscribe(subscriber);
+}