Feature/map fields ItemSelector ItemProcessor MaxConcurrencyPath

__tests__/stateActions/MapStateAction.test.ts

@@ -88,6 +88,44 @@ describe('Map State', () => {
     ]);
   });
 
+  test('should execute action if using `ItemProcessor` field, which is equivalent to `Iterator`', async () => {
+    const definition: MapState = {
+      Type: 'Map',
+      ItemProcessor: {
+        StartAt: 'EntryIterationState',
+        States: {
+          EntryIterationState: {
+            Type: 'Succeed',
+          },
+        },
+      },
+      End: true,
+    };
+    const stateName = 'MapState';
+    const input = [
+      { num1: 5, num2: 3 },
+      { num1: 2, num2: 6 },
+      { num1: 7, num2: 4 },
+    ];
+    const context = {};
+    const options = {
+      stateMachineOptions: undefined,
+      runOptions: undefined,
+      eventLogger: new EventLogger(),
+      rawInput: {},
+    };
+
+    const mapStateAction = new MapStateAction(definition, stateName);
+    const { stateResult } = await mapStateAction.execute(input, context, options);
+
+    expect(stateResult).toHaveLength(3);
+    expect(stateResult).toEqual([
+      { num1: 5, num2: 3 },
+      { num1: 2, num2: 6 },
+      { num1: 7, num2: 4 },
+    ]);
+  });
+
   test('should process `Parameters` field if specified', async () => {
     const definition: MapState = {
       Type: 'Map',
@@ -142,6 +180,60 @@ describe('Map State', () => {
     ]);
   });
 
+  test('should process `ItemSelector` field if specified', async () => {
+    const definition: MapState = {
+      Type: 'Map',
+      Iterator: {
+        StartAt: 'EntryIterationState',
+        States: {
+          EntryIterationState: {
+            Type: 'Succeed',
+          },
+        },
+      },
+      ItemsPath: '$.items',
+      ItemSelector: {
+        'pair.$': '$$.Map.Item.Value',
+        'index.$': '$$.Map.Item.Index',
+      },
+      End: true,
+    };
+    const stateName = 'MapState';
+    const input = {
+      items: [
+        { num1: 5, num2: 3 },
+        { num1: 2, num2: 6 },
+        { num1: 7, num2: 4 },
+      ],
+    };
+    const context = {};
+    const options = {
+      stateMachineOptions: undefined,
+      runOptions: undefined,
+      eventLogger: new EventLogger(),
+      rawInput: {},
+    };
+
+    const mapStateAction = new MapStateAction(definition, stateName);
+    const { stateResult } = await mapStateAction.execute(input, context, options);
+
+    expect(stateResult).toHaveLength(3);
+    expect(stateResult).toEqual([
+      {
+        pair: { num1: 5, num2: 3 },
+        index: 0,
+      },
+      {
+        pair: { num1: 2, num2: 6 },
+        index: 1,
+      },
+      {
+        pair: { num1: 7, num2: 4 },
+        index: 2,
+      },
+    ]);
+  });
+
   test('should throw a runtime error if input is not an array', async () => {
     const definition: MapState = {
       Type: 'Map',

docs/feature-support.md

@@ -44,8 +44,12 @@ The following features all come from the [Amazon States Language specification](
     - [x] `Branches`
   - [x] Map
     - [x] `Iterator`
+    - [x] `ItemProcessor`
+    - [x] `Parameters`
+    - [x] `ItemSelector`
     - [x] `ItemsPath`
     - [x] `MaxConcurrency`
+    - [x] `MaxConcurrencyPath`
   - [x] Fail
     - [x] `Error`
     - [x] `Cause`
@@ -113,12 +117,9 @@ The following features all come from the [Amazon States Language specification](
     - [ ] `Credentials`
   - Map
     - [ ] Distributed mode
-    - [ ] `ItemProcessor`
     - [ ] `ItemReader`
-    - [ ] `ItemSelector`
     - [ ] `ItemBatcher`
     - [ ] `ResultWriter`
-    - [ ] `MaxConcurrencyPath`
     - [ ] `ToleratedFailurePercentage`
     - [ ] `ToleratedFailureCount`
 

src/stateMachine/stateActions/MapStateAction.ts

@@ -11,6 +11,7 @@ import { processPayloadTemplate } from '../InputOutputProcessing';
 import { StatesRuntimeError } from '../../error/predefined/StatesRuntimeError';
 import { ExecutionError } from '../../error/ExecutionError';
 import { ArrayConstraint } from '../jsonPath/constraints/ArrayConstraint';
+import { IntegerConstraint } from '../jsonPath/constraints/IntegerConstraint';
 import pLimit from 'p-limit';
 
 /**
@@ -59,8 +60,9 @@ class MapStateAction extends BaseStateAction<MapState> {
     };
 
     // Handle `Parameters` field if specified
-    if (state.Parameters) {
-      paramValue = processPayloadTemplate(state.Parameters, input, context);
+    if (state.Parameters || state.ItemSelector) {
+      const parameters = (state.ItemSelector ?? state.Parameters)!;
+      paramValue = processPayloadTemplate(parameters, input, context);
     }
 
     // Pass the current parameter value if defined, otherwise pass the current item being iterated
@@ -87,11 +89,21 @@ class MapStateAction extends BaseStateAction<MapState> {
       throw new StatesRuntimeError('Input of Map state must be an array or ItemsPath property must point to an array');
     }
 
-    const iteratorStateMachine = new StateMachine(state.Iterator, {
+    const iteratorDefinition = (state.ItemProcessor ?? state.Iterator)!;
+    const iteratorStateMachine = new StateMachine(iteratorDefinition, {
       ...options.stateMachineOptions,
       validationOptions: { noValidate: true },
     });
-    const limit = pLimit(state.MaxConcurrency || DEFAULT_MAX_CONCURRENCY);
+
+    let maxConcurrency = state.MaxConcurrency;
+    if (state.MaxConcurrencyPath) {
+      maxConcurrency = jsonPathQuery(state.MaxConcurrencyPath, input, context, {
+        constraints: [IntegerConstraint],
+        ignoreDefinedValueConstraint: true,
+      });
+    }
+
+    const limit = pLimit(maxConcurrency || DEFAULT_MAX_CONCURRENCY);
     const processedItemsPromise = items.map((item, i) =>
       limit(() => this.processItem(iteratorStateMachine, item, input, context, i, options))
     );

src/typings/MapState.ts

@@ -3,9 +3,9 @@ import type { CatchableState, RetryableState } from './ErrorHandling';
 import type {
   CanHaveInputPath,
   CanHaveOutputPath,
-  CanHaveParameters,
   CanHaveResultPath,
   CanHaveResultSelector,
+  PayloadTemplate,
 } from './InputOutputProcessing';
 import type { IntermediateState } from './IntermediateState';
 import type { StateMachineDefinition } from './StateMachineDefinition';
@@ -14,16 +14,19 @@ import type { TerminalState } from './TerminalState';
 interface BaseMapState
   extends BaseState,
     CanHaveInputPath,
-    CanHaveParameters,
     CanHaveResultSelector,
     CanHaveResultPath,
     CanHaveOutputPath,
     RetryableState,
     CatchableState {
   Type: 'Map';
-  Iterator: Omit<StateMachineDefinition, 'Version' | 'TimeoutSeconds'>;
+  Iterator?: Omit<StateMachineDefinition, 'Version' | 'TimeoutSeconds'>; // deprecated but still supported, superseded by `ItemProcessor`
+  ItemProcessor?: Omit<StateMachineDefinition, 'Version' | 'TimeoutSeconds'>;
+  Parameters?: PayloadTemplate; // deprecated but still supported, superseded by `ItemSelector`
+  ItemSelector?: PayloadTemplate;
   ItemsPath?: string;
   MaxConcurrency?: number;
+  MaxConcurrencyPath?: string;
 }
 
 export type MapState = (IntermediateState | TerminalState) & BaseMapState;