Skip to content

examples

Directory actions

More options

Directory actions

More options

Latest commit

 

History

History

examples

README.md

FlowSDK Python Examples

This directory contains examples demonstrating various usage patterns of the FlowSDK Python bindings.

Quick Start Examples

1. Simple Async Usage (simple_async_usage.py)

Minimal async MQTT client example - Start here!

PYTHONPATH=../package python3 simple_async_usage.py

The simplest way to use FlowSDK with asyncio. Shows connect, subscribe, publish, and disconnect in just a few lines.

2. Asyncio TCP Client (asyncio_tcp_client_example.py)

Standard async client using TCP

PYTHONPATH=../package python3 asyncio_tcp_client_example.py

Demonstrates the high-level FlowMqttClient API using standard TCP transport with message callbacks and proper error handling.

Secure Transport Examples

3. TLS Transport (asyncio_tls_client_example.py)

MQTT over TLS for secure communication

PYTHONPATH=../package python3 asyncio_tls_client_example.py

Shows how to use MQTT over TLS (port 8883) with the unified FlowMqttClient. Includes:

  • Explicit TLS transport selection
  • Certificate verification (default)
  • SNI support via server_name

4. QUIC Transport (asyncio_quic_client_example.py)

MQTT over QUIC for modern UDP-based transport

PYTHONPATH=../package python3 asyncio_quic_client_example.py

Shows how to use MQTT over QUIC protocol with the unified FlowMqttClient. QUIC provides:

  • Built-in encryption (TLS 1.3)
  • Connection migration
  • Lower latency than TCP
  • No head-of-line blocking

Requirements: QUIC-enabled MQTT broker (e.g., EMQX 5.0+ with QUIC listener on port 14567)

Low-Level Examples

5. Select-Based Client (select_example.py)

Non-blocking I/O using select()

PYTHONPATH=../package python3 select_example.py

Demonstrates manual socket management with the select() system call.

6. Async with Manual I/O (async_example.py)

Asyncio with manual socket operations

PYTHONPATH=../package python3 async_example.py

Shows how to manually integrate the MQTT engine with asyncio event loop using socket I/O.

Testing

8. Binding Tests (test_binding.py)

Low-level FFI binding verification

PYTHONPATH=../package python3 test_binding.py

Architecture Overview

High-Level API (Recommended)

FlowMqttClient → FlowMqttProtocol → MqttEngineFfi → Rust FFI
  • Simple async/await API
  • Automatic network handling
  • Opt-in automatic reconnection with bounded backoff

Low-Level API (Advanced)

Your Code → MqttEngineFfi → Manual I/O → Network

Transport Selection

FlowSDK supports multiple transport protocols via the unified FlowMqttClient:

TCP

client = FlowMqttClient("my_client", transport=TransportType.TCP)
await client.connect("broker.emqx.io", 1883)

TLS

client = FlowMqttClient(
    "my_client", 
    transport=TransportType.TLS,
    server_name="broker.emqx.io"
)
await client.connect("broker.emqx.io", 8883)

QUIC

client = FlowMqttClient(
    "my_client",
    transport=TransportType.QUIC,
    server_name="broker.emqx.io"
)
await client.connect("broker.emqx.io", 14567, server_name="broker.emqx.io")

Learning Path

  1. Start here: simple_async_usage.py - Learn the basics
  2. Next: asyncio_tcp_client_example.py - See full features
  3. Secure: asyncio_tls_client_example.py - Try TLS transport
  4. Modern: asyncio_quic_client_example.py - Try QUIC transport

License

Apache License 2.0

MQTT 5 options

mqtt5_options.py demonstrates property-bearing retained publishes, repeated user properties, Will messages, and persistent-session options:

PYTHONPATH=python/package python3 python/examples/mqtt5_options.py --host localhost --port 1883

Select TLS or QUIC with --transport tls --port 8883 or --transport quic --port 14567; use --ca for a private CA. Authentication can be provided through FLOWSDK_BROKER_USERNAME and FLOWSDK_BROKER_PASSWORD.

Live-broker regression tests

Start an MQTT 5 broker with permissions to publish and subscribe beneath flowsdk-python-test/. Tests use unique client IDs/topics and remove retained test messages. TCP is the default; TLS and QUIC need their respective listeners. Run from the repository root after generating bindings:

FLOWSDK_BROKER_HOST=localhost \
FLOWSDK_BROKER_TRANSPORTS=tcp,tls,quic \
PYTHONPATH=python/package python3 -W error -m unittest discover -s python/tests -p test_broker.py -v

Ports default to 1883/8883/14567; override with FLOWSDK_BROKER_TCP_PORT, FLOWSDK_BROKER_TLS_PORT, and FLOWSDK_BROKER_QUIC_PORT. Optional authentication variables are FLOWSDK_BROKER_USERNAME, FLOWSDK_BROKER_PASSWORD, FLOWSDK_BROKER_CA, FLOWSDK_BROKER_CERT, and FLOWSDK_BROKER_KEY. TLS certificate verification remains enabled.

These tests cover QoS 0/1/2 property round trips, retained messages, batch subscriptions, session resumption, and Will delivery after transport loss. They are skipped unless FLOWSDK_BROKER_HOST is set.