Skip to content

MQTTParserV311

faststream.mqtt.parser.MQTTParserV311 #

MQTTParserV311(path_regex: Pattern[str] | None = None)

Bases: MQTTBaseParser

Parser for MQTT 3.1.1 messages — raw payload, no metadata.

Source code in faststream/mqtt/parser.py 
26
27
28
29
30
def __init__(
    self,
    path_regex: Pattern[str] | None = None,
) -> None:
    self._path_regex = path_regex

parse_message async #

parse_message(msg: Message) -> MQTTMessage
Source code in faststream/mqtt/parser.py 
52
53
54
55
56
57
58
59
60
61
async def parse_message(self, msg: zmqtt.Message) -> MQTTMessage:
    return MQTTMessage(
        raw_message=msg,
        body=msg.payload,
        headers={},
        path=self._extract_path(msg.topic),
        content_type=None,
        reply_to="",
        correlation_id=None,
    )

decode_message async #

decode_message(msg: StreamMessage[Any]) -> DecodedMessage
Source code in faststream/mqtt/parser.py 
63
64
65
66
67
68
69
70
71
@override
async def decode_message(self, msg: "StreamMessage[Any]") -> "DecodedMessage":
    body: bytes = msg.body
    with suppress(Exception):
        m: DecodedMessage = json_loads(body)
        return m
    with suppress(UnicodeDecodeError):
        return body.decode()
    return body