stream: include strategy options on Readable

doc/api/stream.md

@@ -2807,7 +2807,7 @@ added:
 
 Returns whether the stream is readable.
 
-### `stream.Readable.toWeb(streamReadable)`
+### `stream.Readable.toWeb(streamReadable[, options])`
 
 <!-- YAML
 added: v17.0.0
@@ -2816,6 +2816,10 @@ added: v17.0.0
 > Stability: 1 - Experimental
 
 * `streamReadable` {stream.Readable}
+* `options` {Object}
+  * `strategy` {Object}
+    * `highWaterMark` {number}
+    * `size` {Function}
 * Returns: {ReadableStream}
 
 ### `stream.Writable.fromWeb(writableStream[, options])`

lib/internal/streams/readable.js

@@ -1405,8 +1405,10 @@ Readable.fromWeb = function(readableStream, options) {
     options);
 };
 
-Readable.toWeb = function(streamReadable) {
-  return lazyWebStreams().newReadableStreamFromStreamReadable(streamReadable);
+Readable.toWeb = function(streamReadable, options) {
+  return lazyWebStreams().newReadableStreamFromStreamReadable(
+    streamReadable,
+    options);
 };
 
 Readable.wrap = function(src, options) {

lib/internal/webstreams/adapters.js

@@ -359,10 +359,14 @@ function newStreamWritableFromWritableStream(writableStream, options = kEmptyObj
 }
 
 /**
+ * @typedef {import('./queuingstrategies').QueuingStrategy} QueuingStrategy
  * @param {Readable} streamReadable
+ * @param {{
+ *  strategy : QueuingStrategy
+ * }} [options]
  * @returns {ReadableStream}
  */
-function newReadableStreamFromStreamReadable(streamReadable) {
+function newReadableStreamFromStreamReadable(streamReadable, options = kEmptyObject) {
   // Not using the internal/streams/utils isReadableNodeStream utility
   // here because it will return false if streamReadable is a Duplex
   // whose readable option is false. For a Duplex that is not readable,
@@ -386,7 +390,12 @@ function newReadableStreamFromStreamReadable(streamReadable) {
   // back to a minimal strategy that just specifies the highWaterMark
   // and no size algorithm. Using a ByteLengthQueuingStrategy here
   // is unnecessary.
-  const strategy =
+
+  // If there is a strategy available, use it. Otherwise, fallback to default behavior
+  const queuingStrategy = options?.strategy;
+
+  const strategy = queuingStrategy ?
+    queuingStrategy :
     objectMode ?
       new CountQueuingStrategy({ highWaterMark }) :
       { highWaterMark };

test/parallel/test-stream-readable-strategy-option.js

@@ -0,0 +1,75 @@
+'use strict';
+const common = require('../common');
+const { Readable } = require('stream');
+const assert = require('assert');
+const { strictEqual } = require('assert');
+
+{
+  // Strategy 2
+  const streamData = ['a', 'b', 'c', null];
+
+  // Fulfill a Readable object
+  const readable = new Readable({
+    read: common.mustCall(() => {
+      process.nextTick(() => {
+        readable.push(streamData.shift());
+      });
+    }, streamData.length),
+  });
+
+  // Use helper to convert it to a Web ReadableStream using ByteLength strategy
+  const readableStream = Readable.toWeb(readable, {
+    strategy: new ByteLengthQueuingStrategy({ highWaterMark: 1 }),
+  });
+
+  assert(!readableStream.locked);
+  readableStream.getReader().read().then(common.mustCall());
+}
+
+{
+  // Strategy 2
+  const streamData = ['a', 'b', 'c', null];
+
+  // Fulfill a Readable object
+  const readable = new Readable({
+    read: common.mustCall(() => {
+      process.nextTick(() => {
+        readable.push(streamData.shift());
+      });
+    }, streamData.length),
+  });
+
+  // Use helper to convert it to a Web ReadableStream using Count strategy
+  const readableStream = Readable.toWeb(readable, {
+    strategy: new CountQueuingStrategy({ highWaterMark: 1 }),
+  });
+
+  assert(!readableStream.locked);
+  readableStream.getReader().read().then(common.mustCall());
+}
+
+{
+  const desireSizeExpected = 2;
+
+  const stringStream = new ReadableStream(
+    {
+      start(controller) {
+        // Check if the strategy is being assigned on the init of the ReadableStream
+        strictEqual(controller.desiredSize, desireSizeExpected);
+        controller.enqueue('a');
+        controller.enqueue('b');
+        controller.close();
+      },
+    },
+    new CountQueuingStrategy({ highWaterMark: desireSizeExpected })
+  );
+
+  const reader = stringStream.getReader();
+
+  reader.read().then(common.mustCall());
+  reader.read().then(common.mustCall());
+  reader.read().then(({ value, done }) => {
+    strictEqual(value, undefined);
+    strictEqual(done, true);
+  });
+}