Merge pull request #26 from nibble-4bits/feature/map-state-max-concurrency

Feature/map state max concurrency

Author: nibble-4bits

docs/feature-support.md

@@ -13,6 +13,10 @@
   - [x] Pass
   - [x] Wait
   - [x] Succeed
+  - [x] Map
+    - [x] `Iterator`
+    - [x] `ItemsPath`
+    - [x] `MaxConcurrency`
 
 ## Limited support
 
@@ -23,10 +27,6 @@
     - [ ] `HeartbeatSeconds`
     - [ ] `TimeoutSecondsPath`
     - [ ] `HeartbeatSecondsPath`
-  - Map
-    - [x] `Iterator`
-    - [x] `ItemsPath`
-    - [ ] `MaxConcurrency`
   - Choice
     - Boolean expressions
       - [x] `And`

package-lock.json

@@ -60,6 +60,7 @@
     "asl-validator": "^3.0.8",
     "jsonpath-plus": "^6.0.1",
     "lodash": "^4.17.21",
+    "p-limit": "^3.1.0",
     "wildcard-match": "^5.1.2"
   }
 }

package.json

@@ -16,6 +16,7 @@ import { testChoiceRule } from './ChoiceHelper';
 import aslValidator from 'asl-validator';
 import set from 'lodash/set.js';
 import cloneDeep from 'lodash/cloneDeep.js';
+import pLimit from 'p-limit';
 
 export class StateMachine {
   /**
@@ -302,14 +303,11 @@ export class StateMachine {
       return;
     }
 
-    const result = new Array(items.length);
-    let paramValue;
-    for (let i = 0; i < items.length; i++) {
-      const item = items[i];
-
+    const processItem = (item: JSONValue, index: number): Promise<JSONValue> => {
+      let paramValue;
       this.context['Map'] = {
         Item: {
-          Index: i,
+          Index: index,
           Value: item,
         },
       };
@@ -321,8 +319,13 @@ export class StateMachine {
 
       // Pass the current parameter value if defined, otherwise pass the current item being iterated
       const mapStateMachine = new StateMachine(state.Iterator, this.validationOptions);
-      result[i] = await mapStateMachine.run(paramValue ?? item, options);
-    }
+      return mapStateMachine.run(paramValue ?? item, options);
+    };
+
+    const DEFAULT_MAX_CONCURRENCY = 40; // If `MaxConcurrency` is 0 or not specified, default to running 40 iterations concurrently
+    const limit = pLimit(state.MaxConcurrency || DEFAULT_MAX_CONCURRENCY);
+    const input = items.map((item, i) => limit(() => processItem(item, i)));
+    const result = await Promise.all(input);
 
     delete this.context['Map'];
     this.currResult = result;