Fix browser support (#122)

Author: ehmicky

package.json

@@ -43,6 +43,7 @@
 		"concat"
 	],
 	"dependencies": {
+		"@sec-ant/readable-stream": "^0.3.2",
 		"is-stream": "^4.0.1"
 	},
 	"devDependencies": {

readme.md

@@ -60,6 +60,8 @@ const {body: readableStream} = await fetch('https://example.com');
 console.log(await getStream(readableStream));
 ```
 
+This works in any browser, even [the ones](https://developer.mozilla.org/en-US/docs/Web/API/ReadableStream#browser_compatibility) not supporting `ReadableStream.values()` yet.
+
 ### Async iterables
 
 ```js

source/stream.js

@@ -1,17 +1,25 @@
 import {isReadableStream} from 'is-stream';
+import {ponyfill} from './web-stream.js';
 
 export const getAsyncIterable = stream => {
 	if (isReadableStream(stream, {checkOpen: false})) {
 		return getStreamIterable(stream);
 	}
 
-	if (typeof stream?.[Symbol.asyncIterator] !== 'function') {
-		throw new TypeError('The first argument must be a Readable, a ReadableStream, or an async iterable.');
+	if (typeof stream?.[Symbol.asyncIterator] === 'function') {
+		return stream;
 	}
 
-	return stream;
+	// `ReadableStream[Symbol.asyncIterator]` support is missing in multiple browsers, so we ponyfill it
+	if (toString.call(stream) === '[object ReadableStream]') {
+		return ponyfill.asyncIterator.call(stream);
+	}
+
+	throw new TypeError('The first argument must be a Readable, a ReadableStream, or an async iterable.');
 };
 
+const {toString} = Object.prototype;
+
 // The default iterable for Node.js streams does not allow for multiple readers at once, so we re-implement it
 const getStreamIterable = async function * (stream) {
 	if (nodeImports === undefined) {

source/web-stream.js

@@ -0,0 +1,13 @@
+export const ponyfill = {};
+
+const {prototype} = ReadableStream;
+
+// Use this library as a ponyfill instead of a polyfill.
+// I.e. avoid modifying global variables.
+// We can remove this once https://github.com/Sec-ant/readable-stream/issues/2 is fixed
+if (prototype[Symbol.asyncIterator] === undefined && prototype.values === undefined) {
+	await import('@sec-ant/readable-stream');
+	ponyfill.asyncIterator = prototype[Symbol.asyncIterator];
+	delete prototype[Symbol.asyncIterator];
+	delete prototype.values;
+}

test/helpers/index.js

@@ -1,9 +1,23 @@
 import {Duplex, Readable} from 'node:stream';
+import {finished} from 'node:stream/promises';
 
 export const createStream = streamDef => typeof streamDef === 'function'
 	? Duplex.from(streamDef)
 	: Readable.from(streamDef);
 
+// @todo Use ReadableStream.from() after dropping support for Node 18
+export const readableStreamFrom = chunks => new ReadableStream({
+	start(controller) {
+		for (const chunk of chunks) {
+			controller.enqueue(chunk);
+		}
+
+		controller.close();
+	},
+});
+
 // Tests related to big buffers/strings can be slow. We run them serially and
 // with a higher timeout to ensure they do not randomly fail.
 export const BIG_TEST_DURATION = '2m';
+
+export const onFinishedStream = stream => finished(stream, {cleanup: true});

test/stream.js

@@ -1,14 +1,13 @@
 import {once} from 'node:events';
 import {version} from 'node:process';
 import {Readable, Duplex} from 'node:stream';
-import {finished} from 'node:stream/promises';
 import {scheduler, setTimeout as pSetTimeout} from 'node:timers/promises';
 import test from 'ava';
 import onetime from 'onetime';
 import getStream, {getStreamAsArray, MaxBufferError} from '../source/index.js';
 import {fixtureString, fixtureMultiString, prematureClose} from './fixtures/index.js';
+import {onFinishedStream} from './helpers/index.js';
 
-const onFinishedStream = stream => finished(stream, {cleanup: true});
 const noopMethods = {read() {}, write() {}};
 
 // eslint-disable-next-line max-params

test/web-stream-ponyfill.js

@@ -0,0 +1,13 @@
+import test from 'ava';
+
+// Emulate browsers that do not support those methods
+delete ReadableStream.prototype.values;
+delete ReadableStream.prototype[Symbol.asyncIterator];
+
+// Run those tests, but emulating browsers
+await import('./web-stream.js');
+
+test('Should not polyfill ReadableStream', t => {
+	t.is(ReadableStream.prototype.values, undefined);
+	t.is(ReadableStream.prototype[Symbol.asyncIterator], undefined);
+});

test/web-stream.js

@@ -0,0 +1,63 @@
+import test from 'ava';
+import getStream from '../source/index.js';
+import {fixtureString, fixtureMultiString} from './fixtures/index.js';
+import {readableStreamFrom, onFinishedStream} from './helpers/index.js';
+
+test('Can use ReadableStream', async t => {
+	const stream = readableStreamFrom(fixtureMultiString);
+	t.is(await getStream(stream), fixtureString);
+	await onFinishedStream(stream);
+});
+
+test('Can use already ended ReadableStream', async t => {
+	const stream = readableStreamFrom(fixtureMultiString);
+	t.is(await getStream(stream), fixtureString);
+	t.is(await getStream(stream), '');
+	await onFinishedStream(stream);
+});
+
+test('Can use already canceled ReadableStream', async t => {
+	let canceledValue;
+	const stream = new ReadableStream({
+		cancel(canceledError) {
+			canceledValue = canceledError;
+		},
+	});
+	const error = new Error('test');
+	await stream.cancel(error);
+	t.is(canceledValue, error);
+	t.is(await getStream(stream), '');
+	await onFinishedStream(stream);
+});
+
+test('Can use already errored ReadableStream', async t => {
+	const error = new Error('test');
+	const stream = new ReadableStream({
+		start(controller) {
+			controller.error(error);
+		},
+	});
+	t.is(await t.throwsAsync(getStream(stream)), error);
+	t.is(await t.throwsAsync(onFinishedStream(stream)), error);
+});
+
+test('Cancel ReadableStream when maxBuffer is hit', async t => {
+	let canceled = false;
+	const stream = new ReadableStream({
+		start(controller) {
+			controller.enqueue(fixtureString);
+			controller.enqueue(fixtureString);
+			controller.close();
+		},
+		cancel() {
+			canceled = true;
+		},
+	});
+	const error = await t.throwsAsync(
+		getStream(stream, {maxBuffer: 1}),
+		{message: /maxBuffer exceeded/},
+	);
+	t.deepEqual(error.bufferedData, fixtureString[0]);
+	await onFinishedStream(stream);
+	t.true(canceled);
+});