Example

Tool

The testing tools use mosquitto_sub/mosquitto_pub. You may also use MQTTX or test scripts for this purpose.

# Subscribe test topic
mosquitto_sub -h test.mosquitto.org -p 1884 -u rw -P readwrite -t '/cassia/test/up'

# Publish test message
mosquitto_pub -h test.mosquitto.org -p 1884 -u rw -P readwrite -t '/cassia/test/down' -m '{"hello":"world"}'

Publish

import asyncio

from cassiamqtt import CassiaMQTTClient


async def main():
    uri = "mqtt://test.mosquitto.org:1884"
    username = "rw"
    password = "readwrite"
    client_id = "puber"
    topic = "/cassia/test/up"

    async with CassiaMQTTClient(
        uri,
        username=username,
        password=password,
        client_id=client_id,
    ) as client:
        print("connect ok")

        payload = "hello"
        qos = 1
        retain = False
        await client.publish(topic, payload, qos=qos, retain=retain)

        print("pub ok")


asyncio.run(main())

Subscribe

import asyncio

from cassiamqtt import CassiaMQTTClient


async def main():
    uri = "mqtt://test.mosquitto.org:1884"
    username = "rw"
    password = "readwrite"
    client_id = "puber"
    topic = "/cassia/test/down"

    async with CassiaMQTTClient(
        uri,
        username=username,
        password=password,
        client_id=client_id,
    ) as client:
        print("connect ok")

        qos = 1
        await client.subscribe(topic, qos=qos)
        print("sub ok")

        async for msg in client:
            print("recv:", msg)

            for k, v in msg.items():
                print(k, type(k), v, type(v))


asyncio.run(main())

Reconnect

import asyncio
import json
import gc

import cassiamqtt


try:
    from typing import Optional, Tuple
except ImportError:
    pass


class CassiaMqttWrapper:
    DEFAULT_TOPIC_SUB = "/cassia/test/down"
    DEFAULT_TOPIC_PUB = "/cassia/test/up"
    DEFAULT_QOS = 1

    def __init__(
        self,
        uri: str,
        username: Optional[str] = None,
        password: Optional[str] = None,
        client_id: Optional[str] = None,
    ):
        self._uri = uri
        self._user = username
        self._pwd = password
        self._cid = client_id
        self._client: Optional[cassiamqtt.CassiaMQTTClient] = None

    async def _subscribe_topic(self) -> Tuple[bool, str]:
        if not self._client:
            return False, "client not connected"
        return await self._client.subscribe(self.DEFAULT_TOPIC_SUB, self.DEFAULT_QOS)

    async def _handle_messages(self) -> None:
        print("[MQTT] Waiting for messages...")
        async for msg in self._client:
            print("[MQTT] Received:", msg)

    async def runner(self) -> None:
        while True:
            try:
                print("[MQTT] Connecting...")
                async with cassiamqtt.CassiaMQTTClient(
                    self._uri, self._user, self._pwd, self._cid
                ) as client:
                    print("[MQTT] Connected")
                    self._client = client

                    ok, ret = await self._subscribe_topic()
                    if not ok:
                        print(f"[MQTT] Subscribe failed: {ret}, retry in 5s...")
                        self._client = None
                        await asyncio.sleep(5)
                        continue

                    print("[MQTT] Subscribe success")
                    await self._handle_messages()

            except Exception as e:
                print(f"[MQTT] Connection error: {e}")

            finally:
                print("[MQTT] Disconnected, retrying in 5s...")
                self._client = None
                await asyncio.sleep(5)

    async def publish(
        self, topic: str, payload: str, qos: int = 0, retain: bool = False
    ) -> Tuple[bool, str]:
        if not self._client:
            return False, "no client"
        return await self._client.publish(topic, payload, qos, retain)

    async def _mem_monitor(self) -> None:
        while True:
            data = {
                "free": gc.mem_free(),
                "alloc": gc.mem_alloc(),
            }
            print(f"[GC] memory: {data}")

            msg = json.dumps(data)
            await self.publish(self.DEFAULT_TOPIC_PUB, msg, qos=self.DEFAULT_QOS)

            await asyncio.sleep(30)


async def main() -> None:
    uri = "mqtt://test.mosquitto.org:1884"
    username = "rw"
    password = "readwrite"
    client_id = "CC:1B:E0:00:00:02"

    mqtt = CassiaMqttWrapper(
        uri,
        username=username,
        password=password,
        client_id=client_id,
    )

    await asyncio.gather(
        mqtt.runner(),
        mqtt._mem_monitor(),
    )


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