1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103 |
x15
x15
x21
x21
x21
x27
x27
x27
x51
x95
x95
x127
x95
x106
x106
x106
x95
x74
x32
x32
x33
x33
x27
x21
x21 |
|
// Copyright 2018-2025 the Deno authors. MIT license.
// This module is browser compatible.
/**
* Merge multiple streams into a single one, taking order into account, and
* each stream will wait for a chunk to enqueue before the next stream can
* append another chunk.
*
* If a stream ends before other ones, the others will continue adding data in
* order, and the finished one will not add any more data. If you want to cancel
* the other streams when one of them ends, use {@linkcode earlyZipReadableStreams}.
*
* @typeparam T The type of the chunks in the input/output streams.
* @param streams An iterable of `ReadableStream`s to merge.
* @returns A `ReadableStream` that will emit the zipped chunks.
*
* @example Zip 2 streams with the same length
* ```ts
* import { zipReadableStreams } from "@std/streams/zip-readable-streams";
* import { assertEquals } from "@std/assert";
*
* const stream1 = ReadableStream.from(["1", "2", "3"]);
* const stream2 = ReadableStream.from(["a", "b", "c"]);
* const zippedStream = zipReadableStreams(stream1, stream2);
*
* assertEquals(
* await Array.fromAsync(zippedStream),
* ["1", "a", "2", "b", "3", "c"],
* );
* ```
*
* @example Zip 2 streams with different length (first one is shorter)
* ```ts
* import { zipReadableStreams } from "@std/streams/zip-readable-streams";
* import { assertEquals } from "@std/assert";
*
* const stream1 = ReadableStream.from(["1", "2"]);
* const stream2 = ReadableStream.from(["a", "b", "c", "d"]);
* const zippedStream = zipReadableStreams(stream1, stream2);
*
* assertEquals(
* await Array.fromAsync(zippedStream),
* ["1", "a", "2", "b", "c", "d"],
* );
* ```
*
* @example Zip 2 streams with different length (first one is longer)
* ```ts
* import { zipReadableStreams } from "@std/streams/zip-readable-streams";
* import { assertEquals } from "@std/assert";
*
* const stream1 = ReadableStream.from(["1", "2", "3", "4"]);
* const stream2 = ReadableStream.from(["a", "b"]);
* const zippedStream = zipReadableStreams(stream1, stream2);
*
* assertEquals(
* await Array.fromAsync(zippedStream),
* ["1", "a", "2", "b", "3", "4"],
* );
* ```
*
* @example Zip 3 streams
* ```ts
* import { zipReadableStreams } from "@std/streams/zip-readable-streams";
* import { assertEquals } from "@std/assert";
*
* const stream1 = ReadableStream.from(["1"]);
* const stream2 = ReadableStream.from(["a", "b"]);
* const stream3 = ReadableStream.from(["A", "B", "C"]);
* const zippedStream = zipReadableStreams(stream1, stream2, stream3);
*
* assertEquals(
* await Array.fromAsync(zippedStream),
* ["1", "a", "A", "b", "B", "C"],
* );
* ```
*/
export function zipReadableStreams<T>(
...streams: ReadableStream<T>[]
): ReadableStream<T> {
const readers = new Set(streams.map((s) => s.getReader()));
return new ReadableStream<T>({
async start(controller) {
try {
let resolved = 0;
while (resolved !== streams.length) {
for (const reader of readers) {
const { value, done } = await reader.read();
if (!done) {
controller.enqueue(value!);
} else {
resolved++;
readers.delete(reader);
}
}
}
controller.close();
} catch (e) {
controller.error(e);
}
},
});
}
|