85 lines
2.0 KiB
Python
85 lines
2.0 KiB
Python
#!/usr/bin/env python3
|
|
"""Test MQTT over WebSocket (wss) via nginx proxy at /mqtt.
|
|
Usage:
|
|
pip install -r requirements.txt
|
|
python test_ws_mqtt.py
|
|
|
|
Environment:
|
|
MQTT_HOST (default: dev.hub.umagul.net)
|
|
MQTT_PORT (default: 443)
|
|
MQTT_PATH (default: /mqtt)
|
|
MQTT_TOPIC (default: devices/#)
|
|
MQTT_USERNAME, MQTT_PASSWORD (optional)
|
|
TLS_VERIFY (true/false, default: true)
|
|
"""
|
|
import os
|
|
import ssl
|
|
import time
|
|
import json
|
|
import sys
|
|
|
|
import paho.mqtt.client as mqtt
|
|
|
|
HOST = os.getenv("MQTT_HOST", "dev.hub.umagul.net")
|
|
PORT = int(os.getenv("MQTT_PORT", "443"))
|
|
PATH = os.getenv("MQTT_PATH", "/mqtt")
|
|
TOPIC = os.getenv("MQTT_TOPIC", "devices/#")
|
|
USERNAME = os.getenv("MQTT_USERNAME", "testuser")
|
|
PASSWORD = os.getenv("MQTT_PASSWORD", "testpass")
|
|
TLS_VERIFY = os.getenv("TLS_VERIFY", "true").lower() in ("1", "true", "yes")
|
|
|
|
received = 0
|
|
|
|
def on_connect(client, userdata, flags, rc):
|
|
if rc == 0:
|
|
print(f"connected to wss://{HOST}{PATH} (rc={rc})")
|
|
client.subscribe(TOPIC)
|
|
print(f"subscribed to {TOPIC}")
|
|
else:
|
|
print("connect failed rc=", rc)
|
|
|
|
|
|
def on_message(client, userdata, msg):
|
|
global received
|
|
received += 1
|
|
payload = msg.payload.decode(errors="ignore")
|
|
print(f"[{received}] {msg.topic} -> {payload}")
|
|
|
|
|
|
def on_disconnect(client, userdata, rc):
|
|
print("disconnected", rc)
|
|
|
|
|
|
def main():
|
|
client = mqtt.Client(transport="websockets")
|
|
if USERNAME:
|
|
client.username_pw_set(USERNAME, PASSWORD)
|
|
|
|
client.ws_set_options(path=PATH)
|
|
|
|
if TLS_VERIFY:
|
|
client.tls_set() # default system CA
|
|
else:
|
|
client.tls_set(cert_reqs=ssl.CERT_NONE)
|
|
client.tls_insecure_set(True)
|
|
|
|
client.on_connect = on_connect
|
|
client.on_message = on_message
|
|
client.on_disconnect = on_disconnect
|
|
|
|
try:
|
|
client.connect(HOST, PORT, keepalive=60)
|
|
except Exception as e:
|
|
print("connect error:", e)
|
|
sys.exit(1)
|
|
|
|
try:
|
|
client.loop_forever()
|
|
except KeyboardInterrupt:
|
|
print("exiting")
|
|
client.disconnect()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|