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 
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 
40
41
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 
43
44
async def decode_message(self, msg: "StreamMessage[Any]") -> "DecodedMessage":  # noqa: PLR6301
    return decode_message(msg)