Amazon SNSとAmazon SQSでPubSubの仕組みを実装する

## 構成図
以下のような仕組みをつくっていきたいと思います。
![スクリーンショット 2023-06-26 121400.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/484097/b633bd0b-1c44-2c12-dba6-e359bde851e6.png)
PublisherのLambdaはコンソールから「Test」実行し、SNSトピックをPublishし、SQSでキューに追加された後に、SubscriberのLambdaが起動します。SubscriberのLambdaではメッセージに含まれる内容をDynamoDBに追加します。

### 前提条件
* Administrator権限を持つAWSアカウント
* LambdaはPython3.8で記述

以下の順番で内容を記載していきたいと思います。
1. SNSトピックの設定
1. SQSキューの設定
1. DynamoDBの設定
1. Publisher Lambdaの中身とSubscriber Lambdaの中身
1. DynamoDBとCloudWatch Logsで実行確認

## SNSトピックの設定
今回はStandardタイプでトピックを作成します。
![スクリーンショット 2023-06-26 122254.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/484097/745e4ffe-6e08-fdbc-bf5d-13c5dd0b34fc.png)
トピックがうまく配信されない場合に、細かいログも確認したいのでLoggingの設定もします。今回はSQSにトピックを配信するので、ロギングの対象をAmazon SQSとします。既存のIAMロールが存在しない場合は、「Create New Service Role」から画面遷移に従ってロールを作成していきます。今回は既に作成済だったので、対象のロールをセットしています。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/484097/4be075ec-ef30-e5c7-34f9-ee05c2ef5035.png)
その他の項目はデフォルトで「作成」します。一旦ここまでの設定です。

## Amazon SQSの設定
次にSQSのキューを作成していきます。こちらもStandardタイプで作成します。
（※キューの名前を打ち間違いしました。。。）
![スクリーンショット 2023-06-26 131607.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/484097/3ad61c49-63fa-ef34-e477-ab352854b967.png)
後はデフォルトで進めていき、「作成」をします。
以下の2点は追加で設定をした部分です。デッドレターキューはオプションですが、アクセス権限の部分はうまく設定されてないとSNSからメッセージを受信できないエラーとなりました。

### デッドレターキューの設定
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/484097/b910c63f-bda4-669c-402b-01efe3572dbf.png)
### アクセスポリシーの設定
![スクリーンショット 2023-06-26 132026.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/484097/4b28f8c0-bba0-3a64-d0a6-22c6bf8e4696.png)
### SNSトピックのサブスクライブの設定
一つ前の手順で作成したSNSトピックをサブスクライブする設定をします。SQSの画面から実施できます。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/484097/de8e30c6-4198-1906-635d-4421093fd6ba.png)

## 再びAmazon SNSの設定
SQS側でSNSのトピックのサブスクリプションの設定をすると、SNS側のトピックにも反映されます。少し追加の設定をします。
![スクリーンショット 2023-06-26 132841.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/484097/c253802f-06d7-8c3e-801b-039df55e1ad8.png)
まずは`Enable raw message delivery`のチェックボックスをONにします。この部分をONにすることで、Publisher LambdaでSNSにパブリッシュしたメッセージをそのまま受信できるようになります。（最初はOFFにしていましたが、そうするとSNS側で付与される属性情報なども含めたメッセージを受信するカタチになります。）
![スクリーンショット 2023-06-26 133007.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/484097/092444f3-0e54-c137-4fec-7174a7178852.png)
配信に失敗したトピックを格納しておくように、先ほどのデッドレターキューを指定します。オプション設定です。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/484097/ac1f2621-74f2-a2c9-c840-619e5bf91420.png)

## DynamoDBの設定
DynamoDBはシンプルに設定していきます。今回はPartition Keyを`message_id`、Sort Keyを`timestamp`としてみます。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/484097/745e5aa4-b121-37ec-8a98-b9d4317f1c70.png)

