Fixed map, filter and foreach behavior in case of exception

CHANGELOG.md

@@ -1,5 +1,10 @@
 # Changelog
 
+## to be released
+
+* Fixed map, filter and forEach behavior in case of exception. Pending tasks were not cancelled in case of exception and the
+  behavior in that case was not documented as opposed to other functions.
+
 ## 1.0.0
 
 First version.

src/filter.mjs

@@ -7,6 +7,8 @@ import filterLimit from './filterLimit.mjs'
  * The calls to `iteratee` will perform in parallel, but the results array will be in the same order
  * than the original.
  *
+ * If any of the calls to iteratee throws an exception the returned promised will be rejected.
+ *
  * @param {Iterable} iterable An iterable object.
  * @param {Function} iteratee A function that will be called with each member of `iterable`. It will receive
  * three arguments:

src/filterLimit.mjs

@@ -8,6 +8,9 @@ import assert from 'nanoassert'
  * The calls to `iteratee` will perform in parallel, up to the concurrency limit, but the results array will be
  * in the same order than the original.
  *
+ * If any of the calls to iteratee throws an exception the returned promised will be rejected and the remaining
+ * pending tasks will be cancelled.
+ *
  * @param {Iterable} iterable An iterable object.
  * @param {Function} iteratee A function that will be called with each member of `iterable`. It will receive
  * three arguments:

src/filterSeries.mjs

@@ -7,6 +7,9 @@ import filterLimit from './filterLimit.mjs'
  * The calls to `iteratee` will perform sequentially and the results array will be in the same order
  * than the original.
  *
+ * If any of the calls to iteratee throws an exception the returned promised will be rejected and the remaining
+ * pending tasks will be cancelled.
+ *
  * @param {Iterable} iterable An iterable object.
  * @param {Function} iteratee A function that will be called with each member of the iterable. It will receive
  * three arguments:

src/forEach.mjs

@@ -6,6 +6,8 @@ import forEachLimit from './forEachLimit.mjs'
  *
  * Multiple calls to `iteratee` will be performed in parallel.
  *
+ * If any of the calls to iteratee throws an exception the returned promised will be rejected.
+ *
  * @param {Iterable} iterable An iterable object.
  * @param {Function} iteratee A function that will be called with each member of the iterable. It will receive
  * three arguments:

src/forEachLimit.mjs

@@ -6,6 +6,9 @@ import mapLimit from './mapLimit.mjs'
  *
  * Multiple calls to `iteratee` will be performed in parallel, up to the concurrency limit.
  *
+ * If any of the calls to iteratee throws an exception the returned promised will be rejected and the remaining
+ * pending tasks will be cancelled.
+ *
  * @param {Iterable} iterable An iterable object.
  * @param {Function} iteratee A function that will be called with each member of the iterable. It will receive
  * three arguments:

src/forEachSeries.mjs

@@ -6,6 +6,9 @@ import forEachLimit from './forEachLimit.mjs'
  *
  * Multiple calls to `iteratee` will be performed sequentially.
  *
+ * If any of the calls to iteratee throws an exception the returned promised will be rejected and the remaining
+ * pending tasks will be cancelled.
+ *
  * @param {Iterable} iterable An iterable object.
  * @param {Function} iteratee A function that will be called with each member of the iterable. It will receive
  * three arguments:

src/map.mjs

@@ -6,6 +6,8 @@ import mapLimit from './mapLimit.mjs'
  *
  * Multiple calls to `iteratee` will be performed in parallel.
  *
+ * If any of the calls to iteratee throws an exception the returned promised will be rejected.
+ *
  * @param {Iterable} iterable An iterable object.
  * @param {Function} iteratee A function that will be called with each member of the iterable. It will receive
  * three arguments:

src/mapLimit.mjs

@@ -7,6 +7,9 @@ import assert from 'nanoassert'
  *
  * Multiple calls to `iteratee` will be performed in parallel, up to the concurrency limit.
  *
+ * If any of the calls to iteratee throws an exception the returned promised will be rejected and the remaining
+ * pending tasks will be cancelled.
+ *
  * @param {Iterable} iterable An iterable object.
  * @param {Function} iteratee A function that will be called with each member of the iterable. It will receive
  * three arguments:
@@ -43,7 +46,11 @@ async function mapLimit (iterable, iteratee, concurrency) {
     }))
     i += 1
   }
-  return Promise.all(promises)
+  try {
+    return await Promise.all(promises)
+  } finally {
+    queue.cancelAllPending()
+  }
 }
 
 export default mapLimit

src/mapLimit.test.mjs

@@ -51,3 +51,66 @@ test('mapLimit index & iterable', async () => {
   }, 2)
   expect(res).toEqual([0, 2, 4, 6, 8, 10])
 })
+
+test('mapLimit one exception', async () => {
+  const arr = _.range(3)
+  const called = {}
+  arr.forEach((v) => { called[v] = 0 })
+  try {
+    await mapLimit(arr, async (x) => {
+      called[x] += 1
+      await sleepPrecise(10)
+      if (x === 1) {
+        throw new Error('test')
+      }
+      return x * 2
+    }, 1)
+    expect(false).toBe(true)
+  } catch (e) {
+    expect(e.message).toBe('test')
+  }
+  await sleepPrecise(100)
+  expect(called[0]).toBe(1)
+  expect(called[1]).toBe(1)
+  expect(called[2]).toBe(0)
+})
+
+test('mapLimit all exception c 1', async () => {
+  const arr = _.range(3)
+  const called = {}
+  arr.forEach((v) => { called[v] = 0 })
+  try {
+    await mapLimit(arr, async (x) => {
+      called[x] += 1
+      await sleepPrecise(10)
+      throw new Error('test')
+    }, 1)
+    expect(false).toBe(true)
+  } catch (e) {
+    expect(e.message).toBe('test')
+  }
+  await sleepPrecise(100)
+  expect(called[0]).toBe(1)
+  expect(called[1]).toBe(0)
+  expect(called[2]).toBe(0)
+})
+
+test('mapLimit all exception c 2', async () => {
+  const arr = _.range(3)
+  const called = {}
+  arr.forEach((v) => { called[v] = 0 })
+  try {
+    await mapLimit(arr, async (x) => {
+      called[x] += 1
+      await sleepPrecise(10)
+      throw new Error('test')
+    }, 2)
+    expect(false).toBe(true)
+  } catch (e) {
+    expect(e.message).toBe('test')
+  }
+  await sleepPrecise(100)
+  expect(called[0]).toBe(1)
+  expect(called[1]).toBe(1)
+  expect(called[2]).toBe(0)
+})

src/mapSeries.mjs

@@ -6,6 +6,9 @@ import mapLimit from './mapLimit.mjs'
  *
  * Multiple calls to `iteratee` will be performed sequentially.
  *
+ * If any of the calls to iteratee throws an exception the returned promised will be rejected and the remaining
+ * pending tasks will be cancelled.
+ *
  * @param {Iterable} iterable An iterable object.
  * @param {Function} iteratee A function that will be called with each member of the iterable. It will receive
  * three arguments: