Renamed find*

Author: nicolas-van

modern-async.d.ts

@@ -66,6 +66,7 @@ declare module "findIndexLimit" {
 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, queueOrConcurrency: Queue | number): Promise<boolean>;
+    import Queue from "Queue";
 }
 declare module "toArray" {
     export default toArray;
@@ -83,16 +84,13 @@ declare module "filterGenerator" {
 }
 declare module "filter" {
     export default filter;
-    function filter<V>(iterable: Iterable<V> | AsyncIterable<V>, iteratee: (value: V, index: number, iterable: Iterable<V> | AsyncIterable<V>) => Promise<boolean> | boolean, queueOrConcurrency: Queue | number): Promise<V[]>;
-}
-declare module "findLimit" {
-    export default findLimit;
-    function findLimit<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<V>;
+    function filter<V>(iterable: Iterable<V> | AsyncIterable<V>, iteratee: (value: V, index: number, iterable: Iterable<V> | AsyncIterable<V>) => Promise<boolean> | boolean, queueOrConcurrency?: Queue | number): Promise<V[]>;
     import Queue from "Queue";
 }
 declare module "find" {
     export default find;
-    function find<V>(iterable: Iterable<V> | AsyncIterable<V>, iteratee: (value: V, index: number, iterable: Iterable<V> | AsyncIterable<V>) => Promise<boolean> | boolean, ordered?: boolean): Promise<V>;
+    function find<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<V>;
+    import Queue from "Queue";
 }
 declare module "findIndex" {
     export default findIndex;
@@ -102,10 +100,6 @@ declare module "findIndexSeries" {
     export default findIndexSeries;
     function findIndexSeries<V>(iterable: Iterable<V> | AsyncIterable<V>, iteratee: (value: V, index: number, iterable: Iterable<V> | AsyncIterable<V>) => Promise<boolean> | boolean): Promise<number>;
 }
-declare module "findSeries" {
-    export default findSeries;
-    function findSeries<V>(iterable: Iterable<V> | AsyncIterable<V>, iteratee: (value: V, index: number, iterable: Iterable<V> | AsyncIterable<V>) => Promise<boolean> | boolean): Promise<V>;
-}
 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>;
@@ -222,8 +216,6 @@ declare module "modern-async" {
     export { default as findIndex } from "findIndex";
     export { default as findIndexLimit } from "findIndexLimit";
     export { default as findIndexSeries } from "findIndexSeries";
-    export { default as findLimit } from "findLimit";
-    export { default as findSeries } from "findSeries";
     export { default as forEach } from "forEach";
     export { default as forEachLimit } from "forEachLimit";
     export { default as forEachSeries } from "forEachSeries";

src/find.mjs

@@ -1,13 +1,16 @@
 
-import findLimit from './findLimit.mjs'
+import findInternal from './findInternal.mjs'
+import Queue from './Queue.mjs'
 
 /**
  * Returns the first element of an iterable that passes an asynchronous truth test.
  *
- * The calls to `iteratee` will run in parallel.
+ * The calls to `iteratee` will be performed in a queue to limit the concurrency of these calls.
  *
- * 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 result is found and an
+ * Whenever a result is found, 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 result is found and an
  * already started task throws an exception that exception will be plainly ignored.
  *
  * @param {Iterable | AsyncIterable} iterable An iterable or async iterable object.
@@ -16,23 +19,28 @@ import findLimit from './findLimit.mjs'
  *   * `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`.
  * @param {boolean} [ordered] If true this function will return on the first element in the iterable
  * order for which `iteratee` returned true. If false it will be the first in time.
  * @returns {Promise<any | undefined>} A promise that will be resolved with the first found value or rejected if one of the
  * `iteratee` calls throws an exception before finding a value. If no value is found it will return `undefined`.
  * @example
  * import { find, sleep } from 'modern-async'
  *
- * const array = [1, 2, 3]
+ * const array = [1, 2, 3, 4, 5]
  * const result = await find(array, async (v) => {
- *   // these calls will be performed in parallel
+ *   // these calls will be performed in parallel with a maximum of 3
+ *   // concurrent calls
  *   await sleep(Math.random() * 10) // waits a random amount of time between 0ms and 10ms
  *   return v % 2 === 1
- * })
+ * }, 3)
  * console.log(result) // prints 1
  */
-async function find (iterable, iteratee, ordered = false) {
-  return findLimit(iterable, iteratee, Number.POSITIVE_INFINITY, ordered)
+async function find (iterable, iteratee, queueOrConcurrency = 1, ordered = false) {
+  const result = (await findInternal(iterable, iteratee, queueOrConcurrency, ordered))[1]
+  return result
 }
 
 export default find

src/find.test.mjs

@@ -5,6 +5,45 @@ import Deferred from './Deferred.mjs'
 import { range } from 'itertools'
 
 test('find', async () => {
+  const arr = ['a', 'b', 'c']
+  const callCount = {}
+  ;[...range(3)].forEach((i) => { callCount[i] = 0 })
+  const d = new Deferred()
+  const ds = [...range(3)].map(() => new Deferred())
+  const p = find(arr, async (v, i) => {
+    callCount[i] += 1
+    ds[i].resolve()
+    await d.promise
+    return v === 'b'
+  }, 1)
+  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('b')
+  expect(callCount[0]).toBe(1)
+  expect(callCount[1]).toBe(1)
+  expect(callCount[2]).toBe(0)
+})
+
+test('find not found', async () => {
+  const arr = ['a', 'b', 'c']
+  const callCount = {}
+  ;[...range(3)].forEach((i) => { callCount[i] = 0 })
+  const d = new Deferred()
+  const p = find(arr, async (v, i) => {
+    callCount[i] += 1
+    await d.promise
+    return v === 'd'
+  }, 1)
+  d.resolve()
+  const res = await p
+  expect(res).toBe(arr.find((v) => v === 'd'))
+})
+
+test('find infinite concurrency', async () => {
   const arr = ['a', 'b', 'c']
   const callCount = {}
   ;[...range(3)].forEach((i) => { callCount[i] = 0 })
@@ -15,7 +54,7 @@ test('find', async () => {
     ds[i].resolve()
     await d.promise
     return v === 'b'
-  })
+  }, Number.POSITIVE_INFINITY)
   await ds[2].promise
   expect(callCount[0]).toBe(1)
   expect(callCount[1]).toBe(1)
@@ -27,3 +66,27 @@ test('find', async () => {
   expect(callCount[1]).toBe(1)
   expect(callCount[2]).toBe(1)
 })
+
+test('find concurrency 1', async () => {
+  const arr = ['a', 'b', 'c']
+  const callCount = {}
+  ;[...range(3)].forEach((i) => { callCount[i] = 0 })
+  const d = new Deferred()
+  const ds = [...range(3)].map(() => new Deferred())
+  const p = find(arr, async (v, i) => {
+    callCount[i] += 1
+    ds[i].resolve()
+    await d.promise
+    return v === 'b'
+  })
+  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('b')
+  expect(callCount[0]).toBe(1)
+  expect(callCount[1]).toBe(1)
+  expect(callCount[2]).toBe(0)
+})

src/findLimit.mjs

@@ -14,8 +14,6 @@ export { default as find } from './find.mjs'
 export { default as findIndex } from './findIndex.mjs'
 export { default as findIndexLimit } from './findIndexLimit.mjs'
 export { default as findIndexSeries } from './findIndexSeries.mjs'
-export { default as findLimit } from './findLimit.mjs'
-export { default as findSeries } from './findSeries.mjs'
 export { default as forEach } from './forEach.mjs'
 export { default as forEachLimit } from './forEachLimit.mjs'
 export { default as forEachSeries } from './forEachSeries.mjs'