flowsdk 0.5.3

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

"""
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.

@TODO: Add more features here
"""

import asyncio
import time
from flowsdk import FlowMqttClient


def on_message(topic: str, payload: bytes, qos: int):
    """Callback for received messages."""
    print(f"******** MSG RECEIVED ********")
    print(f"Topic: {topic}")
    print(f"Payload: {payload.decode(errors='replace')}")
    print(f"QoS: {qos}")
    print(f"******************************")


async def main():
    # Create client with message callback
    client_id = f"python_proper_{int(time.time() % 10000)}"
    client = FlowMqttClient(client_id, on_message=on_message)
    
    try:
        # Connect to broker
        host = "broker.emqx.io"
        port = 1883
        print(f"Connecting to {host}:{port}...")
        await client.connect(host, port)
        print("✅ Connected!")
        
        # Subscribe and wait for ack
        print("Subscribing to test/python/proper...")
        await client.subscribe("test/python/proper", 1)
        print("✅ Subscribe completed!")
        
        # Publish and wait for ack
        print("Publishing message...")
        await client.publish("test/python/proper", b"Hello from Proper Client!", 1)
        print("✅ Publish completed!")
        
        # Wait a bit to receive the message back
        print("Waiting for messages...")
        await asyncio.sleep(2)
        
    finally:
        print("Disconnecting...")
        await client.disconnect()
        print("✅ Disconnected!")


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