-
Notifications
You must be signed in to change notification settings - Fork 28.2k
/
test-stream-reduce.js
109 lines (101 loc) Β· 3.02 KB
/
test-stream-reduce.js
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
104
105
106
107
108
109
'use strict';
const common = require('../common');
const {
Readable,
} = require('stream');
const assert = require('assert');
function sum(p, c) {
return p + c;
}
{
// Does the same thing as `(await stream.toArray()).reduce(...)`
(async () => {
const tests = [
[[], sum, 0],
[[1], sum, 0],
[[1, 2, 3, 4, 5], sum, 0],
[Array(100).fill().map((_, i) => i), sum, 0],
[['a', 'b', 'c'], sum, ''],
[[1, 2], sum],
[[1, 2, 3], (x, y) => y],
];
for (const [values, fn, initial] of tests) {
const streamReduce = await Readable.from(values)
.reduce(fn, initial);
const arrayReduce = values.reduce(fn, initial);
assert.deepStrictEqual(streamReduce, arrayReduce);
}
// Does the same thing as `(await stream.toArray()).reduce(...)` with an
// asynchronous reducer
for (const [values, fn, initial] of tests) {
const streamReduce = await Readable.from(values)
.map(async (x) => x)
.reduce(fn, initial);
const arrayReduce = values.reduce(fn, initial);
assert.deepStrictEqual(streamReduce, arrayReduce);
}
})().then(common.mustCall());
}
{
// Works with an async reducer, with or without initial value
(async () => {
const six = await Readable.from([1, 2, 3]).reduce(async (p, c) => p + c, 0);
assert.strictEqual(six, 6);
})().then(common.mustCall());
(async () => {
const six = await Readable.from([1, 2, 3]).reduce(async (p, c) => p + c);
assert.strictEqual(six, 6);
})().then(common.mustCall());
}
{
// Works lazily
assert.rejects(Readable.from([1, 2, 3, 4, 5, 6])
.map(common.mustCall((x) => {
return x;
}, 3)) // Two consumed and one buffered by `map` due to default concurrency
.reduce(async (p, c) => {
if (p === 1) {
throw new Error('boom');
}
return c;
}, 0)
, /boom/).then(common.mustCall());
}
{
// Support for AbortSignal
const ac = new AbortController();
assert.rejects(async () => {
await Readable.from([1, 2, 3]).reduce(async (p, c) => {
if (c === 3) {
await new Promise(() => {}); // Explicitly do not pass signal here
}
return Promise.resolve();
}, 0, { signal: ac.signal });
}, {
name: 'AbortError',
}).then(common.mustCall());
ac.abort();
}
{
// Support for AbortSignal - pre aborted
assert.rejects(async () => {
await Readable.from([1, 2, 3]).reduce(async (p, c) => {
if (c === 3) {
await new Promise(() => {}); // Explicitly do not pass signal here
}
return Promise.resolve();
}, 0, { signal: AbortSignal.abort() });
}, {
name: 'AbortError',
}).then(common.mustCall());
}
{
// Error cases
assert.rejects(() => Readable.from([]).reduce(1), /TypeError/);
assert.rejects(() => Readable.from([]).reduce('5'), /TypeError/);
}
{
// Test result is a Promise
const result = Readable.from([1, 2, 3, 4, 5]).reduce(sum, 0);
assert.ok(result instanceof Promise);
}