IoT Platform:例：Python 3 SDK による接続

開発環境の準備

Python 3.0 以降を使用できます。この例では Python 3.8 を使用します。

SDKのダウンロード

この例では、stomp.py および schedule ライブラリを使用します。インストール手順については、「パッケージのインストール」をご参照ください。

サンプルコード

次のサンプルコードは、stomp.py バージョン 7.0.0 に基づいています。

# encoding=utf-8
import time
import sys
import hashlib
import hmac
import base64
import stomp
import ssl
import schedule
import threading
import os
def connect_and_subscribe(conn):
    # セキュリティのため、AccessKey ペアをコードにハードコーディングしないでください。
    # この例では、参照用に環境変数から AccessKey ペアを取得します。
    accessKey = os.environ['ALIBABA_CLOUD_ACCESS_KEY_ID']
    accessSecret = os.environ['ALIBABA_CLOUD_ACCESS_KEY_SECRET']
    consumerGroupId = "${YourConsumerGroupId}"
    # iotInstanceId：インスタンスの ID。
    iotInstanceId = "${YourIotInstanceId}"
    clientId = "${YourClientId}"
    # 署名アルゴリズム。有効な値：hmacmd5、hmacsha1、hmacsha256。
    signMethod = "hmacsha1"
    timestamp = current_time_millis()
    # userName パラメーターの作成方法については、AMQP クライアントの接続に関するドキュメントをご参照ください。
    # バイナリ形式でデータを送信する場合は、userName に encode=base64 パラメーターを追加する必要があります。
    # これにより、サーバーはメッセージをプッシュする前にメッセージ本文を Base64 でエンコードします。
    # 詳細については、本トピックの「バイナリ形式」セクションをご参照ください。
    username = clientId + "|authMode=aksign" + ",signMethod=" + signMethod \
                    + ",timestamp=" + timestamp + ",authId=" + accessKey \
                    + ",iotInstanceId=" + iotInstanceId \
                    + ",consumerGroupId=" + consumerGroupId + "|"
    signContent = "authId=" + accessKey + "&timestamp=" + timestamp
    # 署名を計算します。password の作成方法については、AMQP クライアントの接続に関するドキュメントをご参照ください。
    password = do_sign(accessSecret.encode("utf-8"), signContent.encode("utf-8"))
    conn.set_listener('', MyListener(conn))
    conn.connect(username, password, wait=True)
    # 以前の接続チェックタスクをクリアし、新しいタスクを作成します。
    schedule.clear('conn-check')
    schedule.every(1).seconds.do(do_check,conn).tag('conn-check')
class MyListener(stomp.ConnectionListener):
    def __init__(self, conn):
        self.conn = conn
    def on_error(self, frame):
        print('received an error "%s"' % frame.body)
    def on_message(self, frame):
        print('received a message "%s"' % frame.body)
    def on_heartbeat_timeout(self):
        print('on_heartbeat_timeout')
    def on_connected(self, headers):
        print("successfully connected")
        conn.subscribe(destination='/topic/#', id=1, ack='auto')
        print("successfully subscribe")
    def on_disconnected(self):
        print('disconnected')
        connect_and_subscribe(self.conn)
def current_time_millis():
    return str(int(round(time.time() * 1000)))
def do_sign(secret, sign_content):
    m = hmac.new(secret, sign_content, digestmod=hashlib.sha1)
    return base64.b64encode(m.digest()).decode("utf-8")
# 接続を確認し、切断された場合は再接続します。
def do_check(conn):
    print('check connection, is_connected: %s', conn.is_connected())
    if (not conn.is_connected()):
        try:
            connect_and_subscribe(conn)
        except Exception as e:
            print('disconnected, ', e)
# 接続ステータスをチェックするためのスケジュールされたタスク。
def connection_check_timer():
    while 1:
        schedule.run_pending()
        time.sleep(1)
# 接続エンドポイント。詳細については、AMQP クライアントの接続に関するドキュメントをご参照ください。
# エンドポイントを直接入力します。amqps://プレフィックスは追加しないでください。
conn = stomp.Connection([('${YourHost}', 61614)], heartbeats=(0,300))
conn.set_ssl(for_hosts=[('${YourHost}', 61614)], ssl_version=ssl.PROTOCOL_TLS)
try:
    connect_and_subscribe(conn)
except Exception as e:
    print('connecting failed')
    raise e
# 非同期スレッドでスケジュールされた接続チェックを実行します。
thread = threading.Thread(target=connection_check_timer)
thread.start()

コード内のプレースホルダーを実際の値に置き換えてください。パラメーターの詳細については、「AMQP クライアントの IoT Platform への接続」をご参照ください。

実行結果の例

• 成功：次のようなログメッセージが表示された場合、AMQP クライアントが IoT Platform に接続し、メッセージを受信していることを示します。

  successfully connected
  successfully subscribe
  check connection, is_connected: %s True
  check connection, is_connected: %s True

• 失敗：次のようなログメッセージが表示された場合、AMQP クライアントが IoT Platform への接続に失敗したことを示します。

  ログに基づいてコードまたはネットワーク環境を確認し、問題を解決してから、コードを再度実行できます。

  Could not connect to host ixxx.com, port 61614
  Traceback (most recent call last):
    File "xxx"
      self.xxx
    File "xxx"
      for r xxx
    File "xxx"
      for r xxx
  socket.gaierror: [Errno 11001] getaddrinfo failed

バイナリ形式

バイナリデータを送信するには、Base64 エンコーディングパラメーターを使用してください。STOMP はテキストベースのプロトコルであるため、エンコーディングを行わないと、メッセージ本文が切り捨てられる可能性があります。

次に示すように、userName に encode=base64 パラメーターを追加します。サーバーは、メッセージを送信する前にメッセージ本文を Base64 でエンコードします。

username = clientId + "|authMode=aksign" + ",signMethod=" + signMethod \
                    + ",timestamp=" + timestamp + ",authId=" + accessKey \
                    + ",iotInstanceId=" + iotInstanceId \
                    + ",consumerGroupId=" + consumerGroupId \ 
                    + ",encode=base64"+"|"

関連情報

サーバー側サブスクリプション機能に関連するエラーコードの詳細については、「IoT Platform ログ」トピックのメッセージに関連するエラーコードセクションをご参照ください。