Lambdaで実装するIotトピックのSubscriberとPublisher

# はじめに
最近、過去に業務で設計・実装したAWS IotトピックのSubscriberとPublisherのインフラコードを整理してみました。これを機に、アーキテクチャや実装した内容について記事で紹介しようと思います。

ソースコードは下記repoをご参照ください。
※今回紹介する内容は汎用的なものであり、実際業務で実装したアーキテクチャやアプリのロジックの記載はありません。

https://github.com/to-fmak/iot-topic-subscriber-publisher-demo


# やりたいこと
- Iotデバイス等からIotトピックに送られたメッセージをsubscribeし、DynamoDBに書き込みたい
- 特定のDynamoDBにアイテムが作成or更新or削除された際に、そのイベントの詳細をIotトピックにpublishし、Iotデバイス等で受信したい


# アーキテクチャ
## Subscriber
![iot-demo-subscriber.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/2954734/ae5ae845-2815-574a-9dc2-5cf65537473e.png)

1. （Iotデバイス等から）メッセージがJSON形式でIotトピック（iot/test/subscriber）に送られる
1. subscriber-lambdaがトリガーされる
1. subscriber-lambdaがeventからメッセージを取り出し、指定したフォーマットで指定したDynamoDBテーブル（subscriber-table）にデータを書き込む

## Publisher
![iot-demo-publisher.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/2954734/3ec052c6-3e90-597b-263f-57c4ede43245.png)

1. （業務アプリ等が）指定したDynamoDBテーブル（publisher-table）にアイテムを作成（or更新or削除）する
1. DynamoDB Streamsが変更データをキャプチャし、publisher-lambdaをトリガーする
1. publisher-lambdaが、eventから変更データキャプチャを取り出し、指定したフォーマットで指定したIotトピック（iot/test/publisher）にメッセージをpublishする


# 動作確認
上記repoのTerraformコードをデプロイすれば、動作確認に必要なリソースが作成されます。動作確認の際にMQTTテストクライアントを利用します。

## Subscriber
「AWS IoT」>「MQTT テストクライアント」の管理画面を開き、下記トピック名とメッセージペイロードを入力し、「発行」をクリックします。
```:トピック名
iot/test/subscriber
```
```:メッセージペイロード
{
    "TenantID": "test01",
    "Timestamp": "2024-04-12T12:53:12.195272",
    "Message": "test"
}
```

![Screenshot 2024-04-13 at 14.32.56.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/2954734/53d6782c-fd0e-01ff-12b0-45122c310e96.png)

DynamoDBの管理画面からsubscriber-tableを開き、項目のスキャンを行うと、上記手順で発行したメッセージをもとに作成されたアイテムが表示されます。
![Screenshot 2024-04-13 at 14.35.39.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/2954734/2348b7fe-3ebf-05ed-8098-f4670c862cff.png)

## Publisher
「AWS IoT」>「MQTT テストクライアント」の管理画面を開き、下記トピック名を入力し、「サブスクライブ」をクリックします。
```:トピック名
iot/test/publisher
```
![Screenshot 2024-04-13 at 14.45.43.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/2954734/b9d15b55-f1d5-aba5-f1bf-444f73732dfb.png)

DynamoDBの管理画面からpublisher-tableを開き、下記アイテムを新規作成します。
![Screenshot 2024-04-13 at 14.46.32.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/2954734/811dbca6-d342-f6ca-fb10-98ddac455de5.png)
![Screenshot 2024-04-13 at 14.46.54.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/2954734/4874e62f-340a-247d-6bf7-37c8540f6ba1.png)


「MQTT テストクライアント」の管理画面に戻ってサブスクリプションを確認してみると、アイテムの変更データキャプチャをもとに発行されたメッセージが表示されます。
![Screenshot 2024-04-13 at 14.47.03.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/2954734/bb5f6f36-8353-d0e9-45b1-30e5c5564bbc.png)

アイテムの更新と削除もやってみます。
![Screenshot 2024-04-13 at 14.47.45.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/2954734/b1e5fbf6-b81b-f933-2840-8ece0bdea7b3.png)



# 実装時にハマっていたこと
## インフラ
Terraformを使ってLambdaのトリガーにIot ruleを追加する際に、「aws_iot_topic_rule」のリソースブロックで「function_arn」を指定しただけでは、トリガーがうまく設定されませんでした。「aws_lambda_permission」のリソースも記述する必要がありました。
詳しくは過去記事をご参照ください。

https://qiita.com/to-fmak/items/0da7be74a2fecd3861c6

## アプリ
(今回公開したソースコードには記載されていませんが)Subscriberでは、Iot topicへのメッセージ発行をトリガーにLambda関数を実行し、そのLambda関数に渡すpayloadに、該当topicのtopic名を含める必要がありました。色々調査した結果、topic ruleのクエリでtopic()関数を使うことで実現可能と分かりました。（公式ドキュメントに記載ありました）
詳しくは過去記事をご参照ください。

https://qiita.com/to-fmak/items/e9e28bfd0750f71369a1

## その他
CICDやユニットテスト周りでも色々な試行錯誤がありましたが、機会があればまた記事にしようと思っています。