Renamed filter*

Author: nicolas-van

modern-async.d.ts

@@ -81,18 +81,9 @@ declare module "filterGenerator" {
     function filterGenerator<V>(iterable: Iterable<V> | AsyncIterable<V>, iteratee: (value: V, index: number, iterable: Iterable<V> | AsyncIterable<V>) => Promise<boolean> | boolean, queueOrConcurrency?: Queue | number, ordered?: boolean): AsyncIterable<V>;
     import Queue from "Queue";
 }
-declare module "filterLimit" {
-    export default filterLimit;
-    function filterLimit<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 "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): Promise<V[]>;
-}
-declare module "filterSeries" {
-    export default filterSeries;
-    function filterSeries<V>(iterable: Iterable<V> | AsyncIterable<V>, iteratee: (value: V, index: number, iterable: Iterable<V> | AsyncIterable<V>) => Promise<boolean> | 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[]>;
 }
 declare module "findLimit" {
     export default findLimit;
@@ -227,8 +218,6 @@ declare module "modern-async" {
     export { default as every } from "every";
     export { default as filter } from "filter";
     export { default as filterGenerator } from "filterGenerator";
-    export { default as filterLimit } from "filterLimit";
-    export { default as filterSeries } from "filterSeries";
     export { default as find } from "find";
     export { default as findIndex } from "findIndex";
     export { default as findIndexLimit } from "findIndexLimit";

src/filter.mjs

@@ -1,36 +1,43 @@
 
-import filterLimit from './filterLimit.mjs'
+import Queue from './Queue.mjs'
+import toArray from './toArray.mjs'
+import filterGenerator from './filterGenerator.mjs'
 
 /**
  * Returns an array of all the values in `iterable` which pass an asynchronous truth test.
  *
- * The calls to `iteratee` will perform in parallel. The results will be in the same order than in `iterable`.
+ * The calls to `iteratee` will be performed in a queue to limit the concurrency of these calls.
+ * The results will be in the same order than in `iterable`.
  *
- * 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 `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<any[]>} A promise that will be resolved with an array containing all the values that passed
  * the truth test. This promise will be rejected if any of the `iteratee` calls throws an exception.
  * @example
  * import { filter, sleep } from 'modern-async'
  *
  * const array = [1, 2, 3]
  * const result = await filter(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 % 2 === 1
- * })
+ * }, 2)
  * console.log(result) // prints [1, 3]
- * // total processing time should be ~ 10ms
- * })
+ * // total processing time should be ~ 20ms
  */
-async function filter (iterable, iteratee) {
-  return filterLimit(iterable, iteratee, Number.POSITIVE_INFINITY)
+async function filter (iterable, iteratee, queueOrConcurrency = 1) {
+  return await toArray(filterGenerator(iterable, iteratee, queueOrConcurrency))
 }
 
 export default filter

src/filter.test.mjs

@@ -6,13 +6,13 @@ import { range } from 'itertools'
 
 test('filter base', async () => {
   const arr = [...range(6)]
-  const res = await filter(arr, async (x) => x % 2 === 0)
+  const res = await filter(arr, async (x) => x % 2 === 0, 2)
   expect(res).toEqual([0, 2, 4])
 })
 
 test('filter no async', async () => {
   const arr = [...range(6)]
-  const res = await filter(arr, (x) => x % 2 === 0)
+  const res = await filter(arr, (x) => x % 2 === 0, 2)
   expect(res).toEqual([0, 2, 4])
 })
 
@@ -27,7 +27,49 @@ test('filter concurrency', async () => {
     ds[x].resolve()
     await d.promise
     return x % 2 === 0
-  })
+  }, 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)
+  d.resolve()
+  const res = await p
+  expect(res).toEqual([0, 2, 4])
+  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('filter infinite concurrency base', async () => {
+  const arr = [...range(6)]
+  const res = await filter(arr, async (x) => x % 2 === 0, Number.POSITIVE_INFINITY)
+  expect(res).toEqual([0, 2, 4])
+})
+
+test('filter infinite concurrency no async', async () => {
+  const arr = [...range(6)]
+  const res = await filter(arr, (x) => x % 2 === 0, Number.POSITIVE_INFINITY)
+  expect(res).toEqual([0, 2, 4])
+})
+
+test('filter 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 = filter(arr, async (x) => {
+    called[x] += 1
+    ds[x].resolve()
+    await d.promise
+    return x % 2 === 0
+  }, Number.POSITIVE_INFINITY)
   await ds[5].promise
   expect(called[0]).toBe(1)
   expect(called[1]).toBe(1)
@@ -45,3 +87,45 @@ test('filter concurrency', async () => {
   expect(called[4]).toBe(1)
   expect(called[5]).toBe(1)
 })
+
+test('filter concurrency 1 base', async () => {
+  const arr = [...range(6)]
+  const res = await filter(arr, async (x) => x % 2 === 0)
+  expect(res).toEqual([0, 2, 4])
+})
+
+test('filter concurrency 1 no async', async () => {
+  const arr = [...range(6)]
+  const res = await filter(arr, (x) => x % 2 === 0)
+  expect(res).toEqual([0, 2, 4])
+})
+
+test('filter 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 = filter(arr, async (x) => {
+    called[x] += 1
+    ds[x].resolve()
+    await d.promise
+    return x % 2 === 0
+  })
+  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()
+  const res = await p
+  expect(res).toEqual([0, 2, 4])
+  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)
+})

src/filterLimit.mjs

@@ -10,8 +10,6 @@ export { default as Delayer } from './Delayer.mjs'
 export { default as every } from './every.mjs'
 export { default as filter } from './filter.mjs'
 export { default as filterGenerator } from './filterGenerator.mjs'
-export { default as filterLimit } from './filterLimit.mjs'
-export { default as filterSeries } from './filterSeries.mjs'
 export { default as find } from './find.mjs'
 export { default as findIndex } from './findIndex.mjs'
 export { default as findIndexLimit } from './findIndexLimit.mjs'