Skip to content

Commit 5c59528

Browse files
RichienbRichienb
authoredDec 5, 2023
Add pMapIterable export (#63)
·
v7.0.3v7.0.0
1 parent 136b08a commit 5c59528

8 files changed

+324
-11
lines changed
 

‎assert-in-range.js

+10
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,10 @@
1+
import chalk from 'chalk';
2+
import inRange from 'in-range';
3+
4+
export default function assertInRange(t, value, {start = 0, end}) {
5+
if (inRange(value, {start, end})) {
6+
t.pass();
7+
} else {
8+
t.fail(`${start} ${start <= value ? '≤' : chalk.red('≰')} ${chalk.yellow(value)} ${value <= end ? '≤' : chalk.red('≰')} ${end}`);
9+
}
10+
}

‎index.d.ts

+35-1
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,4 @@
1-
export type Options = {
1+
type BaseOptions = {
22
/**
33
Number of concurrently pending promises returned by `mapper`.
44
@@ -7,7 +7,9 @@ export type Options = {
77
@default Infinity
88
*/
99
readonly concurrency?: number;
10+
};
1011

12+
export type Options = BaseOptions & {
1113
/**
1214
When `true`, the first mapper rejection will be rejected back to the consumer.
1315
@@ -42,6 +44,17 @@ export type Options = {
4244
readonly signal?: AbortSignal;
4345
};
4446

47+
export type IterableOptions = BaseOptions & {
48+
/**
49+
Maximum number of promises returned by `mapper` that have resolved but not yet collected by the consumer of the async iterable. Calls to `mapper` will be limited so that there is never too much backpressure.
50+
51+
Useful whenever you are consuming the iterable slower than what the mapper function can produce concurrently. For example, to avoid making an overwhelming number of HTTP requests if you are saving each of the results to a database.
52+
53+
Default: `options.concurrency`
54+
*/
55+
readonly backpressure?: number;
56+
};
57+
4558
type MaybePromise<T> = T | Promise<T>;
4659

4760
/**
@@ -88,6 +101,27 @@ export default function pMap<Element, NewElement>(
88101
options?: Options
89102
): Promise<Array<Exclude<NewElement, typeof pMapSkip>>>;
90103

104+
/**
105+
@param input - Synchronous or asynchronous iterable that is iterated over concurrently, calling the `mapper` function for each element. Each iterated item is `await`'d before the `mapper` is invoked so the iterable may return a `Promise` that resolves to an item. Asynchronous iterables (different from synchronous iterables that return `Promise` that resolves to an item) can be used when the next item may not be ready without waiting for an asynchronous process to complete and/or the end of the iterable may be reached after the asynchronous process completes. For example, reading from a remote queue when the queue has reached empty, or reading lines from a stream.
106+
@param mapper - Function which is called for every item in `input`. Expected to return a `Promise` or value.
107+
@returns An async iterable that streams each return value from `mapper` in order.
108+
109+
@example
110+
```
111+
import {pMapIterable} from 'p-map';
112+
113+
// Multiple posts are fetched concurrently, with limited concurrency and backpressure
114+
for await (const post of pMapIterable(postIds, getPostMetadata, {concurrency: 8})) {
115+
console.log(post);
116+
};
117+
```
118+
*/
119+
export function pMapIterable<Element, NewElement>(
120+
input: AsyncIterable<Element | Promise<Element>> | Iterable<Element | Promise<Element>>,
121+
mapper: Mapper<Element, NewElement>,
122+
options?: IterableOptions
123+
): AsyncIterable<Exclude<NewElement, typeof pMapSkip>>;
124+
91125
/**
92126
Return this value from a `mapper` function to skip including the value in the returned array.
93127

‎index.js

+104-1
Original file line numberDiff line numberDiff line change
@@ -16,7 +16,7 @@ export default async function pMap(
1616
throw new TypeError('Mapper function is required');
1717
}
1818

19-
if (!((Number.isSafeInteger(concurrency) || concurrency === Number.POSITIVE_INFINITY) && concurrency >= 1)) {
19+
if (!((Number.isSafeInteger(concurrency) && concurrency >= 1) || concurrency === Number.POSITIVE_INFINITY)) {
2020
throw new TypeError(`Expected \`concurrency\` to be an integer from 1 and up or \`Infinity\`, got \`${concurrency}\` (${typeof concurrency})`);
2121
}
2222

@@ -162,4 +162,107 @@ export default async function pMap(
162162
});
163163
}
164164

165+
export function pMapIterable(
166+
iterable,
167+
mapper,
168+
{
169+
concurrency = Number.POSITIVE_INFINITY,
170+
backpressure = concurrency,
171+
} = {},
172+
) {
173+
if (iterable[Symbol.iterator] === undefined && iterable[Symbol.asyncIterator] === undefined) {
174+
throw new TypeError(`Expected \`input\` to be either an \`Iterable\` or \`AsyncIterable\`, got (${typeof iterable})`);
175+
}
176+
177+
if (typeof mapper !== 'function') {
178+
throw new TypeError('Mapper function is required');
179+
}
180+
181+
if (!((Number.isSafeInteger(concurrency) && concurrency >= 1) || concurrency === Number.POSITIVE_INFINITY)) {
182+
throw new TypeError(`Expected \`concurrency\` to be an integer from 1 and up or \`Infinity\`, got \`${concurrency}\` (${typeof concurrency})`);
183+
}
184+
185+
if (!((Number.isSafeInteger(backpressure) && backpressure >= concurrency) || backpressure === Number.POSITIVE_INFINITY)) {
186+
throw new TypeError(`Expected \`backpressure\` to be an integer from \`concurrency\` (${concurrency}) and up or \`Infinity\`, got \`${backpressure}\` (${typeof backpressure})`);
187+
}
188+
189+
return {
190+
async * [Symbol.asyncIterator]() {
191+
const iterator = iterable[Symbol.asyncIterator] === undefined ? iterable[Symbol.iterator]() : iterable[Symbol.asyncIterator]();
192+
193+
const promises = [];
194+
let runningMappersCount = 0;
195+
let isDone = false;
196+
197+
function trySpawn() {
198+
if (isDone || !(runningMappersCount < concurrency && promises.length < backpressure)) {
199+
return;
200+
}
201+
202+
const promise = (async () => {
203+
const {done, value} = await iterator.next();
204+
205+
if (done) {
206+
return {done: true};
207+
}
208+
209+
runningMappersCount++;
210+
211+
// Spawn if still below concurrency and backpressure limit
212+
trySpawn();
213+
214+
try {
215+
const returnValue = await mapper(value);
216+
217+
runningMappersCount--;
218+
219+
if (returnValue === pMapSkip) {
220+
const index = promises.indexOf(promise);
221+
222+
if (index > 0) {
223+
promises.splice(index, 1);
224+
}
225+
}
226+
227+
// Spawn if still below backpressure limit and just dropped below concurrency limit
228+
trySpawn();
229+
230+
return {done: false, value: returnValue};
231+
} catch (error) {
232+
isDone = true;
233+
return {error};
234+
}
235+
})();
236+
237+
promises.push(promise);
238+
}
239+
240+
trySpawn();
241+
242+
while (promises.length > 0) {
243+
const {error, done, value} = await promises[0]; // eslint-disable-line no-await-in-loop
244+
245+
promises.shift();
246+
247+
if (error) {
248+
throw error;
249+
}
250+
251+
if (done) {
252+
return;
253+
}
254+
255+
// Spawn if just dropped below backpressure limit and below the concurrency limit
256+
trySpawn();
257+
258+
if (value === pMapSkip) {
259+
continue;
260+
}
261+
262+
yield value;
263+
}
264+
},
265+
};
266+
}
267+
165268
export const pMapSkip = Symbol('skip');

‎index.test-d.ts

+3-1
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,5 @@
11
import {expectType, expectAssignable} from 'tsd';
2-
import pMap, {type Options, type Mapper, pMapSkip} from './index.js';
2+
import pMap, {pMapIterable, type Options, type Mapper, pMapSkip} from './index.js';
33

44
const sites = [
55
'https://sindresorhus.com',
@@ -48,3 +48,5 @@ expectType<Promise<number[]>>(pMap(numbers, (number: number) => {
4848

4949
return pMapSkip;
5050
}));
51+
52+
expectType<AsyncIterable<string>>(pMapIterable(sites, asyncMapper));

‎package.json

+1
Original file line numberDiff line numberDiff line change
@@ -42,6 +42,7 @@
4242
],
4343
"devDependencies": {
4444
"ava": "^5.2.0",
45+
"chalk": "^5.3.0",
4546
"delay": "^5.0.0",
4647
"in-range": "^3.0.0",
4748
"random-int": "^3.0.0",

‎readme.md

+29
Original file line numberDiff line numberDiff line change
@@ -41,6 +41,19 @@ console.log(result);
4141

4242
Returns a `Promise` that is fulfilled when all promises in `input` and ones returned from `mapper` are fulfilled, or rejects if any of the promises reject. The fulfilled value is an `Array` of the fulfilled values returned from `mapper` in `input` order.
4343

44+
### pMapIterable(input, mapper, options?)
45+
46+
Returns an async iterable that streams each return value from `mapper` in order.
47+
48+
```js
49+
import {pMapIterable} from 'p-map';
50+
51+
// Multiple posts are fetched concurrently, with limited concurrency and backpressure
52+
for await (const post of pMapIterable(postIds, getPostMetadata, {concurrency: 8})) {
53+
console.log(post);
54+
};
55+
```
56+
4457
#### input
4558

4659
Type: `AsyncIterable<Promise<unknown> | unknown> | Iterable<Promise<unknown> | unknown>`
@@ -67,8 +80,22 @@ Minimum: `1`
6780

6881
Number of concurrently pending promises returned by `mapper`.
6982

83+
##### backpressure
84+
85+
**Only for `pMapInterable`**
86+
87+
Type: `number` *(Integer)*\
88+
Default: `options.concurrency`\
89+
Minimum: `options.concurrency`
90+
91+
Maximum number of promises returned by `mapper` that have resolved but not yet collected by the consumer of the async iterable. Calls to `mapper` will be limited so that there is never too much backpressure.
92+
93+
Useful whenever you are consuming the iterable slower than what the mapper function can produce concurrently. For example, to avoid making an overwhelming number of HTTP requests if you are saving each of the results to a database.
94+
7095
##### stopOnError
7196

97+
**Only for `pMap`**
98+
7299
Type: `boolean`\
73100
Default: `true`
74101

@@ -80,6 +107,8 @@ Caveat: When `true`, any already-started async mappers will continue to run unti
80107

81108
##### signal
82109

110+
**Only for `pMap`**
111+
83112
Type: [`AbortSignal`](https://developer.mozilla.org/en-US/docs/Web/API/AbortSignal)
84113

85114
You can abort the promises using [`AbortController`](https://developer.mozilla.org/en-US/docs/Web/API/AbortController).

‎test-multiple-pmapskips-performance.js

+2-2
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,6 @@
11
import test from 'ava';
2-
import inRange from 'in-range';
32
import timeSpan from 'time-span';
3+
import assertInRange from './assert-in-range.js';
44
import pMap, {pMapSkip} from './index.js';
55

66
function generateSkipPerformanceData(length) {
@@ -32,6 +32,6 @@ test('multiple pMapSkips - algorithmic complexity', async t => {
3232
// shorter test. This is not perfect... there is some fluctuation.
3333
// The idea here is to catch a regression that makes `pMapSkip` handling O(n^2)
3434
// on the number of `pMapSkip` items in the input.
35-
t.true(inRange(longerDuration, {start: 1.2 * smallerDuration, end: 15 * smallerDuration}));
35+
assertInRange(t, longerDuration, {start: 1.2 * smallerDuration, end: 15 * smallerDuration});
3636
}
3737
});

‎test.js

+140-6
Original file line numberDiff line numberDiff line change
@@ -1,16 +1,24 @@
11
import test from 'ava';
22
import delay from 'delay';
3-
import inRange from 'in-range';
43
import timeSpan from 'time-span';
54
import randomInt from 'random-int';
6-
import pMap, {pMapSkip} from './index.js';
5+
import assertInRange from './assert-in-range.js';
6+
import pMap, {pMapIterable, pMapSkip} from './index.js';
77

88
const sharedInput = [
99
[async () => 10, 300],
1010
[20, 200],
1111
[30, 100],
1212
];
1313

14+
const longerSharedInput = [
15+
[10, 300],
16+
[20, 200],
17+
[30, 100],
18+
[40, 50],
19+
[50, 25],
20+
];
21+
1422
const errorInput1 = [
1523
[20, 200],
1624
[30, 100],
@@ -33,6 +41,14 @@ const errorInput2 = [
3341
}, 10],
3442
];
3543

44+
const errorInput3 = [
45+
[20, 10],
46+
[async () => {
47+
throw new Error('bar');
48+
}, 100],
49+
[30, 100],
50+
];
51+
3652
const mapper = async ([value, ms]) => {
3753
await delay(ms);
3854

@@ -81,13 +97,13 @@ test('main', async t => {
8197
t.deepEqual(await pMap(sharedInput, mapper), [10, 20, 30]);
8298

8399
// We give it some leeway on both sides of the expected 300ms as the exact value depends on the machine and workload.
84-
t.true(inRange(end(), {start: 290, end: 430}));
100+
assertInRange(t, end(), {start: 290, end: 430});
85101
});
86102

87103
test('concurrency: 1', async t => {
88104
const end = timeSpan();
89105
t.deepEqual(await pMap(sharedInput, mapper, {concurrency: 1}), [10, 20, 30]);
90-
t.true(inRange(end(), {start: 590, end: 760}));
106+
assertInRange(t, end(), {start: 590, end: 760});
91107
});
92108

93109
test('concurrency: 4', async t => {
@@ -217,13 +233,13 @@ test('asyncIterator - main', async t => {
217233
t.deepEqual(await pMap(new AsyncTestData(sharedInput), mapper), [10, 20, 30]);
218234

219235
// We give it some leeway on both sides of the expected 300ms as the exact value depends on the machine and workload.
220-
t.true(inRange(end(), {start: 290, end: 430}));
236+
assertInRange(t, end(), {start: 290, end: 430});
221237
});
222238

223239
test('asyncIterator - concurrency: 1', async t => {
224240
const end = timeSpan();
225241
t.deepEqual(await pMap(new AsyncTestData(sharedInput), mapper, {concurrency: 1}), [10, 20, 30]);
226-
t.true(inRange(end(), {start: 590, end: 760}));
242+
assertInRange(t, end(), {start: 590, end: 760});
227243
});
228244

229245
test('asyncIterator - concurrency: 4', async t => {
@@ -485,3 +501,121 @@ if (globalThis.AbortController !== undefined) {
485501
});
486502
});
487503
}
504+
505+
async function collectAsyncIterable(asyncIterable) {
506+
const values = [];
507+
508+
for await (const value of asyncIterable) {
509+
values.push(value);
510+
}
511+
512+
return values;
513+
}
514+
515+
test('pMapIterable', async t => {
516+
t.deepEqual(await collectAsyncIterable(pMapIterable(sharedInput, mapper)), [10, 20, 30]);
517+
});
518+
519+
test('pMapIterable - empty', async t => {
520+
t.deepEqual(await collectAsyncIterable(pMapIterable([], mapper)), []);
521+
});
522+
523+
test('pMapIterable - iterable that throws', async t => {
524+
let isFirstNextCall = true;
525+
526+
const iterable = {
527+
[Symbol.asyncIterator]() {
528+
return {
529+
async next() {
530+
if (!isFirstNextCall) {
531+
return {done: true};
532+
}
533+
534+
isFirstNextCall = false;
535+
throw new Error('foo');
536+
},
537+
};
538+
},
539+
};
540+
541+
const iterator = pMapIterable(iterable, mapper)[Symbol.asyncIterator]();
542+
543+
await t.throwsAsync(iterator.next(), {message: 'foo'});
544+
});
545+
546+
test('pMapIterable - mapper that throws', async t => {
547+
await t.throwsAsync(collectAsyncIterable(pMapIterable(sharedInput, async () => {
548+
throw new Error('foo');
549+
})), {message: 'foo'});
550+
});
551+
552+
test('pMapIterable - stop on error', async t => {
553+
const output = [];
554+
555+
try {
556+
for await (const value of pMapIterable(errorInput3, mapper)) {
557+
output.push(value);
558+
}
559+
} catch (error) {
560+
t.is(error.message, 'bar');
561+
}
562+
563+
t.deepEqual(output, [20]);
564+
});
565+
566+
test('pMapIterable - concurrency: 1', async t => {
567+
const end = timeSpan();
568+
t.deepEqual(await collectAsyncIterable(pMapIterable(sharedInput, mapper, {concurrency: 1, backpressure: Number.POSITIVE_INFINITY})), [10, 20, 30]);
569+
570+
// It could've only taken this much time if each were run in series
571+
assertInRange(t, end(), {start: 590, end: 760});
572+
});
573+
574+
test('pMapIterable - concurrency: 2', async t => {
575+
const times = new Map();
576+
const end = timeSpan();
577+
578+
t.deepEqual(await collectAsyncIterable(pMapIterable(longerSharedInput, value => {
579+
times.set(value[0], end());
580+
return mapper(value);
581+
}, {concurrency: 2, backpressure: Number.POSITIVE_INFINITY})), [10, 20, 30, 40, 50]);
582+
583+
assertInRange(t, times.get(10), {start: 0, end: 50});
584+
assertInRange(t, times.get(20), {start: 0, end: 50});
585+
assertInRange(t, times.get(30), {start: 200, end: 250});
586+
assertInRange(t, times.get(40), {start: 300, end: 350});
587+
assertInRange(t, times.get(50), {start: 300, end: 350});
588+
});
589+
590+
test('pMapIterable - backpressure', async t => {
591+
let currentValue;
592+
593+
// Concurrency option is forced by an early check
594+
const asyncIterator = pMapIterable(longerSharedInput, async value => {
595+
currentValue = await mapper(value);
596+
return currentValue;
597+
}, {backpressure: 2, concurrency: 2})[Symbol.asyncIterator]();
598+
599+
const {value: value1} = await asyncIterator.next();
600+
t.is(value1, 10);
601+
602+
// If backpressure is not respected, than all items will be evaluated in this time
603+
await delay(600);
604+
605+
t.is(currentValue, 30);
606+
607+
const {value: value2} = await asyncIterator.next();
608+
t.is(value2, 20);
609+
610+
await delay(100);
611+
612+
t.is(currentValue, 40);
613+
});
614+
615+
test('pMapIterable - pMapSkip', async t => {
616+
t.deepEqual(await collectAsyncIterable(pMapIterable([
617+
1,
618+
pMapSkip,
619+
2,
620+
], async value => value)), [1, 2]);
621+
});

0 commit comments

Comments
 (0)
Please sign in to comment.