Python'da WebSocket üzerinden MQTT

Python'da WebSocket üzerinden MQTT kullanmak için paho istemcisini transport="websockets" ile oluşturun, broker'ın WebSocket portuna bağlanın ve gerekirse URL path'ini ws_set_options() ile ayarlayın. Şifreli wss için tls_set() ekleyin ve güvenli WebSocket portunu kullanın. Bunun dışında MQTT API'si aynıdır: aynı callback'ler, aynı publish() ve subscribe().

Bu rehber ws ve wss bağlantılarını, özel path ve header'ları, HTTP ve SOCKS proxy'leri üzerinden bağlanmayı ve paho'nun fırlattığı handshake hatalarının nasıl ayıklanacağını gösteriyor.

Python'dan WebSocket ne zaman mantıklı

Tarayıcılar MQTT'yi yalnızca WebSocket üzerinden konuşabilir, ama Python daha basit ve biraz daha hafif olan düz TCP'yi kullanabilir. Python'da WebSocket'i şu durumlarda seçin:

• Ağınız yalnızca giden HTTP(S)'e izin veriyorsa; örneğin 1883 ve 8883'ü engelleyen bir kurumsal firewall veya otel ağı (bkz. kurumsal firewall arkasında MQTT).
• Broker, Nginx veya bir bulut ingress'i gibi yalnızca HTTP yönlendiren bir reverse proxy ya da load balancer arkasındaysa.
• Bir tarayıcı sorununu yeniden üretmek için bir backend script'inin web dashboard'unuzla birebir aynı uç noktayı kullanmasını istiyorsanız.

Bunların hiçbiri geçerli değilse 1883/8883 üzerinde TCP varsayılan seçimdir. Ek yük farkı TCP ve WebSocket performansı yazısında ölçülüyor.

Şifresiz WebSocket (ws://)

import time

import paho.mqtt.client as mqtt


def on_connect(client, userdata, flags, reason_code, properties):
    print("CONNACK:", reason_code)
    client.subscribe("testmqtt/py-ws/#")


def on_message(client, userdata, msg):
    print(msg.topic, msg.payload.decode())


client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, transport="websockets")
client.ws_set_options(path="/mqtt")  # /mqtt aynı zamanda paho'nun varsayılanı
client.on_connect = on_connect
client.on_message = on_message

client.connect("broker.connectmq.com", 8083, keepalive=60)
client.loop_start()
client.publish("testmqtt/py-ws/hello", "hello over ws", qos=1)
time.sleep(2)
client.loop_stop()
client.disconnect()

transport constructor'da ayarlanmalıdır; mevcut bir istemciyi TCP'den WebSocket'e geçiremezsiniz. Geçerli değerler "tcp", "websockets" ve "unix"'tir; başka bir değer ValueError: transport must be "websockets", "tcp" or "unix", not ... hatası verir.

