feat(iot-actions): Support to send message to IoT Events

Author: yamatatsu

Diff for: packages/@aws-cdk/aws-iot-actions/README.md

@@ -31,6 +31,7 @@ Currently supported are:
 - Put records to Kinesis Data Firehose stream
 - Send messages to SQS queues
 - Publish messages on SNS topics
+- Put messages IoT Events input
 
 ## Republish a message to another MQTT topic
 
@@ -278,3 +279,26 @@ const topicRule = new iot.TopicRule(this, 'TopicRule', {
   ],
 });
 ```
+
+## Put messages IoT Events input
+
+The code snippet below creates an AWS IoT Rule that put messages
+to an IoT Events input when it is triggered:
+
+```ts
+import * as iotevents from '@aws-cdk/aws-iotevents';
+const input = new iotevents.Input(this, 'MyInput', {
+  attributeJsonPaths: ['payload.temperature', 'payload.transactionId'],
+});
+const topicRule = new iot.TopicRule(this, 'TopicRule', {
+  sql: iot.IotSql.fromStringAsVer20160323(
+    "SELECT * FROM 'device/+/data'",
+  ),
+  actions: [
+    new actions.IotEventsPutMessageAction(input, {
+      batchMode: true, // optional property, default is 'false'
+      messageId: '${payload.transactionId}', // optional property, default is a new UUID
+    }),
+  ],
+});
+```

Diff for: packages/@aws-cdk/aws-iot-actions/lib/index.ts

@@ -3,6 +3,7 @@ export * from './cloudwatch-put-metric-action';
 export * from './cloudwatch-set-alarm-state-action';
 export * from './common-action-props';
 export * from './firehose-put-record-action';
+export * from './iotevents-put-message-action';
 export * from './iot-republish-action';
 export * from './kinesis-put-record-action';
 export * from './lambda-function-action';

Diff for: packages/@aws-cdk/aws-iot-actions/lib/iotevents-put-message-action.ts

@@ -0,0 +1,72 @@
+import * as iam from '@aws-cdk/aws-iam';
+import * as iot from '@aws-cdk/aws-iot';
+import * as iotevents from '@aws-cdk/aws-iotevents';
+import { CommonActionProps } from './common-action-props';
+import { singletonActionRole } from './private/role';
+
+/**
+ * Configuration properties of an action for the IoT Events.
+ */
+export interface IotEventsPutMessageActionProps extends CommonActionProps {
+  /**
+   * Whether to process the event actions as a batch.
+   *
+   * When batchMode is true, you can't specify a messageId.
+   *
+   * When batchMode is true and the rule SQL statement evaluates to an Array,
+   * each Array element is treated as a separate message when Events by calling BatchPutMessage.
+   * The resulting array can't have more than 10 messages.
+   *
+   * @default false
+   */
+  readonly batchMode?: boolean;
+
+  /**
+   * The ID of the message.
+   *
+   * When batchMode is true, you can't specify a messageId--a new UUID value will be assigned.
+   * Assign a value to this property to ensure that only one input (message) with a given messageId will be processed by an AWS IoT Events detector.
+   *
+   * @default - none -- a new UUID value will be assigned
+   */
+  readonly messageId?: string;
+}
+
+/**
+ * The action to put the message from an MQTT message to the IoT Events input.
+ */
+export class IotEventsPutMessageAction implements iot.IAction {
+  private readonly batchMode?: boolean;
+  private readonly messageId?: string;
+  private readonly role?: iam.IRole;
+
+  /**
+   * @param input The IoT Events input to put messages.
+   * @param props Optional properties to not use default
+   */
+  constructor(private readonly input: iotevents.IInput, props: IotEventsPutMessageActionProps = {}) {
+    this.batchMode = props.batchMode;
+    this.messageId = props.messageId;
+    this.role = props.role;
+
+    if (this.batchMode && this.messageId) {
+      throw new Error('messageId is not allowed when batchMode is true');
+    }
+  }
+
+  bind(rule: iot.ITopicRule): iot.ActionConfig {
+    const role = this.role ?? singletonActionRole(rule);
+    this.input.grantWrite(role);
+
+    return {
+      configuration: {
+        iotEvents: {
+          batchMode: this.batchMode,
+          inputName: this.input.inputName,
+          messageId: this.messageId,
+          roleArn: role.roleArn,
+        },
+      },
+    };
+  }
+}

Diff for: packages/@aws-cdk/aws-iot-actions/package.json

@@ -81,6 +81,7 @@
     "@aws-cdk/aws-kinesisfirehose-destinations": "0.0.0",
     "@aws-cdk/cdk-build-tools": "0.0.0",
     "@aws-cdk/integ-runner": "0.0.0",
+    "@aws-cdk/integ-tests": "0.0.0",
     "@aws-cdk/pkglint": "0.0.0",
     "@types/jest": "^27.5.2",
     "jest": "^27.5.1",
@@ -90,6 +91,7 @@
     "@aws-cdk/aws-cloudwatch": "0.0.0",
     "@aws-cdk/aws-iam": "0.0.0",
     "@aws-cdk/aws-iot": "0.0.0",
+    "@aws-cdk/aws-iotevents": "0.0.0",
     "@aws-cdk/aws-kinesis": "0.0.0",
     "@aws-cdk/aws-kinesisfirehose": "0.0.0",
     "@aws-cdk/aws-lambda": "0.0.0",
@@ -106,6 +108,7 @@
     "@aws-cdk/aws-cloudwatch": "0.0.0",
     "@aws-cdk/aws-iam": "0.0.0",
     "@aws-cdk/aws-iot": "0.0.0",
+    "@aws-cdk/aws-iotevents": "0.0.0",
     "@aws-cdk/aws-kinesis": "0.0.0",
     "@aws-cdk/aws-kinesisfirehose": "0.0.0",
     "@aws-cdk/aws-lambda": "0.0.0",

Diff for: packages/@aws-cdk/aws-iot-actions/test/iot/integ.iotevents-put-message-action.ts

@@ -0,0 +1,53 @@
+/**
+ * Stack verification steps:
+ * * aws iot-data publish --topic device/mydevice/data --qos 1 --payload (echo '[{"payload":{"deviceId":"001"}},{"payload":{"deviceId":"002"}}]' | base64) --region us-east-1
+ */
+import * as iot from '@aws-cdk/aws-iot';
+import * as iotevents from '@aws-cdk/aws-iotevents';
+import * as logs from '@aws-cdk/aws-logs';
+import * as cdk from '@aws-cdk/core';
+import { IntegTest } from '@aws-cdk/integ-tests';
+import * as actions from '../../lib';
+
+class TestStack extends cdk.Stack {
+  public readonly detectorModelName: string;
+
+  constructor(scope: cdk.App, id: string, props?: cdk.StackProps) {
+    super(scope, id, props);
+
+    const logGroup = new logs.LogGroup(this, 'logs', { removalPolicy: cdk.RemovalPolicy.DESTROY });
+    const topicRule = new iot.TopicRule(this, 'TopicRule', {
+      sql: iot.IotSql.fromStringAsVer20160323(
+        "SELECT * FROM 'device/+/data'",
+      ),
+      errorAction: new actions.CloudWatchLogsAction(logGroup),
+    });
+
+    const input = new iotevents.Input(this, 'MyInput', {
+      attributeJsonPaths: ['payload.deviceId'],
+    });
+
+    const detectorModel = new iotevents.DetectorModel(this, 'MyDetectorModel', {
+      detectorKey: 'payload.deviceId',
+      initialState: new iotevents.State({
+        stateName: 'initialState',
+        onEnter: [{
+          eventName: 'enter',
+          condition: iotevents.Expression.currentInput(input),
+        }],
+      }),
+    });
+
+    topicRule.addAction(
+      new actions.IotEventsPutMessageAction(input, {
+        batchMode: true,
+      }),
+    );
+
+    this.detectorModelName = detectorModel.detectorModelName;
+  }
+}
+
+const app = new cdk.App();
+const stack = new TestStack(app, 'iotevents-put-message-action-test-stack');
+new IntegTest(app, 'iotevents', { testCases: [stack] });

Diff for: packages/@aws-cdk/aws-iot-actions/test/iot/iotevents-put-message-action.integ.snapshot/cdk.out

@@ -0,0 +1 @@
+{"version":"20.0.0"}

Diff for: packages/@aws-cdk/aws-iot-actions/test/iot/iotevents-put-message-action.integ.snapshot/integ.json

@@ -0,0 +1,11 @@
+{
+  "version": "20.0.0",
+  "testCases": {
+    "iotevents/DefaultTest": {
+      "stacks": [
+        "iotevents-put-message-action-test-stack"
+      ],
+      "assertionStack": "ioteventsDefaultTestDeployAssertE216288D"
+    }
+  }
+}