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
|
// |reftest| skip-if(!this.ReadableStream||!this.drainJobQueue)
if ("ignoreUnhandledRejections" in this) {
ignoreUnhandledRejections();
}
// Example of a stream that enqueues data asynchronously, whether the reader
// wants it or not, the "push" model.
let fbStream = new ReadableStream({
start(controller) {
simulatePacketsDriftingIn(controller);
},
});
async function simulatePacketsDriftingIn(controller) {
for (let i = 1; i <= 30; i++) {
let importantData =
(i % 15 == 0 ? "FizzBuzz" :
i % 5 == 0 ? "Buzz":
i % 3 == 0 ? "Fizz" :
String(i));
controller.enqueue(importantData);
await asyncSleep(1 + i % 7);
}
controller.close();
}
const expected = [
"1", "2", "Fizz", "4", "Buzz", "Fizz", "7", "8", "Fizz", "Buzz",
"11", "Fizz", "13", "14", "FizzBuzz", "16", "17", "Fizz", "19", "Buzz",
"Fizz", "22", "23", "Fizz", "Buzz", "26", "Fizz", "28", "29", "FizzBuzz"
];
async function test() {
assertEq(fbStream.locked, false);
let reader = fbStream.getReader();
assertEq(fbStream.locked, true);
let results = [];
while (true) {
let r = await reader.read();
if (r.done) {
break;
}
results.push(r.value);
}
assertEq(results.join("-"), expected.join("-"));
reader.releaseLock();
assertEq(fbStream.locked, false);
}
runAsyncTest(test);
|
