Skip to content

NatsFakePublisher

faststream.nats.publisher.fake.NatsFakePublisher #

NatsFakePublisher(producer, subject)

Bases: FakePublisher

Publisher Interface implementation to use as RPC or REPLY TO answer publisher.

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

subject instance-attribute #

subject = subject

patch_command #

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

publish async #

publish(message, /, *, correlation_id=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, /, *, correlation_id=None)
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)