WIP DO NOT MERGE StepFunctions Dynamic Parallelism

packages/@aws-cdk/aws-stepfunctions/lib/index.ts

@@ -13,6 +13,7 @@ export * from './step-functions-task';
 export * from './states/choice';
 export * from './states/fail';
 export * from './states/parallel';
+export * from './states/map';
 export * from './states/pass';
 export * from './states/state';
 export * from './states/succeed';

packages/@aws-cdk/aws-stepfunctions/lib/states/map.ts

@@ -0,0 +1,145 @@
+import cdk = require('@aws-cdk/core');
+import { Chain } from '../chain';
+import { StateGraph } from '../state-graph';
+import { CatchProps, IChainable, INextable, RetryProps } from '../types';
+import { StateType } from './private/state-type';
+import { renderJsonPath, State } from './state';
+
+/**
+ * Properties for defining a Map state
+ */
+export interface MapProps {
+    /**
+     * An optional description for this state
+     *
+     * @default No comment
+     */
+    readonly comment?: string;
+
+    /**
+     * JSONPath expression to select part of the state to be the input to this state.
+     *
+     * May also be the special value DISCARD, which will cause the effective
+     * input to be the empty object {}.
+     *
+     * @default $
+     */
+    readonly inputPath?: string;
+
+    /**
+     * JSONPath expression to select part of the state to be the output to this state.
+     *
+     * May also be the special value DISCARD, which will cause the effective
+     * output to be the empty object {}.
+     *
+     * @default $
+     */
+    readonly outputPath?: string;
+
+    /**
+     * JSONPath expression to indicate where to inject the state's output
+     *
+     * May also be the special value DISCARD, which will cause the state's
+     * input to become its output.
+     *
+     * @default $
+     */
+    readonly resultPath?: string;
+
+    /**
+     * The “MaxConcurrency” field’s value is an integer that provides an
+     * upper bound on how many invocations of the Iterator may run in parallel.
+     *
+     * @default 0
+     */
+    readonly maxConcurrency?: number;
+}
+
+/**
+ * Define a Map state in the state machine
+ *
+ * A Map state can be used to run one or more state machines at the same
+ * time.
+ *
+ * The Result of a Map state is an array of the results of its substatemachines.
+ */
+export class Map extends State implements INextable {
+    public readonly endStates: INextable[];
+
+    /**
+     * Usually, State Properties are contained in the state.ts file, but maxConcurrency
+     * only exists in this one state (for now).
+     */
+    protected readonly maxConcurrency: string;
+
+    constructor(scope: cdk.Construct, id: string, props: MapProps = {}) {
+        super(scope, id, props);
+
+        this.endStates = [this];
+    }
+
+    /**
+     * Add retry configuration for this state
+     *
+     * This controls if and how the execution will be retried if a particular
+     * error occurs.
+     */
+    public addRetry(props: RetryProps = {}): Map {
+        super._addRetry(props);
+        return this;
+    }
+
+    /**
+     * Add a recovery handler for this state
+     *
+     * When a particular error occurs, execution will continue at the error
+     * handler instead of failing the state machine execution.
+     */
+    public addCatch(handler: IChainable, props: CatchProps = {}): Map {
+        super._addCatch(handler.startState, props);
+        return this;
+    }
+
+    /**
+     * Continue normal execution with the given state
+     */
+    public next(next: IChainable): Chain {
+        super.makeNext(next.startState);
+        return Chain.sequence(this, next);
+    }
+
+    /**
+     * Define one or more graphs to run in map
+     */
+    public dynamicBranch(graph: IChainable): Map {
+        const name = `Dynamic Map '${this.stateId}'`;
+        super.addDynamicBranch(new StateGraph(graph.startState, name));
+        return this;
+    }
+
+    /**
+     * Return the Amazon States Language object for this state
+     */
+    public toStateJson(): object {
+        return {
+            Type: StateType.MAP,
+            Comment: this.comment,
+            ResultPath: renderJsonPath(this.resultPath),
+            MaxConcurrency: this.maxConcurrency,
+            ...this.renderNextEnd(),
+            ...this.renderInputOutput(),
+            ...this.renderRetryCatch(),
+            ...this.renderDynamicMap(),
+        };
+    }
+
+    /**
+     * Validate this state
+     */
+    protected validate(): string[] {
+        if (this.branches.length === 0) {
+            return ['Map must have at least one branch'];
+        }
+        return [];
+    }
+}

packages/@aws-cdk/aws-stepfunctions/lib/states/private/state-type.ts

@@ -8,5 +8,6 @@ export enum StateType {
   WAIT = 'Wait',
   SUCCEED = 'Succeed',
   FAIL = 'Fail',
-  PARALLEL = 'Parallel'
+  PARALLEL = 'Parallel',
+  MAP = 'Map'
 }

packages/@aws-cdk/aws-stepfunctions/lib/states/state.ts

@@ -126,6 +126,7 @@ export abstract class State extends cdk.Construct implements IChainable {
     protected readonly outputPath?: string;
     protected readonly resultPath?: string;
     protected readonly branches: StateGraph[] = [];
+    protected dynamicMap?: StateGraph;
     protected defaultChoice?: State;
 
     /**
@@ -273,7 +274,7 @@ export abstract class State extends cdk.Construct implements IChainable {
     }
 
     /**
-     * Add a paralle branch to this state
+     * Add a parallel branch to this state
      */
     protected addBranch(branch: StateGraph) {
         this.branches.push(branch);
@@ -282,6 +283,13 @@ export abstract class State extends cdk.Construct implements IChainable {
         }
     }
 
+    /**
+     * Add a dynamic map to this state
+     */
+    protected addDynamicBranch(graph: StateGraph) {
+        this.dynamicMap = graph;
+    }
+
     /**
      * Make the indicated state the default choice transition of this state
      */
@@ -334,6 +342,19 @@ export abstract class State extends cdk.Construct implements IChainable {
         };
     }
 
+    /**
+     * Render dynamic parallel branches in ASL JSON format
+     */
+    protected renderDynamicMap(): any {
+        if (this.dynamicMap) {
+            return {
+                Iterator: this.dynamicMap.toGraphJson()
+            };
+        } else {
+            return;
+        }
+    }
+
     /**
      * Render error recovery options in ASL JSON format
      */

packages/@aws-cdk/aws-stepfunctions/test/test.states-language.ts

@@ -368,6 +368,34 @@ export = {
                 }
             });
 
+            test.done();
+        },
+
+        'basic dynamic map'(test: Test) {
+            // GIVEN
+            const stack = new cdk.Stack();
+
+            // WHEN
+            const task1 = new stepfunctions.Map(stack, 'State One', {
+                inputPath: "$.shipped"
+            });
+
+            const task2 = new stepfunctions.Pass(stack, 'State Two');
+            const innerChain = stepfunctions.Chain.start(task2);
+
+            task1.dynamicBranch(innerChain);
+            const chain = stepfunctions.Chain
+                .start(task1);
+
+            // THEN
+            test.deepEqual(render(chain), {
+                StartAt: 'State One',
+                States: {
+                    'State One': { Type: 'Pass', Next: 'State Two' },
+                    'State Two': { Type: 'Pass', End: true },
+                }
+            });
+
             test.done();
         }
     },