flowsdk 0.5.2

Safety-first, realistic, behavior-predictable messaging SDK for MQTT and more.
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75

"""
Example demonstrating the high-level FlowMqttClient for asyncio applications.

This example shows how to use the FlowMqttClient class from the flowsdk package
to easily connect, subscribe, publish, and disconnect from an MQTT broker using
async/await syntax.
"""
import asyncio
import time
import os
import sys

# Add the package to path
sys.path.append(os.path.join( "python", "package"))

from flowsdk import FlowMqttClient, TransportType

async def test_tls():
    print("🚀 Testing TLS Transport Support...")
    
    msg_received = asyncio.Event()

    def on_message(topic: str, payload: bytes, qos: int):
        print(f"📨 Received message on {topic}: {payload.decode()}")
        if topic == "test/python/tls" and payload == b"Hello TLS!":
            msg_received.set()

    # Check if TLS key logging is requested via environment variable
    keylog_file = os.environ.get("SSLKEYLOGFILE")
    enable_key_log = keylog_file is not None
    if enable_key_log:
        print(f"🔑 TLS key logging enabled → {keylog_file}")

    # We'll use broker.emqx.io:8883 which supports TLS
    client = FlowMqttClient(
        client_id=f"python_tls_test_{int(time.time() % 10000)}",
        transport=TransportType.TLS,
        insecure_skip_verify=False,  # Skip verification for simplicity in test
        server_name="broker.emqx.io",
        enable_key_log=enable_key_log,
        on_message=on_message
    )
    
    try:
        host = "broker.emqx.io"
        port = 8883
        print(f"📡 Connecting to {host}:{port} (TLS)...")
        await client.connect(host, port)
        print("✅ TLS Connected!")
        
        await client.subscribe("test/python/tls", 1)
        print("✅ Subscribed!")
        
        await client.publish("test/python/tls", b"Hello TLS!", 1)
        print("✅ Published!")
        
        # Wait for the message with a timeout
        print("⏳ Waiting for message reception...")
        try:
            await asyncio.wait_for(msg_received.wait(), timeout=5.0)
            print("✨ Message verification successful!")
        except asyncio.TimeoutError:
            print("❌ Timeout waiting for message!")
        
    except Exception as e:
        print(f"❌ TLS Test Failed: {e}")
        import traceback
        traceback.print_exc()
    finally:
        print("👋 Disconnecting...")
        await client.disconnect()
        print("✅ Disconnected!")

if __name__ == "__main__":
    asyncio.run(test_tls())