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
// 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",  );});