Skip to content

MQTTFakePublisher

faststream.mqtt.publisher.fake.MQTTFakePublisher #

MQTTFakePublisher(
    producer: ProducerProto[MQTTPublishCommand], topic: str
)

Bases: FakePublisher

Publisher used for RPC / reply-to responses in MQTT.

Source code in faststream/mqtt/publisher/fake.py 
14
15
16
17
18
19
20
def __init__(
    self,
    producer: "ProducerProto[MQTTPublishCommand]",
    topic: str,
) -> None:
    super().__init__(producer=producer)
    self.topic = topic

topic instance-attribute #

topic = topic

patch_command #

patch_command(
    cmd: Union[PublishCommand, MQTTPublishCommand],
) -> MQTTPublishCommand
Source code in faststream/mqtt/publisher/fake.py 
22
23
24
25
26
27
28
29
def patch_command(
    self,
    cmd: Union["PublishCommand", "MQTTPublishCommand"],
) -> "MQTTPublishCommand":
    cmd = super().patch_command(cmd)
    real_cmd = MQTTPublishCommand.from_cmd(cmd)
    real_cmd.destination = self.topic
    return real_cmd

publish async #

publish(
    message: SendableMessage,
    /,
    *,
    correlation_id: str | None = None,
) -> Any | None
Source code in faststream/_internal/endpoint/publisher/fake.py 
49
50
51
52
53
54
55
56
57
58
59
60
async def publish(
    self,
    message: SendableMessage,
    /,
    *,
    correlation_id: str | None = None,
) -> Any | None:
    msg = (
        f"`{self.__class__.__name__}` can be used only to publish "
        "a response for `reply-to` or `RPC` messages."
    )
    raise NotImplementedError(msg)

request async #

request(
    message: SendableMessage,
    /,
    *,
    correlation_id: str | None = None,
) -> Any
Source code in faststream/_internal/endpoint/publisher/fake.py 
62
63
64
65
66
67
68
69
70
71
72
73
async def request(
    self,
    message: "SendableMessage",
    /,
    *,
    correlation_id: str | None = None,
) -> Any:
    msg = (
        f"`{self.__class__.__name__}` can be used only to publish "
        "a response for `reply-to` or `RPC` messages."
    )
    raise NotImplementedError(msg)