Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

stream: include strategy options on Readable #43515

Merged
merged 10 commits into from
Jul 13, 2022
6 changes: 5 additions & 1 deletion doc/api/stream.md
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -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])`
Expand Down
6 changes: 4 additions & 2 deletions lib/internal/streams/readable.js
Original file line number Diff line number Diff line change
Expand Up @@ -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) {
Expand Down
34 changes: 25 additions & 9 deletions lib/internal/webstreams/adapters.js
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand All @@ -382,14 +386,26 @@ function newReadableStreamFromStreamReadable(streamReadable) {

const objectMode = streamReadable.readableObjectMode;
const highWaterMark = streamReadable.readableHighWaterMark;
// When not running in objectMode explicitly, we just fall
// back to a minimal strategy that just specifies the highWaterMark
// and no size algorithm. Using a ByteLengthQueuingStrategy here
// is unnecessary.
const strategy =
objectMode ?
new CountQueuingStrategy({ highWaterMark }) :
{ highWaterMark };

const evaluateStrategyOrFallback = (strategy) => {
Warkanlock marked this conversation as resolved.
Show resolved Hide resolved
// If there is a strategy available, use it
if (strategy)
return strategy;

if (objectMode) {
// When running in objectMode explicitly but no strategy, we just fall
// back to CountQueuingStrategy
return new CountQueuingStrategy({ highWaterMark });
}

// When not running in objectMode explicitly, we just fall
// back to a minimal strategy that just specifies the highWaterMark
// and no size algorithm. Using a ByteLengthQueuingStrategy here
// is unnecessary.
return { highWaterMark };
};

const strategy = evaluateStrategyOrFallback(options?.strategy);

let controller;

Expand Down
75 changes: 75 additions & 0 deletions test/parallel/test-stream-readable-strategy-option.js
Original file line number Diff line number Diff line change
@@ -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);
});
}