Skip to content

MQTTBaseParser

faststream.mqtt.parser.MQTTBaseParser #

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

Base parser for MQTT messages — shared parse + decode logic.

Source code in faststream/mqtt/parser.py 
19
20
21
22
23
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 
33
34
async def parse_message(self, msg: zmqtt.Message) -> MQTTMessage:
    raise NotImplementedError

decode_message async #

decode_message(msg: StreamMessage[Any]) -> DecodedMessage
Source code in faststream/mqtt/parser.py 
36
37
async def decode_message(self, msg: "StreamMessage[Any]") -> "DecodedMessage":
    return decode_message(msg)