stream: add diagnostics_channel event for completion

Author: bengl

lib/internal/webstreams/readablestream.js

@@ -129,6 +129,12 @@ const {
 
 const assert = require('internal/assert');
 
+const dc = require('diagnostics_channel');
+let streamDoneChannel;
+function getStreamDoneChannel () {
+  return streamDoneChannel ||= dc.channel('stream.web.done');
+}
+
 const kCancel = Symbol('kCancel');
 const kClose = Symbol('kClose');
 const kChunk = Symbol('kChunk');
@@ -424,6 +430,9 @@ class ReadableStream {
       preventCancel = false,
     } = options;
 
+    const stream = this;
+    const channel = getStreamDoneChannel();
+
     const reader = new ReadableStreamDefaultReader(this);
     let done = false;
     let started = false;
@@ -455,6 +464,9 @@ class ReadableStream {
           current = undefined;
           done = true;
           readableStreamReaderGenericRelease(reader);
+          if (channel.hasSubscribers) {
+            channel.publish({ stream });
+          }
           promise.resolve({ done: true, value: undefined });
         },
         [kError](error) {
@@ -468,23 +480,25 @@ class ReadableStream {
     }
 
     async function returnSteps(value) {
-      if (done)
-        return { done: true, value };
-      done = true;
+      if (!done) {
+        done = true;
 
-      if (reader[kState].stream === undefined) {
-        throw new ERR_INVALID_STATE.TypeError(
-          'The reader is not bound to a ReadableStream');
+        if (reader[kState].stream === undefined) {
+          throw new ERR_INVALID_STATE.TypeError(
+            'The reader is not bound to a ReadableStream');
+        }
+        assert(!reader[kState].readRequests.length);
+        if (!preventCancel) {
+          const result = readableStreamReaderGenericCancel(reader, value);
+          readableStreamReaderGenericRelease(reader);
+          await result;
+        } else {
+          readableStreamReaderGenericRelease(reader);
+        }
       }
-      assert(!reader[kState].readRequests.length);
-      if (!preventCancel) {
-        const result = readableStreamReaderGenericCancel(reader, value);
-        readableStreamReaderGenericRelease(reader);
-        await result;
-        return { done: true, value };
+      if (channel.hasSubscribers) {
+        channel.publish({ stream });
       }
-
-      readableStreamReaderGenericRelease(reader);
       return { done: true, value };
     }
 
@@ -789,6 +803,12 @@ class ReadableStreamDefaultReader {
     }
     const readRequest = new DefaultReadRequest();
     readableStreamDefaultReaderRead(this, readRequest);
+    if (getStreamDoneChannel().hasSubscribers) {
+      readRequest.promise.then(({ done }) => {
+        if (done)
+          getStreamDoneChannel().publish({ stream: this[kState].stream });
+      })
+    }
     return readRequest.promise;
   }
 
@@ -906,6 +926,12 @@ class ReadableStreamBYOBReader {
     }
     const readIntoRequest = new ReadIntoRequest();
     readableStreamBYOBReaderRead(this, view, readIntoRequest);
+    if (getStreamDoneChannel().hasSubscribers) {
+      readIntoRequest.promise.then(({ done }) => {
+        if (done)
+          getStreamDoneChannel().publish({ stream: this[kState].stream });
+      })
+    }
     return readIntoRequest.promise;
   }
 

test/parallel/test-bootstrap-modules.js

@@ -44,6 +44,7 @@ const expectedModules = new Set([
   'Internal Binding v8',
   'Internal Binding worker',
   'NativeModule buffer',
+  'NativeModule diagnostics_channel',
   'NativeModule events',
   'NativeModule fs',
   'NativeModule internal/abort_controller',

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);
+}