Skip to content

ZmqttProducerV311

faststream.mqtt.publisher.producer.ZmqttProducerV311 #

ZmqttProducerV311(
    parser: Optional[CustomCallable],
    decoder: Optional[CustomCallable],
    id_generator: IdGenerator = gen_cor_id,
)

Bases: ZmqttBaseProducer

Producer for MQTT 3.1.1 — publishes raw bytes only.

Headers, correlation_id, and other metadata are not supported. Use MQTT 5.0 for those features. Request/reply is supported via an explicit reply_to topic provided by the caller.

Source code in faststream/mqtt/publisher/producer.py 
 97
 98
 99
100
101
102
103
def __init__(
    self,
    parser: Optional["CustomCallable"],
    decoder: Optional["CustomCallable"],
    id_generator: IdGenerator = gen_cor_id,
) -> None:
    super().__init__(MQTTParserV311(), parser, decoder, id_generator)

codec instance-attribute #

codec: CodecProto = DefaultCodec()

serializer instance-attribute #

serializer: SerializerProto | None = None

id_generator instance-attribute #

id_generator = id_generator

publish async #

publish(cmd: MQTTPublishCommand) -> None
Source code in faststream/mqtt/publisher/producer.py 
105
106
107
108
109
110
111
112
113
114
115
116
@override
async def publish(self, cmd: "MQTTPublishCommand") -> None:
    if cmd.headers:
        msg = "MQTT 3.1.1 does not support message headers. Use MQTT 5.0."
        raise FeatureNotSupportedException(msg)
    payload, _ = await self.codec.encode(cmd.body, self.serializer)
    await self._connected_client.publish(
        cmd.destination,
        payload,
        qos=zmqtt.QoS(cmd.qos),
        retain=cmd.retain,
    )

request async #

request(cmd: MQTTPublishCommand) -> Message

Request/reply for MQTT 3.1.1 via explicit reply topic.

The caller must supply cmd.reply_to. FastStream subscribes to that topic, publishes the raw request payload, then waits for the first message on the reply topic. The handler side must publish its response to the same topic (e.g. via @broker.publisher).

Source code in faststream/mqtt/publisher/producer.py 
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
@override
async def request(self, cmd: "MQTTPublishCommand") -> "zmqtt.Message":
    """Request/reply for MQTT 3.1.1 via explicit reply topic.

    The caller must supply ``cmd.reply_to``.  FastStream subscribes to
    that topic, publishes the raw request payload, then waits for the
    first message on the reply topic.  The handler side must publish
    its response to the same topic (e.g. via ``@broker.publisher``).
    """
    if not cmd.reply_to:
        msg = "MQTT 3.1.1 request() requires an explicit reply_to topic."
        raise FeatureNotSupportedException(msg)

    sub = self._connected_client.subscribe(cmd.reply_to)
    await sub.start()

    try:
        payload, _ = await self.codec.encode(cmd.body, self.serializer)
        await self._connected_client.publish(
            cmd.destination,
            payload,
            qos=cmd.qos,
            retain=cmd.retain,
        )
        return await asyncio.wait_for(
            sub.get_message(),
            timeout=cmd.timeout or 30.0,
        )
    finally:
        await sub.stop()

publish_batch async #

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

connect #

connect(
    client: MQTTClient,
    serializer: Optional[SerializerProto],
    codec: Optional[CodecProto] = None,
) -> None
Source code in faststream/mqtt/publisher/producer.py 
52
53
54
55
56
57
58
59
60
def connect(
    self,
    client: "zmqtt.MQTTClient",
    serializer: Optional["SerializerProto"],
    codec: Optional["CodecProto"] = None,
) -> None:
    self._client = client
    self.serializer = serializer
    self.codec = codec or DefaultCodec()

disconnect #

disconnect() -> None
Source code in faststream/mqtt/publisher/producer.py 
62
63
64
def disconnect(self) -> None:
    self._client = None
    self.serializer = None