didier amyot
0d627c9051
The class was already implemented in "https://deno.land/std@0.144.0/streams/conversion.ts". So, I litterally lost 1 day of work! I hate myself.
64 lines
2.2 KiB
TypeScript
64 lines
2.2 KiB
TypeScript
// Copyright 2018-2023 the Deno authors. All rights reserved. MIT license.
|
|
// This module is browser compatible.
|
|
|
|
/** Create a `ReadableStream` from any kind of iterable.
|
|
*
|
|
* ```ts
|
|
* import { readableStreamFromIterable } from "https://deno.land/std@$STD_VERSION/streams/readable_stream_from_iterable.ts";
|
|
*
|
|
* const r1 = readableStreamFromIterable(["foo, bar, baz"]);
|
|
* const r2 = readableStreamFromIterable(async function* () {
|
|
* await new Promise(((r) => setTimeout(r, 1000)));
|
|
* yield "foo";
|
|
* await new Promise(((r) => setTimeout(r, 1000)));
|
|
* yield "bar";
|
|
* await new Promise(((r) => setTimeout(r, 1000)));
|
|
* yield "baz";
|
|
* }());
|
|
* ```
|
|
*
|
|
* If the produced iterator (`iterable[Symbol.asyncIterator]()` or
|
|
* `iterable[Symbol.iterator]()`) is a generator, or more specifically is found
|
|
* to have a `.throw()` method on it, that will be called upon
|
|
* `readableStream.cancel()`. This is the case for the second input type above:
|
|
*
|
|
* ```ts
|
|
* import { readableStreamFromIterable } from "https://deno.land/std@$STD_VERSION/streams/readable_stream_from_iterable.ts";
|
|
*
|
|
* const r3 = readableStreamFromIterable(async function* () {
|
|
* try {
|
|
* yield "foo";
|
|
* } catch (error) {
|
|
* console.log(error); // Error: Cancelled by consumer.
|
|
* }
|
|
* }());
|
|
* const reader = r3.getReader();
|
|
* console.log(await reader.read()); // { value: "foo", done: false }
|
|
* await reader.cancel(new Error("Cancelled by consumer."));
|
|
* ```
|
|
*/
|
|
export function readableStreamFromIterable<T>(
|
|
iterable: Iterable<T> | AsyncIterable<T>,
|
|
): ReadableStream<T> {
|
|
const iterator: Iterator<T> | AsyncIterator<T> =
|
|
(iterable as AsyncIterable<T>)[Symbol.asyncIterator]?.() ??
|
|
(iterable as Iterable<T>)[Symbol.iterator]?.();
|
|
return new ReadableStream({
|
|
async pull(controller) {
|
|
const { value, done } = await iterator.next();
|
|
if (done) {
|
|
controller.close();
|
|
} else {
|
|
controller.enqueue(value);
|
|
}
|
|
},
|
|
async cancel(reason) {
|
|
if (typeof iterator.throw == "function") {
|
|
try {
|
|
await iterator.throw(reason);
|
|
} catch { /* `iterator.throw()` always throws on site. We catch it. */ }
|
|
}
|
|
},
|
|
});
|
|
}
|