Güvenli WebSocket (wss://)

Hızlı bir testin ötesindeki her şey için wss kullanın. Aynı kod, artı tls_set(), güvenli port ve gerçek kimlik bilgileri:

import os

import paho.mqtt.client as mqtt

client = mqtt.Client(
    mqtt.CallbackAPIVersion.VERSION2,
    client_id="py-wss-demo",
    transport="websockets",
    protocol=mqtt.MQTTv5,
)
client.username_pw_set(os.environ["MQTT_USER"], os.environ["MQTT_PASS"])
client.tls_set()                     # sertifikayı ve hostname'i doğrular
client.ws_set_options(path="/mqtt")

client.on_connect = lambda c, u, f, rc, p: print("CONNACK:", rc)
client.connect("broker.connectmq.com", 8084, keepalive=60)
client.loop_forever()

paho önce TLS handshake'ini, sonra şifreli kanal içinde HTTP upgrade'ini, ardından MQTT CONNECT'i gerçekleştirir. Python'da TLS üzerinden MQTT yazısındaki tüm TLS seçenekleri ve hataları aynen geçerlidir.

Path'ler ve özel header'lar

WebSocket path'i broker'ın veya proxy'nin beklediğiyle eşleşmelidir ve evrensel bir standart yoktur. En yaygını /mqtt'dir (EMQX, HiveMQ ve birçok yönetilen broker'da varsayılan), RabbitMQ'nun Web MQTT eklentisi /ws kullanır ve reverse proxy'ler istediğiniz herhangi bir path'i eşleyebilir. Yanlış path, WebSocket bağlantı hatalarının bir numaralı nedenidir; bkz. WebSocket path 404.

ws_set_options() header da kabul eder: ya varsayılanlarla birleştirilen bir dict olarak ya da varsayılan header'ları alıp son kümeyi döndüren bir fonksiyon olarak:

import os

TOKEN = os.environ["GATEWAY_TOKEN"]

def add_headers(headers):
    headers["Authorization"] = f"Bearer {TOKEN}"   # broker'ın önündeki bir auth proxy'si için
    headers["Origin"] = "https://dashboard.example.com"
    return headers

client.ws_set_options(path="/mqtt", headers=add_headers)

paho varsayılan olarak Origin: http://host:port (TLS ile https://) ve Sec-WebSocket-Protocol: mqtt gönderir. Bir proxy izin listesi uyguladığında Origin'i ezmek önemlidir; bu konu CORS ve WebSocket origin kontrolleri yazısında ele alınıyor. MQTT kimlik bilgilerinin hâlâ username_pw_set() ile CONNECT paketinde gittiğini unutmayın; header'lar yalnızca broker'ın önündeki HTTP bileşenleri içindir.

HTTP veya SOCKS proxy üzerinden bağlanma

paho proxy'lerden PySocks kütüphanesiyle tünel açar. Önce pip install pysocks ile kurun, ardından proxy'yi açıkça yapılandırın:

import socks  # PySocks paketinden

import paho.mqtt.client as mqtt

client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, transport="websockets")
client.tls_set()
client.proxy_set(proxy_type=socks.HTTP, proxy_addr="proxy.corp.local", proxy_port=3128)
client.connect("broker.connectmq.com", 8084)
client.loop_forever()

socks.HTTP ile PySocks, broker.connectmq.com:8084'e bir HTTP CONNECT tüneli açar; paho da TLS'i, WebSocket upgrade'ini ve MQTT'yi bunun içinden geçirir. SOCKS proxy için socks.SOCKS5 kullanın (isteğe bağlı olarak proxy_username / proxy_password ile).

İki ayrıntı çok zaman kazandırır:

• paho HTTPS_PROXY ve HTTP_PROXY'yi yok sayar. Okuduğu tek ortam değişkeni mqtt_proxy'dir; örneğin export mqtt_proxy=http://proxy.corp.local:3128 veya socks://host:1080, o da yalnızca PySocks kuruluysa. no_proxy'deki host'lara doğrudan bağlanılır.
• Birçok kurumsal proxy CONNECT'e yalnızca 443 portu için izin verir. Tünel reddedilirse 443'te bir broker uç noktasına veya bir proxy istisnasına ihtiyacınız vardır.

Handshake hataları ve çözümleri

WebSocket sorunları, ConnectionError'ın bir alt sınıfı olan WebsocketConnectionError olarak ortaya çıkar. connect() ile hemen fırlatılır; connect_async() ve loop_start() ile paho sessizce yeniden dener, bu yüzden hata ayıklarken client.enable_logger()'ı açın.

Son satırdaki MQTT reason code'larını reason code arama aracıyla bulabilirsiniz.

Bağlantıyı nasıl test edersiniz

İşe Python olmadan HTTP upgrade'ini kontrol ederek başlayın. Sağlıklı bir WebSocket uç noktası 101 Switching Protocols yanıtı verir:

curl -i -N --max-time 5 \
  -H "Connection: Upgrade" -H "Upgrade: websocket" \
  -H "Sec-WebSocket-Version: 13" \
  -H "Sec-WebSocket-Key: dGhlIHNhbXBsZSBub25jZQ==" \
  -H "Sec-WebSocket-Protocol: mqtt" \
  https://broker.connectmq.com:8084/mqtt

404 path'in yanlış olduğunu, 400 veya 426 genellikle bir header'ın silindiğini, TLS hatası ise yanlış portla konuştuğunuzu gösterir. Ardından uçtan uca kontrol yapın: online MQTT istemcisini açın, aynı broker'a wss ile bağlanın, testmqtt/py-ws/#'e abone olun ve ws_client.py'yi çalıştırın. Tarayıcı mesajınızı alıyorsa Python, path ve broker doğrudur. Proxy'ler ve load balancer'lar dahil daha fazla kontrol WebSocket MQTT bağlantısını test etme yazısında.

Tarayıcıda abone olun, Python'dan publish edin

Bunun yerine aiomqtt kullanmak

Kodunuz asyncio tabanlıysa, paho'yu saran aiomqtt aynı seçenekleri constructor argümanları olarak sunar:

import asyncio
import ssl

import aiomqtt


async def main():
    async with aiomqtt.Client(
        "broker.connectmq.com",
        8084,
        transport="websockets",
        websocket_path="/mqtt",
        tls_context=ssl.create_default_context(),
    ) as client:
        await client.subscribe("testmqtt/py-ws/#")
        await client.publish("testmqtt/py-ws/hello", "hello from aiomqtt")
        async for message in client.messages:
            print(message.topic, message.payload)
            break


asyncio.run(main())

Python'da WebSocket üzerinden MQTT, bir constructor argümanı artı doğru port ve path'ten ibarettir. Doğrulama açıkken wss kullanın, kimsenin tahmin etmek zorunda kalmaması için path'i açıkça belirtin, proxy'leri proxy_set() veya mqtt_proxy ile yapılandırın ve kodunuzu suçlamadan önce upgrade'i curl ile test edin.

Sıkça sorulan sorular