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
|
// Copyright Node.js contributors. All rights reserved. MIT License.
import { Buffer } from "../buffer.ts";
import Readable from "./readable.ts";
import Writable from "./writable.ts";
import { pipeline } from "./promises.ts";
import { deferred } from "../../async/mod.ts";
import {
assert,
assertEquals,
assertThrowsAsync,
} from "../../testing/asserts.ts";
Deno.test("Promise pipeline works correctly", async () => {
let pipelineExecuted = 0;
const pipelineExecutedExpected = 1;
const pipelineExpectedExecutions = deferred();
let finished = false;
// deno-lint-ignore no-explicit-any
const processed: any[] = [];
const expected = [
Buffer.from("a"),
Buffer.from("b"),
Buffer.from("c"),
];
const read = new Readable({
read() {},
});
const write = new Writable({
write(data, _enc, cb) {
processed.push(data);
cb();
},
});
write.on("finish", () => {
finished = true;
});
for (let i = 0; i < expected.length; i++) {
read.push(expected[i]);
}
read.push(null);
pipeline(read, write).then(() => {
pipelineExecuted++;
if (pipelineExecuted == pipelineExecutedExpected) {
pipelineExpectedExecutions.resolve();
}
assert(finished);
assertEquals(processed, expected);
});
const pipelineTimeout = setTimeout(
() => pipelineExpectedExecutions.reject(),
1000,
);
await pipelineExpectedExecutions;
clearTimeout(pipelineTimeout);
assertEquals(pipelineExecuted, pipelineExecutedExpected);
});
Deno.test("Promise pipeline throws on readable destroyed", async () => {
const read = new Readable({
read() {},
});
const write = new Writable({
write(_data, _enc, cb) {
cb();
},
});
read.push("data");
read.destroy();
await assertThrowsAsync(
() => pipeline(read, write),
Error,
"Premature close",
);
});
|
