Skip to content

FakeProducer

faststream.mqtt.testing.FakeProducer #

FakeProducer(
    broker: MQTTBroker, brokers: Sequence[MQTTBroker]
)

Bases: ZmqttBaseProducer

In-memory producer that routes messages directly to matching subscribers.

Encodes messages in the wire format matching the broker's configured MQTT version: V311 envelope for 3.1.1, PublishProperties for 5.0.

Source code in faststream/mqtt/testing.py 
179
180
181
182
183
184
185
186
187
188
189
190
191
192
def __init__(
    self,
    broker: MQTTBroker,
    brokers: Sequence[MQTTBroker],
) -> None:
    self.broker = broker
    self.brokers = brokers
    self.serializer: SerializerProto | None = None

    version = _broker_version(broker)
    default = _parser_for_version(version)
    self._parser = ParserComposition(broker._parser, default.parse_message)
    self._decoder = ParserComposition(broker._decoder, default.decode_message)
    self.codec = broker.config.broker_codec or DefaultCodec()

broker instance-attribute #

broker = broker

brokers instance-attribute #

brokers = brokers

serializer instance-attribute #

serializer: SerializerProto | None = None

codec instance-attribute #

codec = broker.config.broker_codec or DefaultCodec()

subscribers property #

subscribers: Iterable[MQTTBaseSubscriber]

publish async #

publish(cmd: MQTTPublishCommand) -> None
Source code in faststream/mqtt/testing.py 
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
@override
async def publish(self, cmd: MQTTPublishCommand) -> None:
    msg = await build_message(
        message=cmd.body,
        topic=cmd.destination,
        version=self._version,
        qos=cmd.qos,
        retain=cmd.retain,
        reply_to=cmd.reply_to,
        correlation_id=cmd.correlation_id,
        headers=cmd.headers,
        serializer=self.broker.config.fd_config._serializer,
        codec=self.codec,
    )

    # For shared subscriptions, only deliver to one subscriber per group
    seen_shared_groups: set[str] = set()

    for handler in self.subscribers:
        handler_topic = handler.topic
        if not mqtt_topic_matches(handler_topic, cmd.destination):
            continue

        if handler_topic.startswith("$share/"):
            _, group, _ = handler_topic.split("/", 2)
            if group in seen_shared_groups:
                continue
            seen_shared_groups.add(group)

        await handler.process_message(msg)

request async #

request(cmd: MQTTPublishCommand) -> Message
Source code in faststream/mqtt/testing.py 
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
@override
async def request(self, cmd: MQTTPublishCommand) -> "zmqtt.Message":
    msg = await build_message(
        message=cmd.body,
        topic=cmd.destination,
        version=self._version,
        qos=cmd.qos,
        retain=cmd.retain,
        correlation_id=cmd.correlation_id,
        headers=cmd.headers,
        serializer=self.broker.config.fd_config._serializer,
        codec=self.codec,
    )

    for handler in self.subscribers:
        if not mqtt_topic_matches(handler.topic, cmd.destination):
            continue

        with anyio.fail_after(cmd.timeout or 30.0):
            result = await handler.process_message(msg)

        return await build_message(
            message=result.body,
            topic=cmd.destination,
            version=self._version,
            correlation_id=result.correlation_id,
            headers=result.headers,
            serializer=self.broker.config.fd_config._serializer,
            codec=self.codec,
        )

    raise SubscriberNotFound

publish_batch async #

publish_batch(cmd: MQTTPublishCommand) -> None
Source code in faststream/mqtt/publisher/producer.py 
69
70
71
72
@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 
40
41
42
43
44
45
46
47
48
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 
50
51
52
def disconnect(self) -> None:
    self._client = None
    self.serializer = None