Rewrote "every"

Author: nicolas-van

modern-async.d.ts

@@ -63,18 +63,9 @@ declare module "findIndexLimit" {
     function findIndexLimit<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 "everyLimit" {
-    export default everyLimit;
-    function everyLimit<V>(iterable: Iterable<V> | AsyncIterable<V>, iteratee: (value: V, index: number, iterable: Iterable<V> | AsyncIterable<V>) => Promise<boolean> | boolean, queueOrConcurrency: Queue | number): Promise<boolean>;
-    import Queue from "Queue";
-}
 declare module "every" {
     export default every;
-    function every<V>(iterable: Iterable<V> | AsyncIterable<V>, iteratee: (value: V, index: number, iterable: Iterable<V> | AsyncIterable<V>) => Promise<boolean> | boolean): Promise<boolean>;
-}
-declare module "everySeries" {
-    export default everySeries;
-    function everySeries<V>(iterable: Iterable<V> | AsyncIterable<V>, iteratee: (value: V, index: number, iterable: Iterable<V> | AsyncIterable<V>) => Promise<boolean> | boolean): Promise<boolean>;
+    function every<V>(iterable: Iterable<V> | AsyncIterable<V>, iteratee: (value: V, index: number, iterable: Iterable<V> | AsyncIterable<V>) => Promise<boolean> | boolean, queueOrConcurrency: Queue | number): Promise<boolean>;
 }
 declare module "toArray" {
     export default toArray;
@@ -234,8 +225,6 @@ declare module "modern-async" {
     export { default as delayCancellable } from "delayCancellable";
     export { default as Delayer } from "Delayer";
     export { default as every } from "every";
-    export { default as everyLimit } from "everyLimit";
-    export { default as everySeries } from "everySeries";
     export { default as filter } from "filter";
     export { default as filterGenerator } from "filterGenerator";
     export { default as filterLimit } from "filterLimit";

src/every.mjs

@@ -1,21 +1,30 @@
 
-import everyLimit from './everyLimit.mjs'
+import findIndexLimit from './findIndexLimit.mjs'
+import Queue from './Queue.mjs'
+import asyncWrap from './asyncWrap.mjs'
+import assert from 'nanoassert'
 
 /**
  * Returns `true` if all elements of an iterable pass a truth test and `false` otherwise.
  *
- * The iteratee will be run in parallel. If any truth test returns `false` the promise is immediately resolved.
+ * The calls to `iteratee` will be performed in a queue to limit the concurrency of these calls.
+ * If any truth test returns `false` the promise is immediately resolved.
  *
- * In case of exception in one of the iteratee calls the promise returned by this function will be rejected
- * with the exception. In the very specific case where a test returns `false` and an already started task throws
- * an exception that exception will be plainly ignored.
+ * Whenever a test returns `false`, all the remaining tasks will be cancelled as long
+ * as they didn't started already. In case of exception in one of the iteratee calls the promise
+ * returned by this function will be rejected with the exception and the remaining pending
+ * tasks will also be cancelled. In the very specific case where a test returns `false` and an
+ * already started task throws an exception that exception will be plainly ignored.
  *
  * @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<boolean>} A promise that will be resolved to `true` if all values pass the truth test and `false`
  * if a least one of them doesn't pass it. That promise will be rejected if one of the truth test throws
  * an exception.
@@ -25,15 +34,22 @@ import everyLimit from './everyLimit.mjs'
  * const array = [1, 2, 3]
  *
  * const result = await every(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(10) // waits 10ms
  *   return v > 0
- * })
+ * }, 2)
  * console.log(result) // prints true
- * // total processing time should be ~ 10ms
+ * // total processing time should be ~ 20ms
  */
-async function every (iterable, iteratee) {
-  return everyLimit(iterable, iteratee, Number.POSITIVE_INFINITY)
+async function every (iterable, iteratee, queueOrConcurrency = 1) {
+  assert(typeof iteratee === 'function', 'iteratee must be a function')
+  iteratee = asyncWrap(iteratee)
+  const index = await findIndexLimit(iterable, async (value, index, iterable) => {
+    return !(await iteratee(value, index, iterable))
+  }, queueOrConcurrency, false)
+  const result = index === -1
+  return result
 }
 
 export default every

src/every.test.mjs

@@ -8,7 +8,123 @@ import delay from './delay.mjs'
 // eslint-disable-next-line require-jsdoc
 class TestError extends Error {}
 
-test('every all pass', async () => {
+test('every compatibility', async () => {
+  let d = new Deferred()
+  let p = every([...range(3)], async (v) => {
+    await d.promise
+    return true
+  }, 1)
+  d.resolve()
+  expect(await p).toBe([...range(3)].every((v) => true))
+
+  d = new Deferred()
+  p = every([...range(3)], async (v) => {
+    await d.promise
+    return v !== 2
+  }, 1)
+  d.resolve()
+  expect(await p).toBe([...range(3)].every((v) => v !== 2))
+
+  d = new Deferred()
+  p = every([...range(3)], async (v) => {
+    await d.promise
+    return false
+  }, 1)
+  d.resolve()
+  expect(await p).toBe([...range(3)].every((v) => false))
+
+  d = new Deferred()
+  p = every([], async (v) => {
+    await d.promise
+    return false
+  }, 1)
+  d.resolve()
+  expect(await p).toBe([].every((v) => false))
+
+  d = new Deferred()
+  p = every([], async (v) => {
+    await d.promise
+    return true
+  }, 1)
+  d.resolve()
+  expect(await p).toBe([].every((v) => true))
+})
+
+test('every parallel', async () => {
+  let d = new Deferred()
+  let p = every([...range(3)], async (v) => {
+    await d.promise
+    return true
+  }, 10)
+  d.resolve()
+  expect(await p).toBe([...range(3)].every((v) => true))
+
+  d = new Deferred()
+  p = every([...range(3)], async (v) => {
+    await d.promise
+    return v !== 2
+  }, 10)
+  d.resolve()
+  expect(await p).toBe([...range(3)].every((v) => v !== 2))
+
+  d = new Deferred()
+  p = every([...range(3)], async (v) => {
+    await d.promise
+    return false
+  }, 10)
+  d.resolve()
+  expect(await p).toBe([...range(3)].every((v) => false))
+
+  d = new Deferred()
+  p = every([], async (v) => {
+    await d.promise
+    return false
+  }, 10)
+  d.resolve()
+  expect(await p).toBe([].every((v) => false))
+
+  d = new Deferred()
+  p = every([], async (v) => {
+    await d.promise
+    return true
+  }, 10)
+  d.resolve()
+  expect(await p).toBe([].every((v) => true))
+})
+
+test('every first in time', async () => {
+  const ds = [...range(3)].map(() => new Deferred())
+  const p = every(range(3), async (v, i) => {
+    await ds[i]
+    return false
+  }, 3)
+  ds[2].resolve()
+  const res = await p
+  expect(res).toBe(false)
+})
+
+test('every error', async () => {
+  const callList = [...range(3)].map(() => 0)
+  const p = every([...range(3)], async (v, i) => {
+    callList[i] += 1
+    if (i === 1) {
+      throw new TestError()
+    }
+    return true
+  }, 1)
+  try {
+    await p
+    expect(true).toStrictEqual(false)
+  } catch (e) {
+    expect(e).toBeInstanceOf(TestError)
+  }
+  await delay()
+  expect(callList[0]).toStrictEqual(1)
+  expect(callList[1]).toStrictEqual(1)
+  expect(callList[2]).toStrictEqual(0)
+})
+
+test('every infinite concurrency all pass', async () => {
   const callCount = {}
   ;[...range(3)].forEach((i) => { callCount[i] = 0 })
   const d = new Deferred()
@@ -18,7 +134,7 @@ test('every all pass', async () => {
     ds[i].resolve()
     await d.promise
     return true
-  })
+  }, Number.POSITIVE_INFINITY)
   await ds[2].promise
   expect(callCount[0]).toBe(1)
   expect(callCount[1]).toBe(1)
@@ -31,7 +147,7 @@ test('every all pass', async () => {
   expect(callCount[2]).toBe(1)
 })
 
-test('every no all pass', async () => {
+test('every infinite concurrency no all pass', async () => {
   const callCount = {}
   ;[...range(3)].forEach((i) => { callCount[i] = 0 })
   const d = new Deferred()
@@ -45,7 +161,7 @@ test('every no all pass', async () => {
     } else {
       return true
     }
-  })
+  }, Number.POSITIVE_INFINITY)
   await ds[2].promise
   expect(callCount[0]).toBe(1)
   expect(callCount[1]).toBe(1)
@@ -58,19 +174,90 @@ test('every no all pass', async () => {
   expect(callCount[2]).toBe(1)
 })
 
-test('every error', async () => {
+test('every infinite concurrency error', async () => {
   const p = every([...range(3)], async (v, i) => {
     if (i === 1) {
       throw new TestError()
     }
     return true
+  }, Number.POSITIVE_INFINITY)
+
+  try {
+    await p
+    expect(true).toStrictEqual(false)
+  } catch (e) {
+    expect(e).toBeInstanceOf(TestError)
+  }
+  await delay()
+})
+
+test('every concurrency 1 all pass', async () => {
+  const callCount = {}
+  ;[...range(3)].forEach((i) => { callCount[i] = 0 })
+  const d = new Deferred()
+  const ds = [...range(3)].map(() => new Deferred())
+  const p = every([...range(3)], async (v, i) => {
+    callCount[i] += 1
+    ds[i].resolve()
+    await d.promise
+    return true
   })
+  await ds[0].promise
+  expect(callCount[0]).toBe(1)
+  expect(callCount[1]).toBe(0)
+  expect(callCount[2]).toBe(0)
+  d.resolve()
+  const res = await p
+  expect(res).toBe(true)
+  expect(callCount[0]).toBe(1)
+  expect(callCount[1]).toBe(1)
+  expect(callCount[2]).toBe(1)
+})
 
+test('every concurrency 1 no all pass', async () => {
+  const callCount = {}
+  ;[...range(3)].forEach((i) => { callCount[i] = 0 })
+  const d = new Deferred()
+  const ds = [...range(3)].map(() => new Deferred())
+  const p = every([...range(3)], async (v, i) => {
+    callCount[i] += 1
+    ds[i].resolve()
+    await d.promise
+    if (i === 1) {
+      return false
+    } else {
+      return true
+    }
+  })
+  await ds[0].promise
+  expect(callCount[0]).toBe(1)
+  expect(callCount[1]).toBe(0)
+  expect(callCount[2]).toBe(0)
+  d.resolve()
+  const res = await p
+  expect(res).toBe(false)
+  expect(callCount[0]).toBe(1)
+  expect(callCount[1]).toBe(1)
+  expect(callCount[2]).toBe(0)
+})
+
+test('every concurrency 1 error', async () => {
+  const callList = [...range(3)].map(() => 0)
+  const p = every([...range(3)], async (v, i) => {
+    callList[i] += 1
+    if (i === 1) {
+      throw new TestError()
+    }
+    return true
+  })
   try {
     await p
     expect(true).toStrictEqual(false)
   } catch (e) {
     expect(e).toBeInstanceOf(TestError)
   }
   await delay()
+  expect(callList[0]).toStrictEqual(1)
+  expect(callList[1]).toStrictEqual(1)
+  expect(callList[2]).toStrictEqual(0)
 })