Skip to content

Commit 5023aa1

Browse files
committed
stream: add map method to Readable:
1 parent 640bfb8 commit 5023aa1

File tree

3 files changed

+80
-0
lines changed

3 files changed

+80
-0
lines changed

lib/internal/streams/operators.js

Lines changed: 15 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,15 @@
1+
'use strict';
2+
3+
const { AbortError } = require('internal/errors');
4+
const compose = require('internal/streams/compose');
5+
6+
module.exports.map = function map(stream, fn) {
7+
return compose(stream, async function* (source, { signal }) {
8+
for await (const item of source) {
9+
if (signal.aborted) {
10+
throw new AbortError('The iteration has been interrupted');
11+
}
12+
yield await fn(item, { signal });
13+
}
14+
});
15+
};

lib/stream.js

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -29,6 +29,7 @@ const {
2929
promisify: { custom: customPromisify },
3030
} = require('internal/util');
3131

32+
const { map } = require('internal/streams/operators');
3233
const compose = require('internal/streams/compose');
3334
const { pipeline } = require('internal/streams/pipeline');
3435
const { destroyer } = require('internal/streams/destroy');
@@ -40,6 +41,9 @@ const promises = require('stream/promises');
4041
const Stream = module.exports = require('internal/streams/legacy').Stream;
4142
Stream.isDisturbed = require('internal/streams/utils').isDisturbed;
4243
Stream.Readable = require('internal/streams/readable');
44+
Stream.Readable.prototype.map = function(fn) {
45+
return map(this, fn);
46+
};
4347
Stream.Writable = require('internal/streams/writable');
4448
Stream.Duplex = require('internal/streams/duplex');
4549
Stream.Transform = require('internal/streams/transform');

test/parallel/test-stream-map.js

Lines changed: 61 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,61 @@
1+
'use strict';
2+
3+
const common = require('../common');
4+
const {
5+
Readable,
6+
} = require('stream');
7+
const assert = require('assert');
8+
const { setTimeout } = require('timers/promises');
9+
10+
{
11+
// Map works on synchronous streams with a synchronous mapper
12+
const stream = Readable.from([1, 2, 3, 4, 5]).map((x) => x + x);
13+
const result = [2, 4, 6, 8, 10];
14+
(async () => {
15+
for await (const item of stream) {
16+
assert.strictEqual(item, result.shift());
17+
}
18+
})().then(common.mustCall());
19+
}
20+
21+
{
22+
// Map works on synchronous streams with an asynchronous mapper
23+
const stream = Readable.from([1, 2, 3, 4, 5]).map(async (x) => {
24+
await Promise.resolve();
25+
return x + x;
26+
});
27+
const result = [2, 4, 6, 8, 10];
28+
(async () => {
29+
for await (const item of stream) {
30+
assert.strictEqual(item, result.shift());
31+
}
32+
})().then(common.mustCall());
33+
}
34+
35+
{
36+
// Map works on asynchronous streams with a asynchronous mapper
37+
const stream = Readable.from([1, 2, 3, 4, 5]).map(async (x) => {
38+
return x + x;
39+
}).map((x) => x * x);
40+
const result = [4, 8, 12, 16, 20];
41+
(async () => {
42+
for await (const item of stream) {
43+
assert.strictEqual(item, result.shift());
44+
}
45+
})().then(common.mustCall());
46+
}
47+
48+
{
49+
// Allow cancellation of iteration through an AbortSignal
50+
51+
const stream = Readable.from([1, 2, 3, 4, 5]).map(async (x, { signal }) => {
52+
return setTimeout(1e15, { signal });
53+
});
54+
(async () => {
55+
const iterator = stream[Symbol.asyncIterator]();
56+
iterator.next();
57+
iterator.return();
58+
})().catch(common.mustCall((err) => {
59+
assert.equals(err.name, 'AbortError');
60+
}));
61+
}

0 commit comments

Comments
 (0)