AWS IotでPub/Subしてみる - 闘うITエンジニアの覚え書き

AWS IoT SDK を使用してMQTT接続するクライアントを書く。

参考
AWS IoT への Raspberry Pi の接続 - https://docs.aws.amazon.com/ja_jp/iot/latest/developerguide/iot-sdk-setup.html

ポリシーの作成 †

証明書の作成 †

モノの登録 †

証明書にモノにアタッチ †

エンドポイントの確認 †

いくつか方法はあるが、ここでは AWSIoTPythonSDK を使用した。
尚、テスト用の Publish は マネージメントコンソール から簡単に行う事ができるが、一応 boto3 から Publish する処理を作る為、ここでは boto3 もインストールしておく。
※aws configure は事前にしておく事。

GitHub
https://github.com/aws/aws-iot-device-sdk-python

mkdir test_iot && test_iot
python3 -m venv venv
source venv/bin/activate
pip install boto3
pip install AWSIoTPythonSDK

https://github.com/aws/aws-iot-device-sdk-python/blob/master/samples/basicPubSub/basicPubSub.py を参考に任意のトピックをサブスクライブするクライアントを作成する。

client.py

# coding: utf-8

from AWSIoTPythonSDK.MQTTLib import AWSIoTMQTTClient
import logging
import time
import argparse
import json

# MQTTの設定情報
host = 設定画面で確認したエンドポイント
port = 8883
rootCAPath = ルートCA証明書ファイルPATH
certificatePath = モノの証明書ファイルPATH
privateKeyPath = プライベートキーのファイルPATH
clientId = 'sample-thing1'  # Thing Name と同じである事が推奨されている
topic = 'topic/SampleTopic1'  # サブスクライブするTopic
# topic = 'topic/#'  # トピックの前方一致でサブスクライブする事も可能

# メッセージ受信時のコールバック
def customCallback(client, userdata, message):
    print("Received a new message: ")
    print(message.payload)
    print("from topic: ")
    print(message.topic)
    print("--------------\n\n")

# ログ設定
logger = logging.getLogger("AWSIoTPythonSDK.core")
logger.setLevel(logging.DEBUG)
streamHandler = logging.StreamHandler()
formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
streamHandler.setFormatter(formatter)
logger.addHandler(streamHandler)

# MQTTクライアントの初期化/設定
myAWSIoTMQTTClient = AWSIoTMQTTClient(clientId)
myAWSIoTMQTTClient.configureEndpoint(host, port)
myAWSIoTMQTTClient.configureCredentials(rootCAPath, privateKeyPath, certificatePath)
myAWSIoTMQTTClient.configureAutoReconnectBackoffTime(1, 32, 20) 
myAWSIoTMQTTClient.configureOfflinePublishQueueing(-1)  # Infinite offline Publish queueing
myAWSIoTMQTTClient.configureDrainingFrequency(2)  # Draining: 2 Hz
myAWSIoTMQTTClient.configureConnectDisconnectTimeout(10)  # 10 sec
myAWSIoTMQTTClient.configureMQTTOperationTimeout(5)  # 5 sec

# 接続
myAWSIoTMQTTClient.connect()

# サブスクライブ
myAWSIoTMQTTClient.subscribe(topic, 1, customCallback)

while True:
    time.sleep(1)

test_publish.py

# coding: utf-8

import boto3
import json
from datetime import datetime

def main():
    client = boto3.client('iot-data')
    now = datetime.now()
    payload = json.dumps({
        'message': f'Hello AWS Iot! {now}'
    }).encode()
    response = client.publish(
        topic='topic/SampleTopic1',
        qos=0,
        payload=payload
    )   
    print(response)


if __name__ == '__main__':
    main()

※ https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/iot-data.html#IoTDataPlane.Client.publish

デバイス側のクライアントを起動

python client.py

テストメッセージをPublish

python test_publish.py

結果(デバイス側のクライアント)

2018-01-01 14:26:39,574 - AWSIoTPythonSDK.core.protocol.internal.workers - DEBUG - Dispatching [message] event
Received a new message: 
b'{"message": "Hello AWS Iot! 2018-01-01 14:26:39.287980"}'
from topic: 
topic/SampleTopic1
--------------

ワイルドカードを使用したトピックのサブスクライブ †