fix(stepfunctions): improve Task payload encoding (#2706)

Improve referencing data fields for StepFunctions tasks, in preparation
of callback task implementaion.

Get rid of `JsonPath`, and in its place we have 2 new classes:

- `Data`, for fields that come from the user payload
  (`$.My.Field`). Settle on the term "data" since that's the
  term used in most of StepFunctions' docs.
- `Context`, for fields that come from the service-defined
  task "context" (like `$$.Execution.StartTime`, and in particular
  `$$.Task.Token`).

These classes have been moved from the `-tasks` module to the
`aws-stepfunctions` module, where it seems to make more sense for them
to live.

Add support for SQS and SNS tasks to publish an arbitrary JSON
structure that can reference fields from context and execution data.

Remove `NumberValue` since we can now encode Tokens in regular number
values.

BREAKING CHANGES:

- **stepfunctions**: `JsonPath.stringFromPath` (and others) are now called `Data.stringAt()`.
  The `DataField` class now lives in the main stepfunctions module.
- **stepfunctions**: `PublishToTopic` property `messageObject` used to take a JSON string, now pass `sfn.TaskInput.fromObject()` or `sfn.TaskInput.fromText()` into the `message` field.
- **stepfunctions**: `SendToQueue` property `messageBody` used to take a JSON string, now pass `sfn.TaskInput.fromObject()` or `sfn.TaskInput.fromText()` into the `message` field.
- **stepfunctions**: Instead of passing `NumberValue`s to StepFunctions tasks, pass regular
  numbers.

Author: rix0rrr

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

@@ -5,6 +5,4 @@ export * from './run-ecs-task-base-types';
 export * from './publish-to-topic';
 export * from './send-to-queue';
 export * from './run-ecs-ec2-task';
-export * from './run-ecs-fargate-task';
-export * from './number-value';
-export * from './json-path';
+export * from './run-ecs-fargate-task';

packages/@aws-cdk/aws-stepfunctions-tasks/lib/json-path.ts

@@ -1,26 +1,15 @@
 import iam = require('@aws-cdk/aws-iam');
 import sns = require('@aws-cdk/aws-sns');
 import sfn = require('@aws-cdk/aws-stepfunctions');
-import cdk = require('@aws-cdk/cdk');
-import { renderString } from './json-path';
 
 /**
  * Properties for PublishTask
  */
 export interface PublishToTopicProps {
   /**
-   * The text message to send to the queue.
-   *
-   * Exactly one of `message` and `messageObject` is required.
-   */
-  readonly message?: string;
-
-  /**
-   * Object to be JSON-encoded and used as message
-   *
-   * Exactly one of `message`, `messageObject` and `messagePath` is required.
+   * The text message to send to the topic.
    */
-  readonly messageObject?: string;
+  readonly message: sfn.TaskInput;
 
   /**
    * If true, send a different message to every subscription type
@@ -48,12 +37,9 @@ export interface PublishToTopicProps {
  */
 export class PublishToTopic implements sfn.IStepFunctionsTask {
   constructor(private readonly topic: sns.ITopic, private readonly props: PublishToTopicProps) {
-    if ((props.message === undefined) === (props.messageObject === undefined)) {
-      throw new Error(`Supply exactly one of 'message' or 'messageObject'`);
-    }
   }
 
-  public bind(task: sfn.Task): sfn.StepFunctionsTaskProperties {
+  public bind(_task: sfn.Task): sfn.StepFunctionsTaskProperties {
     return {
       resourceArn: 'arn:aws:states:::sns:publish',
       policyStatements: [new iam.PolicyStatement()
@@ -62,11 +48,11 @@ export class PublishToTopic implements sfn.IStepFunctionsTask {
       ],
       parameters: {
         TopicArn: this.topic.topicArn,
-        ...(this.props.messageObject
-            ? { Message: new cdk.Token(() => task.node.stringifyJson(this.props.messageObject)) }
-            : renderString('Message', this.props.message)),
-        MessageStructure: this.props.messagePerSubscriptionType ? "json" : undefined,
-        ...renderString('Subject', this.props.subject),
+        ...sfn.FieldUtils.renderObject({
+          Message: this.props.message.value,
+          MessageStructure: this.props.messagePerSubscriptionType ? "json" : undefined,
+          Subject: this.props.subject,
+        })
       }
     };
   }

packages/@aws-cdk/aws-stepfunctions-tasks/lib/number-value.ts

@@ -1,5 +1,3 @@
-import { NumberValue } from "./number-value";
-
 export interface ContainerOverride {
   /**
    * Name of the container inside the task definition
@@ -23,21 +21,21 @@ export interface ContainerOverride {
    *
    * @Default The default value from the task definition.
    */
-  readonly cpu?: NumberValue;
+  readonly cpu?: number;
 
   /**
    * Hard memory limit on the container
    *
    * @Default The default value from the task definition.
    */
-  readonly memoryLimit?: NumberValue;
+  readonly memoryLimit?: number;
 
   /**
    * Soft memory limit on the container
    *
    * @Default The default value from the task definition.
    */
-  readonly memoryReservation?: NumberValue;
+  readonly memoryReservation?: number;
 }
 
 /**

packages/@aws-cdk/aws-stepfunctions-tasks/lib/publish-to-topic.ts

@@ -3,7 +3,6 @@ import ecs = require('@aws-cdk/aws-ecs');
 import iam = require('@aws-cdk/aws-iam');
 import sfn = require('@aws-cdk/aws-stepfunctions');
 import cdk = require('@aws-cdk/cdk');
-import { renderNumber, renderString, renderStringList } from './json-path';
 import { ContainerOverride } from './run-ecs-task-base-types';
 
 /**
@@ -162,17 +161,17 @@ function renderOverrides(containerOverrides?: ContainerOverride[]) {
 
   const ret = new Array<any>();
   for (const override of containerOverrides) {
-    ret.push({
-      ...renderString('Name', override.containerName),
-      ...renderStringList('Command', override.command),
-      ...renderNumber('Cpu', override.cpu),
-      ...renderNumber('Memory', override.memoryLimit),
-      ...renderNumber('MemoryReservation', override.memoryReservation),
+    ret.push(sfn.FieldUtils.renderObject({
+      Name: override.containerName,
+      Command: override.command,
+      Cpu: override.cpu,
+      Memory: override.memoryLimit,
+      MemoryReservation: override.memoryReservation,
       Environment: override.environment && override.environment.map(e => ({
-        ...renderString('Name', e.name),
-        ...renderString('Value', e.value),
+        Name: e.name,
+        Value: e.value,
       }))
-    });
+    }));
   }
 
   return { ContainerOverrides: ret };

packages/@aws-cdk/aws-stepfunctions-tasks/lib/run-ecs-task-base-types.ts

@@ -1,17 +1,15 @@
 import iam = require('@aws-cdk/aws-iam');
 import sqs = require('@aws-cdk/aws-sqs');
 import sfn = require('@aws-cdk/aws-stepfunctions');
-import { renderNumber, renderString } from './json-path';
-import { NumberValue } from './number-value';
 
 /**
  * Properties for SendMessageTask
  */
 export interface SendToQueueProps {
   /**
-   * The message body to send to the queue.
+   * The text message to send to the queue.
    */
-  readonly messageBody: string;
+  readonly messageBody: sfn.TaskInput;
 
   /**
    * The length of time, in seconds, for which to delay a specific message.
@@ -20,7 +18,7 @@ export interface SendToQueueProps {
    *
    * @default Default value of the queue is used
    */
-  readonly delaySeconds?: NumberValue;
+  readonly delaySeconds?: number;
 
   /**
    * The token used for deduplication of sent messages.
@@ -59,10 +57,12 @@ export class SendToQueue implements sfn.IStepFunctionsTask {
       ],
       parameters: {
         QueueUrl: this.queue.queueUrl,
-        ...renderString('MessageBody', this.props.messageBody),
-        ...renderNumber('DelaySeconds', this.props.delaySeconds),
-        ...renderString('MessageDeduplicationId', this.props.messageDeduplicationId),
-        ...renderString('MessageGroupId', this.props.messageGroupId),
+        ...sfn.FieldUtils.renderObject({
+          MessageBody: this.props.messageBody.value,
+          DelaySeconds: this.props.delaySeconds,
+          MessageDeduplicationId: this.props.messageDeduplicationId,
+          MessageGroupId: this.props.messageGroupId,
+        })
       }
     };
   }

packages/@aws-cdk/aws-stepfunctions-tasks/lib/run-ecs-task-base.ts

@@ -4,7 +4,6 @@ import ecs = require('@aws-cdk/aws-ecs');
 import sfn = require('@aws-cdk/aws-stepfunctions');
 import { Stack } from '@aws-cdk/cdk';
 import tasks = require('../lib');
-import { JsonPath, NumberValue } from '../lib';
 
 let stack: Stack;
 let vpc: ec2.Vpc;
@@ -64,7 +63,7 @@ test('Running a Fargate Task', () => {
       {
         containerName: 'TheContainer',
         environment: [
-          {name: 'SOME_KEY', value: JsonPath.stringFromPath('$.SomeKey')}
+          {name: 'SOME_KEY', value: sfn.Data.stringAt('$.SomeKey')}
         ]
       }
     ]
@@ -162,7 +161,7 @@ test('Running an EC2 Task with bridge network', () => {
       {
         containerName: 'TheContainer',
         environment: [
-          {name: 'SOME_KEY', value: JsonPath.stringFromPath('$.SomeKey')}
+          {name: 'SOME_KEY', value: sfn.Data.stringAt('$.SomeKey')}
         ]
       }
     ]
@@ -296,9 +295,9 @@ test('Running an EC2 Task with overridden number values', () => {
     containerOverrides: [
       {
         containerName: 'TheContainer',
-        command: JsonPath.listFromPath('$.TheCommand'),
-        cpu: NumberValue.fromNumber(5),
-        memoryLimit: JsonPath.numberFromPath('$.MemoryLimit'),
+        command: sfn.Data.listAt('$.TheCommand'),
+        cpu: 5,
+        memoryLimit: sfn.Data.numberAt('$.MemoryLimit'),
       }
     ]
   });