## Publisher用のLambdaを作成
Lambdaが起動したときのログ出力と、SNSにトピックをパブリッシュしているシンプルなLambdaです。`Message`パラメータには、Test実行時に指定した文字列を渡すようにしています。
```python
import boto3
from datetime import datetime
import json

client = boto3.client('sns')

def logging(errorLv, LambdaName, errorMsg):
    loggingDateStr=(datetime.now()).strftime('%Y%m%d %H:%M:%S')
    print(loggingDateStr + " " + LambdaName + " " + "[" + errorLv + "] " + errorMsg)
    return

def lambda_handler(event, context):
    
    logging("info", context.function_name, "lambda started")
    
    params = {
        'TopicArn': 'arn:aws:sns:ap-northeast-1:xxxxxxxxxxxx:myFirstTopic_std',
        'Subject' : 'Published From: deliverTopicToSNS_Test01',
        'Message' : event['message']
    }
    
    response = client.publish(
        TopicArn = params['TopicArn'],
        Subject = params['Subject'],
        Message = params['Message']
    )
    
    print(json.dumps(response))
    return response

```
以下がTest実行時に渡される文字列です。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/484097/a63ba19a-5c85-8dd2-9ee6-18b8645c1d12.png)

## Subscriber用のLambdaを作成
SQSから渡されるメッセージの内容をDynamoDBにputしているLambdaです。本来はメッセージを受信して処理が完了したら、メッセージを削除する流れがよいと思います。
```python
import json
import boto3
from datetime import datetime

dynamodb = boto3.resource('dynamodb')
table = dynamodb.Table('sns_sqs_lambda_table')
sqs_client = boto3.client("sqs")
queue_url = "https://sqs.ap-northeast-1.amazonaws.com/xxxxxxxxxxxx/SnsToSnsQueue"

def lambda_handler(event, context):
    
    message = event['Records'][0]['body']
    message_id = event['Records'][0]['messageId']
   
    try:
        result = table.put_item(Item={
            'message_id':message_id,
            'subject': 'fixed subject',
            'message': message,
            'timestamp': datetime.now().isoformat(timespec='seconds')
        })
        
        # sqs_client.delete_message(QueueUrl=queue_url, ReceiptHandle=event['Records'][0]['ReceiptHandle'])
        
    except Exception as e:
        print(e)
        raise e
   
    print(result)
    return result
```
### Lambdaを起動するようにSQSで追加設定
SQSのキューにメッセージが格納された後に、作成したSubscriberのLambdaが起動されるように設定をします。この設定はSQS側で実施します。
Configure Lambda FUnction Triggerから対象のLambdaを選択するだけです。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/484097/bb399b57-32e9-f4e3-15b4-fdc64f530d2a.png)
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/484097/05626ba4-f2c7-833f-3d1e-074ba7c48b58.png)
この設定が完了すると、Lambda画面にSQSからのトリガーが追加されます。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/484097/f5effd18-3c89-a1e1-30bc-89f294346f38.png)
ここまでで一通り設定が完了しました。

## テスト実行と結果確認
Publisher用のLambdaから「Test」を実行します。全てうまく起動していれば、DynamoDBに新規ItemがPutされているはずです。
DynamoDBのテーブルを確認すると、無事に1件新規のItemが追加されていました！
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/484097/45aee1aa-8f88-ed6b-7289-b71b15ba5630.png)
念のためPublisher用Lambdaも確認します。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/484097/49050ea6-b95d-979e-f8ee-de250debc66a.png)
Subscriber用のLambdaも確認します。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/484097/2558fdf1-8af6-8377-4d26-959e2216c66d.png)
Lambdaも無事に動作していることが確認できました。
SNSトピックがちゃんと配信されたかどうかはsnsのログから確認できます。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/484097/295915cf-2577-c46a-b1e1-f06bf1eac86b.png)
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/484097/c8aaffd0-407c-ec40-11e4-9c164966e282.png)
こちらも無事に動いていることが確認できました。

## 終わりに
ここまでLambda⇒SNS⇒SQS⇒Lambda⇒DynamoDBの構成を構築してきました。次回はこの構成に手を加えてAPI GatewayやEmailで処理完了通知などを受け取る仕組みを実装していきたいと思います。