Pythonで接続する

以下は、ブローカーでTOKEN認証モードが設定されたpaho.mqtt.python ↗パッケージを使用した例です。

以下の例では、シンプルなサブスクライバーを作成し、設定されたトピックにメッセージを送信し、メッセージが受信されるまで待機してから終了します。

例を実行する前に、以下の環境変数を設定してください：

• BROKER_FQDN - 例：YOUR-BROKER.YOUR-NAMESPACE.cloudflarepubsub.com（ポートやmqtts://スキームなし）
• BROKER_TOKEN（有効な認証トークン）
• BROKER_TOPIC - 例：test/topicまたはhello/world

以下の例はPython 3.8を使用していますが、Python 3.6以上で実行可能です。

# paho-mqttがインストールされていることを確認してください
pip3 install paho-mqtt

次の内容でpubsub.pyというファイルを作成し、python3 pubsub.pyを使用して例を実行します：

# ライブラリをインストールするには：pip install paho-mqtt

import os
import paho.mqtt.client as mqtt
import sys


# すべての環境変数が設定されていることを確認する
def check_env(env):
    if env is None:
        sys.exit("BROKER_FQDN、BROKER_TOKEN、およびBROKER_TOPICを設定する必要があります。")
    return env


# クライアントがサーバーからCONNACK応答を受信したときのコールバック。
def on_connect(ctx, userdata, flags, rc, properties):
    print("接続しました: {}".format(ctx._host))
    ctx.subscribe(topic)
    client.publish(topic, "PythonとPub/Subからのこんにちは！")


# サーバーからPUBLISHメッセージを受信したときのコールバック。
def on_message(ctx, userdata, msg):
    print("{}: {}".format(msg.topic, msg.payload))
    # さようなら
    client.disconnect()


# MQTTブローカーのFQDNを指定します：<broker name>.<namespace>.cloudflarepubsub.com
fqdn = check_env(os.environ.get("BROKER_FQDN"))

# 任意のユーザー名と/brokers/YOUR_BROKER/credentialsエンドポイントからのトークン
# トークンはPub/Sub APIによって発行されたbase64エンコードされたJWTである必要があります
username = "anything"
password = check_env(os.environ.get("BROKER_TOKEN")).strip("\"")

# 購読および発行するトピック名を指定します
topic = check_env(os.environ.get("BROKER_TOPIC"))

# MQTTクライアントを作成します
client = mqtt.Client(client_id="", protocol=mqtt.MQTTv5)

# ユーザー名とパスワードを設定します
client.username_pw_set(username, password)

# TLSを有効にします
client.tls_set()

# ブローカーに接続し、コールバック関数を登録します
client.connect(fqdn, 8883)
client.on_connect = on_connect
client.on_message = on_message

# メッセージを受信するまで待機します
client.loop_forever()