'use strict';
|
const AggregateError = require('aggregate-error');
|
|
module.exports = async (
|
iterable,
|
mapper,
|
{
|
concurrency = Infinity,
|
stopOnError = true
|
} = {}
|
) => {
|
return new Promise((resolve, reject) => {
|
if (typeof mapper !== 'function') {
|
throw new TypeError('Mapper function is required');
|
}
|
|
if (!((Number.isSafeInteger(concurrency) || concurrency === Infinity) && concurrency >= 1)) {
|
throw new TypeError(`Expected \`concurrency\` to be an integer from 1 and up or \`Infinity\`, got \`${concurrency}\` (${typeof concurrency})`);
|
}
|
|
const result = [];
|
const errors = [];
|
const iterator = iterable[Symbol.iterator]();
|
let isRejected = false;
|
let isIterableDone = false;
|
let resolvingCount = 0;
|
let currentIndex = 0;
|
|
const next = () => {
|
if (isRejected) {
|
return;
|
}
|
|
const nextItem = iterator.next();
|
const index = currentIndex;
|
currentIndex++;
|
|
if (nextItem.done) {
|
isIterableDone = true;
|
|
if (resolvingCount === 0) {
|
if (!stopOnError && errors.length !== 0) {
|
reject(new AggregateError(errors));
|
} else {
|
resolve(result);
|
}
|
}
|
|
return;
|
}
|
|
resolvingCount++;
|
|
(async () => {
|
try {
|
const element = await nextItem.value;
|
result[index] = await mapper(element, index);
|
resolvingCount--;
|
next();
|
} catch (error) {
|
if (stopOnError) {
|
isRejected = true;
|
reject(error);
|
} else {
|
errors.push(error);
|
resolvingCount--;
|
next();
|
}
|
}
|
})();
|
};
|
|
for (let i = 0; i < concurrency; i++) {
|
next();
|
|
if (isIterableDone) {
|
break;
|
}
|
}
|
});
|
};
|
