lib: add support for readable byte streams to .toWeb()

Add support for the creation of ReadableByteStream to Readable.toWeb()
and Duplex.toWeb()
This enables the use of .getReader({ mode: "byob" }) on
e.g. socket().toWeb()

Refs: #56004 (comment)
Refs: https://developer.mozilla.org/en-US/docs/Web/API/Streams_API/Using_readable_byte_streams
PR-URL: #58664
Reviewed-By: Matteo Collina <matteo.collina@gmail.com>
Reviewed-By: Ethan Arrowood <ethan@arrowood.dev>
Reviewed-By: Mattias Buelens <mattias@buelens.com>
Reviewed-By: James M Snell <jasnell@gmail.com>

Author: seriousme

doc/api/stream.md

@@ -3174,6 +3174,9 @@ Returns whether the stream has been read from or cancelled.
 <!-- YAML
 added: v17.0.0
 changes:
+  - version: REPLACEME
+    pr-url: https://github.com/nodejs/node/pull/58664
+    description: Add 'type' option to specify 'bytes'.
   - version: v24.0.0
     pr-url: https://github.com/nodejs/node/pull/57513
     description: Marking the API stable.
@@ -3194,6 +3197,7 @@ changes:
       If no value is provided, the size will be `1` for all the chunks.
       * `chunk` {any}
       * Returns: {number}
+  * `type` {string} Must be 'bytes' or undefined.
 * Returns: {ReadableStream}
 
 ### `stream.Writable.fromWeb(writableStream[, options])`
