Renamed forEach*

Author: nicolas-van

modern-async.d.ts

@@ -92,18 +92,10 @@ declare module "findIndex" {
     function findIndex<V>(iterable: Iterable<V> | AsyncIterable<V>, iteratee: (value: V, index: number, iterable: Iterable<V> | AsyncIterable<V>) => Promise<boolean> | boolean, queueOrConcurrency: Queue | number, ordered?: boolean): Promise<number>;
     import Queue from "Queue";
 }
-declare module "forEachLimit" {
-    export default forEachLimit;
-    function forEachLimit<V>(iterable: Iterable<V> | AsyncIterable<V>, iteratee: (value: V, index: number, iterable: Iterable<V> | AsyncIterable<V>) => Promise<void> | void, queueOrConcurrency: Queue | number): Promise<void>;
-    import Queue from "Queue";
-}
 declare module "forEach" {
     export default forEach;
-    function forEach<V>(iterable: Iterable<V> | AsyncIterable<V>, iteratee: (value: V, index: number, iterable: Iterable<V> | AsyncIterable<V>) => Promise<void> | void): Promise<void>;
-}
-declare module "forEachSeries" {
-    export default forEachSeries;
-    function forEachSeries<V>(iterable: Iterable<V> | AsyncIterable<V>, iteratee: (value: V, index: number, iterable: Iterable<V> | AsyncIterable<V>) => Promise<void> | void): Promise<void>;
+    function forEach<V>(iterable: Iterable<V> | AsyncIterable<V>, iteratee: (value: V, index: number, iterable: Iterable<V> | AsyncIterable<V>) => Promise<void> | void, queueOrConcurrency: Queue | number): Promise<void>;
+    import Queue from "Queue";
 }
 declare module "mapLimit" {
     export default mapLimit;
@@ -207,8 +199,6 @@ declare module "modern-async" {
     export { default as find } from "find";
     export { default as findIndex } from "findIndex";
     export { default as forEach } from "forEach";
-    export { default as forEachLimit } from "forEachLimit";
-    export { default as forEachSeries } from "forEachSeries";
     export { default as map } from "map";
     export { default as mapGenerator } from "mapGenerator";
     export { default as mapLimit } from "mapLimit";

src/forEach.mjs

@@ -1,35 +1,44 @@
 
-import forEachLimit from './forEachLimit.mjs'
+import mapGenerator from './mapGenerator.mjs'
+import Queue from './Queue.mjs'
 
 /**
  * Calls a function on each element of iterable.
  *
- * Multiple calls to `iteratee` will be performed in parallel.
+ * The calls to `iteratee` will be performed in a queue to limit the concurrency of these calls.
  *
- * If any of the calls to iteratee throws an exception the returned promise will be rejected.
+ * If any of the calls to iteratee throws an exception the returned promise will be rejected and the remaining
+ * pending tasks will be cancelled.
  *
  * @param {Iterable | AsyncIterable} iterable An iterable or async iterable object.
  * @param {Function} iteratee A function that will be called with each member of the iterable. It will receive
  * three arguments:
  *   * `value`: The current value to process
  *   * `index`: The index in the iterable. Will start from 0.
  *   * `iterable`: The iterable on which the operation is being performed.
+ * @param {Queue | number} queueOrConcurrency If a queue is specified it will be used to schedule the calls to
+ * `iteratee`. If a number is specified it will be used as the concurrency of a Queue that will be created
+ * implicitly for the same purpose. Defaults to `1`.
  * @returns {Promise} A promise that will be resolved when all the calls to `iteratee` have been done.
  * This promise will be rejected if any call to `iteratee` throws an exception.
  * @example
  * import { forEach, sleep } from 'modern-async'
  *
  * const array = [1, 2, 3]
  * await forEach(array, async (v) => {
- *   // these calls will be performed in parallel
+ *   // these calls will be performed in parallel with a maximum of 2
+ *   // concurrent calls
  *   await sleep(Math.random() * 10) // waits a random amount of time between 0ms and 10ms
  *   console.log(v)
- * })
- * // prints 1, 2 and 3 in a random order
- * })
+ * }, 2)
+ * // prints 1, 2 and 3 in a random order (it will always print 1 or 2 before printing 3 due to
+ * // the concurrency limit and the internal scheduling order)
  */
-async function forEach (iterable, iteratee) {
-  return forEachLimit(iterable, iteratee, Number.POSITIVE_INFINITY)
+async function forEach (iterable, iteratee, queueOrConcurrency = 1) {
+  // eslint-disable-next-line no-unused-vars
+  for await (const _el of mapGenerator(iterable, iteratee, queueOrConcurrency)) {
+    // do nothing
+  }
 }
 
 export default forEach

src/forEach.test.mjs

@@ -3,14 +3,15 @@ import { expect, test } from '@jest/globals'
 import forEach from './forEach.mjs'
 import Deferred from './Deferred.mjs'
 import { range } from 'itertools'
+import delay from './delay.mjs'
 
 test('forEach base', async () => {
   const arr = [...range(6)]
   const called = {}
   arr.forEach((v) => { called[v] = 0 })
   await forEach(arr, async (x) => {
     called[x] += 1
-  })
+  }, 2)
   expect(called[0]).toBe(1)
   expect(called[1]).toBe(1)
   expect(called[2]).toBe(1)
@@ -25,7 +26,7 @@ test('forEach no async', async () => {
   arr.forEach((v) => { called[v] = 0 })
   await forEach(arr, (x) => {
     called[x] += 1
-  })
+  }, 2)
   expect(called[0]).toBe(1)
   expect(called[1]).toBe(1)
   expect(called[2]).toBe(1)
@@ -44,7 +45,72 @@ test('forEach concurrency', async () => {
     called[x] += 1
     ds[x].resolve()
     await d.promise
-  })
+  }, 2)
+  await ds[1].promise
+  expect(called[0]).toBe(1)
+  expect(called[1]).toBe(1)
+  expect(called[2]).toBe(0)
+  expect(called[3]).toBe(0)
+  expect(called[4]).toBe(0)
+  expect(called[5]).toBe(0)
+  await delay()
+  expect(called[0]).toBe(1)
+  expect(called[1]).toBe(1)
+  expect(called[2]).toBe(0)
+  expect(called[3]).toBe(0)
+  expect(called[4]).toBe(0)
+  expect(called[5]).toBe(0)
+  d.resolve()
+  await p
+  expect(called[0]).toBe(1)
+  expect(called[1]).toBe(1)
+  expect(called[2]).toBe(1)
+  expect(called[3]).toBe(1)
+  expect(called[4]).toBe(1)
+  expect(called[5]).toBe(1)
+})
+
+test('forEach infinite concurrency base', async () => {
+  const arr = [...range(6)]
+  const called = {}
+  arr.forEach((v) => { called[v] = 0 })
+  await forEach(arr, async (x) => {
+    called[x] += 1
+  }, Number.POSITIVE_INFINITY)
+  expect(called[0]).toBe(1)
+  expect(called[1]).toBe(1)
+  expect(called[2]).toBe(1)
+  expect(called[3]).toBe(1)
+  expect(called[4]).toBe(1)
+  expect(called[5]).toBe(1)
+})
+
+test('forEach infinite concurrency no async', async () => {
+  const arr = [...range(6)]
+  const called = {}
+  arr.forEach((v) => { called[v] = 0 })
+  await forEach(arr, (x) => {
+    called[x] += 1
+  }, Number.POSITIVE_INFINITY)
+  expect(called[0]).toBe(1)
+  expect(called[1]).toBe(1)
+  expect(called[2]).toBe(1)
+  expect(called[3]).toBe(1)
+  expect(called[4]).toBe(1)
+  expect(called[5]).toBe(1)
+})
+
+test('forEach infinite concurrency concurrency', async () => {
+  const arr = [...range(6)]
+  const called = {}
+  arr.forEach((v) => { called[v] = 0 })
+  const d = new Deferred()
+  const ds = arr.map(() => new Deferred())
+  const p = forEach(arr, async (x) => {
+    called[x] += 1
+    ds[x].resolve()
+    await d.promise
+  }, Number.POSITIVE_INFINITY)
   await ds[5].promise
   expect(called[0]).toBe(1)
   expect(called[1]).toBe(1)
@@ -61,3 +127,61 @@ test('forEach concurrency', async () => {
   expect(called[4]).toBe(1)
   expect(called[5]).toBe(1)
 })
+
+test('forEach concurrency 1 base', async () => {
+  const arr = [...range(6)]
+  const called = {}
+  arr.forEach((v) => { called[v] = 0 })
+  await forEach(arr, async (x) => {
+    called[x] += 1
+  })
+  expect(called[0]).toBe(1)
+  expect(called[1]).toBe(1)
+  expect(called[2]).toBe(1)
+  expect(called[3]).toBe(1)
+  expect(called[4]).toBe(1)
+  expect(called[5]).toBe(1)
+})
+
+test('forEach concurrency 1 no async', async () => {
+  const arr = [...range(6)]
+  const called = {}
+  arr.forEach((v) => { called[v] = 0 })
+  await forEach(arr, (x) => {
+    called[x] += 1
+  })
+  expect(called[0]).toBe(1)
+  expect(called[1]).toBe(1)
+  expect(called[2]).toBe(1)
+  expect(called[3]).toBe(1)
+  expect(called[4]).toBe(1)
+  expect(called[5]).toBe(1)
+})
+
+test('forEach concurrency 1 concurrency', async () => {
+  const arr = [...range(6)]
+  const called = {}
+  arr.forEach((v) => { called[v] = 0 })
+  const d = new Deferred()
+  const ds = arr.map(() => new Deferred())
+  const p = forEach(arr, async (x) => {
+    called[x] += 1
+    ds[x].resolve()
+    await d.promise
+  })
+  await ds[0].promise
+  expect(called[0]).toBe(1)
+  expect(called[1]).toBe(0)
+  expect(called[2]).toBe(0)
+  expect(called[3]).toBe(0)
+  expect(called[4]).toBe(0)
+  expect(called[5]).toBe(0)
+  d.resolve()
+  await p
+  expect(called[0]).toBe(1)
+  expect(called[1]).toBe(1)
+  expect(called[2]).toBe(1)
+  expect(called[3]).toBe(1)
+  expect(called[4]).toBe(1)
+  expect(called[5]).toBe(1)
+})