pamoja.mqtt

  1"""Idiomatic async MQTT client facade.
  2
  3Wraps the native :class:`pamoja._native.MqttClient` with a Python-native surface:
  4awaitable methods, ``async for`` iteration over messages, ``async with``
  5lifecycle management, a string enum for quality of service, and keyword
  6construction. It adds ergonomics only; every operation delegates to the Rust
  7core.
  8"""
  9
 10from __future__ import annotations
 11
 12import enum
 13from typing import AsyncIterator, Optional, Union
 14
 15from pamoja._native import MqttClient as _NativeMqttClient
 16from pamoja._native import MqttMessage
 17
 18__all__ = ["MqttClient", "MqttMessage", "Qos"]
 19
 20
 21class Qos(str, enum.Enum):
 22    """MQTT delivery guarantee, mirroring the protocol's quality-of-service levels."""
 23
 24    #: Fire and forget; the broker does not acknowledge delivery.
 25    AT_MOST_ONCE = "AtMostOnce"
 26    #: Delivered at least once and acknowledged.
 27    AT_LEAST_ONCE = "AtLeastOnce"
 28    #: Delivered exactly once via a four-step handshake.
 29    EXACTLY_ONCE = "ExactlyOnce"
 30
 31
 32class MqttClient:
 33    """An MQTT client transport.
 34
 35    Construct it with broker settings, :meth:`connect`, then :meth:`publish`,
 36    :meth:`subscribe`, and read inbound messages with :meth:`recv` or by
 37    iterating the client with ``async for``. The client also works as an async
 38    context manager, connecting on entry and disconnecting on exit.
 39
 40    Example::
 41
 42        async with MqttClient(client_id="sensor-1", host="localhost", port=1883) as client:
 43            await client.subscribe("sensors/+/temperature")
 44            await client.publish("sensors/1/temperature", "21.5")
 45            async for message in client:
 46                print(message.topic, message.payload.decode())
 47    """
 48
 49    def __init__(
 50        self,
 51        *,
 52        client_id: str,
 53        host: str,
 54        port: int,
 55        keep_alive_secs: Optional[int] = None,
 56        capacity: Optional[int] = None,
 57        qos: Optional[Qos] = None,
 58    ) -> None:
 59        """Create a disconnected client from the given broker settings.
 60
 61        :param client_id: The MQTT client identifier presented to the broker.
 62        :param host: The broker hostname or IP address.
 63        :param port: The broker TCP port, conventionally 1883 for plaintext MQTT.
 64        :param keep_alive_secs: Keep-alive interval in seconds. Defaults to 30.
 65        :param capacity: Bound on outstanding client requests. Defaults to 64.
 66        :param qos: Default quality of service. Defaults to ``Qos.AT_LEAST_ONCE``.
 67        """
 68        qos_value = qos.value if isinstance(qos, Qos) else qos
 69        self._native = _NativeMqttClient(
 70            client_id=client_id,
 71            host=host,
 72            port=port,
 73            keep_alive_secs=keep_alive_secs,
 74            capacity=capacity,
 75            qos=qos_value,
 76        )
 77
 78    async def connect(self) -> None:
 79        """Connect to the broker and start the background event loop.
 80
 81        :raises PamojaError: If the connection cannot be established.
 82        """
 83        await self._native.connect()
 84
 85    async def publish(self, topic: str, payload: Union[str, bytes]) -> None:
 86        """Publish a payload to a topic.
 87
 88        :param topic: The destination topic.
 89        :param payload: The message body; ``str`` payloads are encoded as UTF-8.
 90        """
 91        data = payload.encode("utf-8") if isinstance(payload, str) else bytes(payload)
 92        await self._native.publish(topic, data)
 93
 94    async def subscribe(self, topic: str) -> None:
 95        """Subscribe to a topic filter.
 96
 97        :param topic: The topic or wildcard filter to subscribe to.
 98        """
 99        await self._native.subscribe(topic)
100
101    async def recv(self) -> Optional[MqttMessage]:
102        """Await the next message from any subscribed topic.
103
104        :returns: The next message, or ``None`` once the connection has ended.
105        """
106        return await self._native.recv()
107
108    async def is_connected(self) -> bool:
109        """Report whether the client currently holds an active connection."""
110        return await self._native.is_connected()
111
112    async def disconnect(self) -> None:
113        """Close the connection and stop the background event loop."""
114        await self._native.disconnect()
115
116    async def messages(self) -> AsyncIterator[MqttMessage]:
117        """Yield messages from subscribed topics until the connection ends."""
118        while True:
119            message = await self._native.recv()
120            if message is None:
121                return
122            yield message
123
124    def __aiter__(self) -> AsyncIterator[MqttMessage]:
125        """Iterate incoming messages, so a client can be used with ``async for``."""
126        return self.messages()
127
128    async def __aenter__(self) -> "MqttClient":
129        await self.connect()
130        return self
131
132    async def __aexit__(self, *_exc: object) -> None:
133        await self.disconnect()