Node JS Design Patterns book

5_4_async_map.js

async function mapAsync(iterable, callback, concurrency) {
  if (!concurrency || !Number.isInteger(concurrency)) {
    throw new Error('"Concurrency" param must be a positive integer.');
  }

  // Split the iterable into a set of chunks. The execution of chunks is then
  // done in series to enforce the provided concurrency level.
  let slices = [];
  for (const [index, item] of iterable.entries()) {
    const sliceIndex = Math.floor(index / concurrency); 
    slices[sliceIndex] = [].concat((slices[sliceIndex] || []), item); 
  }
  
  console.log(`\r\nStart iteration of ${slices.length} slices. ${new Date()}`);
  
  // Serial execution of slices cumulating the results of each async slice mapping.
  let results = [];
  for (const [index, slice] of slices.entries()) {
    console.log(`\r\nProcessing slice ${index + 1}...`);

    const promises = slice.map(item => callback(item));
    const sliceMapRes = await Promise.all(promises);

    results = [...results, ...sliceMapRes];

    console.log(`End processing slice ${index + 1}\r\n`);
  }

  console.log(`End iteration of slices. ${new Date()}`);

  return results;
}

async function asyncValue(value, delay) {
  return new Promise((resolve, reject) => {
    const timeout = setTimeout(
      () => {
        resolve(value);
        clearTimeout(timeout);
      },
      delay || 0
    );
  });
}

const target = [1, 2, 3, 4, 5, 6, 7, 8, 9, 10];
const mapFn = item => asyncValue(item * 2, 1000);
const concurrency = 3;

mapAsync(target, mapFn, concurrency)
  .then(res => {
    console.log(res);
  })
  .catch(err => console.error(err));