Skip to content

MQTTMetricsSettingsProvider

faststream.mqtt.prometheus.provider.MQTTMetricsSettingsProvider #

MQTTMetricsSettingsProvider()

Bases: MetricsSettingsProvider[Message, MQTTPublishCommand]

Source code in faststream/mqtt/prometheus/provider.py 
17
18
def __init__(self) -> None:
    self.messaging_system = "mqtt"

messaging_system instance-attribute #

messaging_system = 'mqtt'

get_consume_attrs_from_message #

get_consume_attrs_from_message(
    msg: StreamMessage[Message],
) -> ConsumeAttrs
Source code in faststream/mqtt/prometheus/provider.py 
20
21
22
23
24
25
26
27
28
def get_consume_attrs_from_message(
    self,
    msg: "StreamMessage[zmqtt.Message]",
) -> ConsumeAttrs:
    return {
        "destination_name": msg.raw_message.topic,
        "message_size": len(msg.body),
        "messages_count": 1,
    }

get_publish_destination_name_from_cmd #

get_publish_destination_name_from_cmd(
    cmd: MQTTPublishCommand,
) -> str
Source code in faststream/mqtt/prometheus/provider.py 
30
31
32
33
34
def get_publish_destination_name_from_cmd(
    self,
    cmd: MQTTPublishCommand,
) -> str:
    return cmd.destination