@@ -3361,17 +3365,22 @@ duplex.write('hello');
 duplex.once('readable', () => console.log('readable', duplex.read()));
 ```
 
-### `stream.Duplex.toWeb(streamDuplex)`
+### `stream.Duplex.toWeb(streamDuplex[, options])`
 
 <!-- YAML
 added: v17.0.0
 changes:
+  - version: REPLACEME
+    pr-url: https://github.com/nodejs/node/pull/58664
+    description: Add 'type' option to specify 'bytes'.
   - version: v24.0.0
     pr-url: https://github.com/nodejs/node/pull/57513
     description: Marking the API stable.
 -->
 
 * `streamDuplex` {stream.Duplex}
+* `options` {Object}
+  * `type` {string} Must be 'bytes' or undefined.
 * Returns: {Object}
   * `readable` {ReadableStream}
   * `writable` {WritableStream}

doc/api/webstreams.md

@@ -1773,7 +1773,7 @@ text(readable).then((data) => {
 [Streams]: stream.md
 [WHATWG Streams Standard]: https://streams.spec.whatwg.org/
 [`stream.Duplex.fromWeb`]: stream.md#streamduplexfromwebpair-options
-[`stream.Duplex.toWeb`]: stream.md#streamduplextowebstreamduplex
+[`stream.Duplex.toWeb`]: stream.md#streamduplextowebstreamduplex-options
 [`stream.Duplex`]: stream.md#class-streamduplex
 [`stream.Readable.fromWeb`]: stream.md#streamreadablefromwebreadablestream-options
 [`stream.Readable.toWeb`]: stream.md#streamreadabletowebstreamreadable-options

lib/internal/streams/duplex.js

@@ -191,8 +191,8 @@ Duplex.fromWeb = function(pair, options) {
     options);
 };
 
-Duplex.toWeb = function(duplex) {
-  return lazyWebStreams().newReadableWritablePairFromDuplex(duplex);
+Duplex.toWeb = function(duplex, options) {
+  return lazyWebStreams().newReadableWritablePairFromDuplex(duplex, options);
 };
 
 let duplexify;

lib/internal/webstreams/adapters.js

@@ -73,6 +73,7 @@ const {
   validateBoolean,
   validateFunction,
   validateObject,
+  validateOneOf,
 } = require('internal/validators');
 
 const {
@@ -417,7 +418,8 @@ function newStreamWritableFromWritableStream(writableStream, options = kEmptyObj
  * @typedef {import('./queuingstrategies').QueuingStrategy} QueuingStrategy
  * @param {Readable} streamReadable
  * @param {{
- *  strategy : QueuingStrategy
+ *  strategy? : QueuingStrategy
+ *  type? : 'bytes',
  * }} [options]
  * @returns {ReadableStream}
  */
@@ -432,6 +434,12 @@ function newReadableStreamFromStreamReadable(streamReadable, options = kEmptyObj
       'stream.Readable',
       streamReadable);
   }
+  validateObject(options, 'options');
+  if (options.type !== undefined) {
+    validateOneOf(options.type, 'options.type', ['bytes', undefined]);
+  }
+
+  const isBYOB = options.type === 'bytes';
 
   if (isDestroyed(streamReadable) || !isReadable(streamReadable)) {
     const readable = new ReadableStream();
@@ -443,6 +451,9 @@ function newReadableStreamFromStreamReadable(streamReadable, options = kEmptyObj
   const highWaterMark = streamReadable.readableHighWaterMark;
 
   const evaluateStrategyOrFallback = (strategy) => {
+    // If the stream is BYOB, we only use highWaterMark
+    if (isBYOB)
+      return { highWaterMark };
     // If there is a strategy available, use it
     if (strategy)
       return strategy;
@@ -491,7 +502,19 @@ function newReadableStreamFromStreamReadable(streamReadable, options = kEmptyObj
   streamReadable.on('data', onData);
 
   return new ReadableStream({
-    start(c) { controller = c; },
+    type: isBYOB ? 'bytes' : undefined,
+    start(c) {
+      controller = c;
+      if (isBYOB) {
+        streamReadable.once('end', () => {
+          // close the controller
+          controller.close();
+          // And unlock the last BYOB read request
+          controller.byobRequest?.respond(0);
+          wasCanceled = true;
+        });
+      }
+    },
 
     pull() { streamReadable.resume(); },
 
@@ -601,9 +624,10 @@ function newStreamReadableFromReadableStream(readableStream, options = kEmptyObj
 
 /**
  * @param {Duplex} duplex
+ * @param {{ type?: 'bytes' }} [options]
  * @returns {ReadableWritablePair}
  */
-function newReadableWritablePairFromDuplex(duplex) {
+function newReadableWritablePairFromDuplex(duplex, options = kEmptyObject) {
   // Not using the internal/streams/utils isWritableNodeStream and
   // isReadableNodeStream utilities here because they will return false
   // if the duplex was created with writable or readable options set to
@@ -615,9 +639,11 @@ function newReadableWritablePairFromDuplex(duplex) {
     throw new ERR_INVALID_ARG_TYPE('duplex', 'stream.Duplex', duplex);
   }
 
+  validateObject(options, 'options');
+
   if (isDestroyed(duplex)) {
     const writable = new WritableStream();
-    const readable = new ReadableStream();
+    const readable = new ReadableStream({ type: options.type });
     writable.close();
     readable.cancel();
     return { readable, writable };
@@ -633,8 +659,8 @@ function newReadableWritablePairFromDuplex(duplex) {
 
   const readable =
     isReadable(duplex) ?
-      newReadableStreamFromStreamReadable(duplex) :
-      new ReadableStream();
+      newReadableStreamFromStreamReadable(duplex, options) :
+      new ReadableStream({ type: options.type });
 
   if (!isReadable(duplex))
     readable.cancel();

test/parallel/test-stream-duplex.js

@@ -131,3 +131,26 @@ process.on('exit', () => {
     assert.deepStrictEqual(Buffer.from(result.value), dataToRead);
   }));
 }
+
+// Duplex.toWeb BYOB
+{
+  const dataToRead = Buffer.from('hello');
+  const dataToWrite = Buffer.from('world');
+
+  const duplex = Duplex({
+    read() {
+      this.push(dataToRead);
+      this.push(null);
+    },
+    write: common.mustCall((chunk) => {
+      assert.strictEqual(chunk, dataToWrite);
+    })
+  });
+
+  const { writable, readable } = Duplex.toWeb(duplex, { type: 'bytes' });
+  writable.getWriter().write(dataToWrite);
+  const data = new Uint8Array(dataToRead.length);
+  readable.getReader({ mode: 'byob' }).read(data).then(common.mustCall((result) => {
+    assert.deepStrictEqual(Buffer.from(result.value), dataToRead);
+  }));
+}

test/parallel/test-stream-readable-to-web-byob.js

@@ -0,0 +1,49 @@
+'use strict';
+require('../common');
+const { Readable } = require('stream');
+const assert = require('assert');
+const common = require('../common');
+
+let count = 0;
+
+const nodeStream = new Readable({
+  read(size) {
+    if (this.destroyed) {
+      return;
+    }
+    // Simulate a stream that pushes sequences of 16 bytes
+    const buffer = Buffer.alloc(size);
+    for (let i = 0; i < size; i++) {
+      buffer[i] = count++ % 16;
+    }
+    this.push(buffer);
+  }
+});
+
+// Test validation of 'type' option
+assert.throws(
+  () => {
+    Readable.toWeb(nodeStream, { type: 'wrong type' });
+  },
+  {
+    code: 'ERR_INVALID_ARG_VALUE'
+  }
+);
+
+// Test normal operation with ReadableByteStream
+const webStream = Readable.toWeb(nodeStream, { type: 'bytes' });
+const reader = webStream.getReader({ mode: 'byob' });
+const expected = new Uint8Array(16);
+for (let i = 0; i < 16; i++) {
+  expected[i] = count++;
+}
+
+for (let i = 0; i < 1000; i++) {
+  // Read 16 bytes of data from the stream
+  const receive = new Uint8Array(16);
+  reader.read(receive).then(common.mustCall((result) => {
+    // Verify the data received
+    assert.ok(!result.done);
+    assert.deepStrictEqual(result.value, expected);
+  }));
+}

test/parallel/test-stream-readable-to-web-termination-byob.js

@@ -0,0 +1,15 @@
+'use strict';
+require('../common');
+const { Readable } = require('stream');
+const assert = require('assert');
+const common = require('../common');
+{
+  const r = Readable.from([]);
+  // Cancelling reader while closing should not cause uncaught exceptions
+  r.on('close', common.mustCall(() => reader.cancel()));
+
+  const reader = Readable.toWeb(r, { type: 'bytes' }).getReader({ mode: 'byob' });
+  reader.read(new Uint8Array(16)).then(common.mustCall((result) => {
+    assert.ok(result.done);
+  }));
+}