Skip to content

ZmqttBaseProducer

faststream.mqtt.publisher.producer.ZmqttBaseProducer #

ZmqttBaseProducer(
    default_parser: Any,
    parser: Optional[CustomCallable],
    decoder: Optional[CustomCallable],
)

Bases: ProducerProto[MQTTPublishCommand]

Source code in faststream/mqtt/publisher/producer.py 
25
26
27
28
29
30
31
32
33
34
35
def __init__(
    self,
    default_parser: Any,
    parser: Optional["CustomCallable"],
    decoder: Optional["CustomCallable"],
) -> None:
    self.serializer: SerializerProto | None = None
    self._client: zmqtt.MQTTClient | None = None

    self._parser = ParserComposition(parser, default_parser.parse_message)
    self._decoder = ParserComposition(decoder, default_parser.decode_message)

serializer instance-attribute #

serializer: SerializerProto | None = None

connect #

connect(
    client: MQTTClient,
    serializer: Optional[SerializerProto],
) -> None
Source code in faststream/mqtt/publisher/producer.py 
37
38
39
40
41
42
43
def connect(
    self,
    client: "zmqtt.MQTTClient",
    serializer: Optional["SerializerProto"],
) -> None:
    self._client = client
    self.serializer = serializer

disconnect #

disconnect() -> None
Source code in faststream/mqtt/publisher/producer.py 
45
46
47
def disconnect(self) -> None:
    self._client = None
    self.serializer = None

publish async #

publish(cmd: MQTTPublishCommand) -> None
Source code in faststream/mqtt/publisher/producer.py 
56
57
58
@override
async def publish(self, cmd: "MQTTPublishCommand") -> None:
    raise NotImplementedError

request async #

request(cmd: MQTTPublishCommand) -> Any
Source code in faststream/mqtt/publisher/producer.py 
60
61
62
@override
async def request(self, cmd: "MQTTPublishCommand") -> Any:
    raise NotImplementedError

publish_batch async #

publish_batch(cmd: MQTTPublishCommand) -> None
Source code in faststream/mqtt/publisher/producer.py 
64
65
66
67
@override
async def publish_batch(self, cmd: "MQTTPublishCommand") -> None:
    msg = "MQTT does not support batch publishing."
    raise FeatureNotSupportedException(msg)