feat(async): async functions in parallel are now called with an add…

…itional index parameter

test(async): added test for the new index parameter in `parallel`

cdn/radash.esm.js

@@ -383,21 +383,23 @@ const parallel = async (limit, array, func) => {
     index,
     item
   }));
-  const processor = async (res) => {
+  const processor = (queueIndex) => async (res) => {
     const results2 = [];
     while (true) {
       const next = work.pop();
       if (!next)
         return res(results2);
-      const [error, result] = await tryit(func)(next.item);
+      const [error, result] = await tryit(func)(next.item, queueIndex);
       results2.push({
         error,
         result,
         index: next.index
       });
     }
   };
-  const queues = list(1, limit).map(() => new Promise(processor));
+  const queues = list(1, limit).map(
+    (queueIndex) => new Promise(processor(queueIndex - 1))
+  );
   const itemResults = await Promise.all(queues);
   const [errors, results] = fork(
     sort(itemResults.flat(), (r) => r.index),

cdn/radash.js

@@ -386,21 +386,23 @@ var radash = (function (exports) {
       index,
       item
     }));
-    const processor = async (res) => {
+    const processor = (queueIndex) => async (res) => {
       const results2 = [];
       while (true) {
         const next = work.pop();
         if (!next)
           return res(results2);
-        const [error, result] = await tryit(func)(next.item);
+        const [error, result] = await tryit(func)(next.item, queueIndex);
         results2.push({
           error,
           result,
           index: next.index
         });
       }
     };
-    const queues = list(1, limit).map(() => new Promise(processor));
+    const queues = list(1, limit).map(
+      (queueIndex) => new Promise(processor(queueIndex - 1))
+    );
     const itemResults = await Promise.all(queues);
     const [errors, results] = fork(
       sort(itemResults.flat(), (r) => r.index),

package.json

@@ -1,6 +1,6 @@
 {
   "name": "radash",
-  "version": "10.8.1",
+  "version": "10.8.2",
   "description": "Functional utility library - modern, simple, typed, powerful",
   "main": "dist/cjs/index.cjs",
   "module": "dist/esm/index.mjs",

src/async.ts

@@ -111,28 +111,32 @@ export class AggregateError extends Error {
 export const parallel = async <T, K>(
   limit: number,
   array: readonly T[],
-  func: (item: T) => Promise<K>
+  func: (item: T, index: number) => Promise<K>
 ): Promise<K[]> => {
   const work = array.map((item, index) => ({
     index,
     item
   }))
   // Process array items
-  const processor = async (res: (value: WorkItemResult<K>[]) => void) => {
-    const results: WorkItemResult<K>[] = []
-    while (true) {
-      const next = work.pop()
-      if (!next) return res(results)
-      const [error, result] = await tryit(func)(next.item)
-      results.push({
-        error,
-        result: result as K,
-        index: next.index
-      })
+  const processor =
+    (queueIndex: number) =>
+    async (res: (value: WorkItemResult<K>[]) => void) => {
+      const results: WorkItemResult<K>[] = []
+      while (true) {
+        const next = work.pop()
+        if (!next) return res(results)
+        const [error, result] = await tryit(func)(next.item, queueIndex)
+        results.push({
+          error,
+          result: result as K,
+          index: next.index
+        })
+      }
     }
-  }
   // Create queues
-  const queues = list(1, limit).map(() => new Promise(processor))
+  const queues = list(1, limit).map(
+    queueIndex => new Promise(processor(queueIndex - 1))
+  )
   // Wait for all queues to complete
   const itemResults = (await Promise.all(queues)) as WorkItemResult<K>[][]
   const [errors, results] = fork(

src/tests/async.test.ts

@@ -321,6 +321,19 @@ describe('async module', () => {
       })
       assert.deepEqual(Math.max(...tracking), 3)
     })
+    test('returns queue index in the async function', async () => {
+      const [errors, results] = await _.try(async () => {
+        return _.parallel(5, _.list(1, 5), async (num, index) => {
+          await _.sleep(1000)
+          return `q_${index}`
+        })
+      })()
+      assert.isUndefined(errors)
+      assert.deepEqual(
+        new Set(results),
+        new Set(['q_0', 'q_1', 'q_2', 'q_3', 'q_4'])
+      )
+    })
   })
 
   describe('_.retry', () => {