Skip to content

MQTTParserV311

faststream.mqtt.parser.MQTTParserV311 #

Bases: MQTTBaseParser

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

parse_message async #

parse_message(msg: Message) -> MQTTMessage
Source code in faststream/mqtt/parser.py 
28
29
30
31
32
33
34
35
36
async def parse_message(self, msg: zmqtt.Message) -> MQTTMessage:
    return MQTTMessage(
        raw_message=msg,
        body=msg.payload,
        headers={},
        content_type=None,
        reply_to="",
        correlation_id=None,
    )

decode_message async #

decode_message(msg: StreamMessage[Any]) -> DecodedMessage
Source code in faststream/mqtt/parser.py 
38
39
40
41
42
43
44
45
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