AWS IOT 规则的行动
本单元中更高层次的构件的API是实验性的并正在积极开发中. 在今后任何版本中,它们都须作非背向相容的修改或去除。 这些是 不适用 语义版本 模型和断开的更改 在发行笔记中宣布. 这意味着虽然你可能使用它们,但可能需要更新 当更新到此软件包的更新版本时,您的源代码.
此库包含集成类,以发送数据到任意数量
支持AWS服务. 这些班级的情况应传至
TopicRule 定义 aws-cdk-lib/aws-iot。 。 。 .
目前得到支持的有:
- 向另一个 MQTT 主题重现消息
- 启动 Lambda 函数
- 将对象放入 S3 桶
- 将日志放入云表日志
- 捕捉云表测量标准
- 更改云表提醒的状态
- 将记录放入 Kinesis 数据流
- 将记录放入 Amazon 数据 Firehose 流
- 向 SQS 队列发送信件
- 发布关于 SNS 主题的信息
- 将信件写入 DynamoDB 的列中
- 插入消息 IOT 事件输入
- 向 HTTPS 端点发送信件
向另一个 MQTT 主题重现消息
下面的代码片段创建了 AWS IOT 规则,将信件重印到 当 MQTT 主题被触发时 .
new iot.TopicRule(this, 'TopicRule', {
sql: iot.IotSql.fromStringAsVer20160323("SELECT topic(2) as device_id, timestamp() as timestamp, temperature FROM 'device/+/data'"),
actions: [
new actions.IotRepublishMqttAction('${topic()}/republish', {
qualityOfService: actions.MqttQualityOfService.AT_LEAST_ONCE, // optional property, default is MqttQualityOfService.ZERO_OR_MORE_TIMES
}),
],
});
启动 Lambda 函数
下面的代码片段创建了 AWS IOT 规则,它引用了 Lambda 函数 当它被触发.
const func = new lambda.Function(this, 'MyFunction', {
runtime: lambda.Runtime.NODEJS_LATEST,
handler: 'index.handler',
code: lambda.Code.fromInline(`
exports.handler = (event) => {
console.log("It is test for lambda action of AWS IoT Rule.", event);
};`
),
});
new iot.TopicRule(this, 'TopicRule', {
sql: iot.IotSql.fromStringAsVer20160323("SELECT topic(2) as device_id, timestamp() as timestamp, temperature FROM 'device/+/data'"),
actions: [new actions.LambdaFunctionAction(func)],
});
将对象放入 S3 桶
下面的代码片段创建 AWS IOT 规则,将对象放入 S3 桶 当它被触发.
const bucket = new s3.Bucket(this, 'MyBucket');
new iot.TopicRule(this, 'TopicRule', {
sql: iot.IotSql.fromStringAsVer20160323("SELECT topic(2) as device_id FROM 'device/+/data'"),
actions: [new actions.S3PutObjectAction(bucket)],
});
财产 key 联 合 国 S3PutObjectAction 给出值 ${topic()}/${timestamp()} 默认。 这个 ${topic()}
财务报告和审定财务报表 ${timestamp()} 称为替代模板。 更多信息见
此文档。 。 。 .
在以上样本中, ${topic()} 被给定的 MQTT 主题替换为 device/001/data还有 ${timestamp()} 替换
以毫秒为单位的当前时间戳数 1636289461203因此,如果 MQTT 经纪人收到 MQTT 话题
device/001/data 打开 2021-11-07T00:00:00.000ZS3桶对象将被放置到 device/001/data/1636243200000。 。 。 .
您也可以设置特定 key 现改为:
const bucket = new s3.Bucket(this, 'MyBucket');
new iot.TopicRule(this, 'TopicRule', {
sql: iot.IotSql.fromStringAsVer20160323(
"SELECT topic(2) as device_id, year, month, day FROM 'device/+/data'",
),
actions: [
new actions.S3PutObjectAction(bucket, {
key: '${year}/${month}/${day}/${topic(2)}',
}),
],
});
如果您想要设置 S3 桶对象的访问控制, 您可以指定 accessControl 现改为:
const bucket = new s3.Bucket(this, 'MyBucket');
new iot.TopicRule(this, 'TopicRule', {
sql: iot.IotSql.fromStringAsVer20160323("SELECT * FROM 'device/+/data'"),
actions: [
new actions.S3PutObjectAction(bucket, {
accessControl: s3.BucketAccessControl.PUBLIC_READ,
}),
],
});
将日志放入云表日志
下面的代码片段创建了 AWS IOT 规则,将日志放入 Cloud Watch Logs 当它被触发.
import * as logs from 'aws-cdk-lib/aws-logs';
const logGroup = new logs.LogGroup(this, 'MyLogGroup');
new iot.TopicRule(this, 'TopicRule', {
sql: iot.IotSql.fromStringAsVer20160323("SELECT topic(2) as device_id FROM 'device/+/data'"),
actions: [new actions.CloudWatchLogsAction(logGroup)],
});
捕捉云表测量标准
下面的代码片段创建了 AWS IOT 规则,用于捕捉 Cloud Watch 度量衡 当它被触发.
const topicRule = new iot.TopicRule(this, 'TopicRule', {
sql: iot.IotSql.fromStringAsVer20160323(
"SELECT topic(2) as device_id, namespace, unit, value, timestamp FROM 'device/+/data'",
),
actions: [
new actions.CloudWatchPutMetricAction({
metricName: '${topic(2)}',
metricNamespace: '${namespace}',
metricUnit: '${unit}',
metricValue: '${value}',
metricTimestamp: '${timestamp}',
}),
],
});
启动 Step 函数状态机器
下面的代码片段创建了 AWS IOT 规则,启动 Step 函数状态机器 当它被触发.
const stateMachine = new stepfunctions.StateMachine(this, 'SM', {
definitionBody: stepfunctions.DefinitionBody.fromChainable(new stepfunctions.Wait(this, 'Hello', { time: stepfunctions.WaitTime.duration(Duration.seconds(10)) })),
});
new iot.TopicRule(this, 'TopicRule', {
sql: iot.IotSql.fromStringAsVer20160323("SELECT * FROM 'device/+/data'"),
actions: [
new actions.StepFunctionsStateMachineAction(stateMachine),
],
});
更改亚马逊云表提醒的状态
下面的代码片段创建了 AWS IOT 规则, 当亚马逊云表提醒被触发时会改变其状态 :
import * as cloudwatch from 'aws-cdk-lib/aws-cloudwatch';
const metric = new cloudwatch.Metric({
namespace: 'MyNamespace',
metricName: 'MyMetric',
dimensionsMap: { MyDimension: 'MyDimensionValue' },
});
const alarm = new cloudwatch.Alarm(this, 'MyAlarm', {
metric: metric,
threshold: 100,
evaluationPeriods: 3,
datapointsToAlarm: 2,
});
const topicRule = new iot.TopicRule(this, 'TopicRule', {
sql: iot.IotSql.fromStringAsVer20160323("SELECT topic(2) as device_id FROM 'device/+/data'"),
actions: [
new actions.CloudWatchSetAlarmStateAction(alarm, {
reason: 'AWS Iot Rule action is triggered',
alarmStateToSet: cloudwatch.AlarmState.ALARM,
}),
],
});
将记录放入 Kinesis 数据流
下面的代码片段创建了 AWS IOT 规则,将记录放入 Kinesis 数据 当它被触发时会流出.
import * as kinesis from 'aws-cdk-lib/aws-kinesis';
const stream = new kinesis.Stream(this, 'MyStream');
const topicRule = new iot.TopicRule(this, 'TopicRule', {
sql: iot.IotSql.fromStringAsVer20160323("SELECT * FROM 'device/+/data'"),
actions: [
new actions.KinesisPutRecordAction(stream, {
partitionKey: '${newuuid()}',
}),
],
});
将记录放入 Amazon 数据 Firehose 流
下面的代码片段创建 AWS IOT 规则,将记录放入记录 当它被触发时,它会变成亚马逊数据Firehose流.
import * as firehose from 'aws-cdk-lib/aws-kinesisfirehose';
const bucket = new s3.Bucket(this, 'MyBucket');
const stream = new firehose.DeliveryStream(this, 'MyStream', {
destination: new firehose.S3Bucket(bucket),
});
const topicRule = new iot.TopicRule(this, 'TopicRule', {
sql: iot.IotSql.fromStringAsVer20160323("SELECT * FROM 'device/+/data'"),
actions: [
new actions.FirehosePutRecordAction(stream, {
batchMode: true,
recordSeparator: actions.FirehoseRecordSeparator.NEWLINE,
}),
],
});
将信件发送到 SQS 队列
下面的代码片段创建了发送消息的 AWS IOT 规则 当SQS队列被触发时 :
import * as sqs from 'aws-cdk-lib/aws-sqs';
const queue = new sqs.Queue(this, 'MyQueue');
const topicRule = new iot.TopicRule(this, 'TopicRule', {
sql: iot.IotSql.fromStringAsVer20160323(
"SELECT topic(2) as device_id, year, month, day FROM 'device/+/data'",
),
actions: [
new actions.SqsQueueAction(queue, {
useBase64: true, // optional property, default is 'false'
}),
],
});
就 SNS 专题发布信件
下面的代码片段创建和AWS IOT规则,当SNS主题被触发时发布消息:
import * as sns from 'aws-cdk-lib/aws-sns';
const topic = new sns.Topic(this, 'MyTopic');
const topicRule = new iot.TopicRule(this, 'TopicRule', {
sql: iot.IotSql.fromStringAsVer20160323(
"SELECT topic(2) as device_id, year, month, day FROM 'device/+/data'",
),
actions: [
new actions.SnsTopicAction(topic, {
messageFormat: actions.SnsActionMessageFormat.JSON, // optional property, default is SnsActionMessageFormat.RAW
}),
],
});
将信件的属性写入 DynamoDB
下面的代码片段创建了 AWS IOT 规则,用于写入全部或部分 使用 DynamoDBv2 动作向 DynamoDB 发送 MQTT 消息.
import * as dynamodb from 'aws-cdk-lib/aws-dynamodb';
declare const table: dynamodb.Table;
const topicRule = new iot.TopicRule(this, 'TopicRule', {
sql: iot.IotSql.fromStringAsVer20160323(
"SELECT * FROM 'device/+/data'",
),
actions: [
new actions.DynamoDBv2PutItemAction(table)
],
});
插入消息 IOT 事件输入
下面的代码片段创建了 AWS IOT 规则,将消息放入 当IOT 事件被触发时, 向 IOT 事件输入:
import * as iotevents from '@aws-cdk/aws-iotevents-alpha';
import * as iam from 'aws-cdk-lib/aws-iam';
declare const role: iam.IRole;
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
role: role, // optional property, default is a new UUID
}),
],
});
向 HTTPS 端点发送信件
下面的代码片段创建了发送消息的 AWS IOT 规则 当触发时, 到 HTTPS 端点 :
const topicRule = new iot.TopicRule(this, 'TopicRule', {
sql: iot.IotSql.fromStringAsVer20160323(
"SELECT topic(2) as device_id, year, month, day FROM 'device/+/data'",
),
});
topicRule.addAction(
new actions.HttpsAction('https://example.com/endpoint', {
confirmationUrl: 'https://example.com',
headers: [
{ key: 'key0', value: 'value0' },
{ key: 'key1', value: 'value1' },
],
auth: { serviceName: 'serviceName', signingRegion: 'us-east-1' },
}),
);
可以使批量工作降低成本并提高效率:
import { Size } from 'aws-cdk-lib';
declare const topicRule: iot.TopicRule;
topicRule.addAction(
new actions.HttpsAction('https://example.com/endpoint', {
batchConfig: {
maxBatchOpenDuration: Duration.millis(100),
maxBatchSize: 5,
maxBatchSizeBytes: Size.kibibytes(1),
},
}),
);
有关批量配置的更多信息,请参见: AWS IOT 核心文件。 。 。 .
将数据写入打开搜索服务
下面的代码片段创建 AWS IOT 规则来写入数据 打开搜索服务,当它被触发时:
import * as opensearch from 'aws-cdk-lib/aws-opensearchservice';
declare const domain: opensearch.Domain;
const topicRule = new iot.TopicRule(this, 'TopicRule', {
sql: iot.IotSql.fromStringAsVer20160323(
"SELECT topic(2) as device_id, year, month, day FROM 'device/+/data'",
),
});
topicRule.addAction(new actions.OpenSearchAction(domain, {
id: 'my-id',
index: 'my-index',
type: 'my-type',
}));