MQTT over WebSocket: Paho で接続

次のページを参考にしました。
PythonでMQTT(Paho) - 2023

こちらで立てたブローカーを使います。
MQTT over WebSocket: localhost でのテスト

ライブラリーのインストール

sudo pacman -S python-paho-mqtt

Publish

pub_8083.py

#! /usr/bin/python

import random
import time

from paho.mqtt import client as mqtt_client

broker = 'localhost'
port = 8083
topic = "example/testTopic"

client_id = f'publish-{random.randint(0, 1000)}'

def connect_mqtt():
    def on_connect(client, userdata, flags, rc):
        if rc == 0:
            print("Connected to MQTT Broker!")
        else:
            print("Failed to connect, return code %d\n", rc)

    client = mqtt_client.Client(client_id)
    client.on_connect = on_connect
    client.connect(broker, port)
    return client

def publish(client):
    msg_count = 1
    while True:
        time.sleep(1)
        msg = f"messages: {msg_count}"
        result = client.publish(topic, msg)
        status = result[0]
        if status == 0:
            print(f"Send `{msg}` to topic `{topic}`")
        else:
            print(f"Failed to send message to topic {topic}")
        msg_count += 1
        if msg_count > 5:
            break

def run():
    client = connect_mqtt()
    client.loop_start()
    publish(client)
    client.loop_stop()

if __name__ == '__main__':
    run()
#

Subscribe

sub_8083.py

#! /usr/bin/python

import random
from paho.mqtt import client as mqtt_client

broker = 'localhost'
port = 8083
topic = "example/testTopic"

client_id = f'subscribe-{random.randint(0, 100)}'

def connect_mqtt() -> mqtt_client:
    def on_connect(client, userdata, flags, rc):
        if rc == 0:
            print("Connected to MQTT Broker!")
        else:
            print("Failed to connect, return code %d\n", rc)

    client = mqtt_client.Client(client_id)
    client.on_connect = on_connect
    client.connect(broker, port)
    return client

def subscribe(client: mqtt_client):
    def on_message(client, userdata, msg):
        print(f"Received `{msg.payload.decode()}` from `{msg.topic}` topic")

    client.subscribe(topic)
    client.on_message = on_message

def run():
    client = connect_mqtt()
    subscribe(client)
    client.loop_forever()

if __name__ == '__main__':
    run()
#

確認したバージョン

$ python --version
Python 3.13.1