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, MqttTls, MqttWill
 17
 18__all__ = ["MqttClient", "MqttMessage", "MqttTls", "MqttWill", "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        max_packet_size: Optional[int] = None,
 59        username: Optional[str] = None,
 60        password: Optional[str] = None,
 61        will: Optional[MqttWill] = None,
 62        tls: Optional[MqttTls] = None,
 63    ) -> None:
 64        """Create a disconnected client from the given broker settings.
 65
 66        :param client_id: The MQTT client identifier presented to the broker.
 67        :param host: The broker hostname or IP address.
 68        :param port: The broker TCP port, conventionally 1883 for plaintext MQTT.
 69        :param keep_alive_secs: Keep-alive interval in seconds. Defaults to 30.
 70        :param capacity: Bound on outstanding client requests. Defaults to 64.
 71        :param qos: Default quality of service. Defaults to ``Qos.AT_LEAST_ONCE``.
 72        :param max_packet_size: The largest packet the connection sends or accepts, in
 73            bytes. Defaults to 10,240. A publish that would be larger is refused and the
 74            connection stays up, but a larger packet arriving from the broker ends the
 75            connection, so every client that shares a topic needs a limit that fits it.
 76        :param username: The name to sign in to the broker with.
 77        :param password: The password to sign in with, which needs a username. It travels
 78            in the clear unless the connection uses ``tls``.
 79        :param will: A message the broker publishes if the connection ends without a
 80            goodbye.
 81        :param tls: TLS settings; a connection with them is secured, conventionally on
 82            port 8883.
 83        :raises PamojaError: If a password comes without a username.
 84        """
 85        qos_value = qos.value if isinstance(qos, Qos) else qos
 86        self._native = _NativeMqttClient(
 87            client_id=client_id,
 88            host=host,
 89            port=port,
 90            keep_alive_secs=keep_alive_secs,
 91            capacity=capacity,
 92            qos=qos_value,
 93            max_packet_size=max_packet_size,
 94            username=username,
 95            password=password,
 96            will=will,
 97            tls=tls,
 98        )
 99
100    async def connect(self) -> None:
101        """Connect to the broker and start the background event loop.
102
103        :raises PamojaError: If the connection cannot be established.
104        """
105        await self._native.connect()
106
107    async def publish(
108        self,
109        topic: str,
110        payload: Union[str, bytes],
111        *,
112        qos: Optional[Qos] = None,
113        retain: bool = False,
114    ) -> None:
115        """Publish a payload to a topic, returning once it is queued for the broker.
116
117        This returns before the broker acknowledges the message;
118        :meth:`publish_confirmed` waits for that.
119
120        :param topic: The destination topic.
121        :param payload: The message body; ``str`` payloads are encoded as UTF-8.
122        :param qos: The quality of service for this message. Defaults to the client's.
123        :param retain: Whether the broker keeps the message for clients that subscribe
124            later. An empty retained message clears the one the broker holds.
125        """
126        data = payload.encode("utf-8") if isinstance(payload, str) else bytes(payload)
127        qos_value = qos.value if isinstance(qos, Qos) else qos
128        await self._native.publish(topic, data, qos=qos_value, retain=retain)
129
130    async def send(self, topic: str, payload: Union[str, bytes]) -> None:
131        """Publish a payload to a topic at the client's default quality of service.
132
133        This is the ``send`` every link has, so a profile's :class:`~pamoja.profile.Node`
134        or a :class:`~pamoja.profile.RuleEngine` publishes over an MQTT client the way it
135        publishes over any other link.
136
137        :param topic: The destination topic.
138        :param payload: The message body; ``str`` payloads are encoded as UTF-8.
139        """
140        await self.publish(topic, payload)
141
142    async def publish_confirmed(
143        self,
144        topic: str,
145        payload: Union[str, bytes],
146        *,
147        qos: Optional[Qos] = None,
148        retain: bool = False,
149    ) -> None:
150        """Publish a payload to a topic and wait for the broker to acknowledge it.
151
152        The broker answers with a ``PUBACK`` at ``Qos.AT_LEAST_ONCE`` and a ``PUBCOMP``
153        at ``Qos.EXACTLY_ONCE``; at ``Qos.AT_MOST_ONCE`` MQTT acknowledges nothing, so
154        this returns once the connection has taken the message.
155
156        :param topic: The destination topic.
157        :param payload: The message body; ``str`` payloads are encoded as UTF-8.
158        :param qos: The quality of service for this message. Defaults to the client's.
159        :param retain: Whether the broker keeps the message for clients that subscribe
160            later.
161        :raises PamojaError: If the connection ends before the acknowledgment, when the
162            message may or may not have arrived.
163        """
164        data = payload.encode("utf-8") if isinstance(payload, str) else bytes(payload)
165        qos_value = qos.value if isinstance(qos, Qos) else qos
166        await self._native.publish_confirmed(topic, data, qos=qos_value, retain=retain)
167
168    async def subscribe(self, topic: str) -> None:
169        """Subscribe to a topic filter.
170
171        :param topic: The topic or wildcard filter to subscribe to.
172        """
173        await self._native.subscribe(topic)
174
175    async def recv(self) -> Optional[MqttMessage]:
176        """Await the next message from any subscribed topic.
177
178        Wrap it in :func:`asyncio.wait_for` to stop waiting; the receive it cancels
179        leaves its message queued for the next one.
180
181        :returns: The next message, or ``None`` once the connection has ended.
182        :raises PamojaError: Once, saying why, when the connection ended on its own,
183            because the broker went away or another client connected with the same id.
184        """
185        return await self._native.recv()
186
187    async def is_connected(self) -> bool:
188        """Report whether the client currently holds an active connection."""
189        return await self._native.is_connected()
190
191    async def disconnect(self) -> None:
192        """Close the connection and stop the background event loop."""
193        await self._native.disconnect()
194
195    async def messages(self) -> AsyncIterator[MqttMessage]:
196        """Yield messages from subscribed topics until the connection ends.
197
198        A connection that ends on its own raises its reason out of the loop.
199        """
200        while True:
201            message = await self._native.recv()
202            if message is None:
203                return
204            yield message
205
206    def __aiter__(self) -> AsyncIterator[MqttMessage]:
207        """Iterate incoming messages, so a client can be used with ``async for``."""
208        return self.messages()
209
210    async def __aenter__(self) -> "MqttClient":
211        await self.connect()
212        return self
213
214    async def __aexit__(self, *_exc: object) -> None:
215        await self.disconnect()