Skip to content

FakeProducer

faststream.nats.testing.FakeProducer #

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

Bases: NatsFastProducer

Source code in faststream/nats/testing.py 
142
143
144
145
146
147
148
149
150
151
152
153
def __init__(
    self,
    broker: NatsBroker,
    brokers: Sequence[NatsBroker],
) -> None:
    self.broker = broker
    self.brokers = brokers

    default = NatsParser(pattern="", is_ack_disabled=True)
    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

codec instance-attribute #

codec = broker.config.broker_codec or DefaultCodec()

subscribers property #

subscribers: Iterable[LogicSubscriber[Any]]

publish async #

publish(cmd: NatsPublishCommand) -> None
Source code in faststream/nats/testing.py 
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
@override
async def publish(self, cmd: "NatsPublishCommand") -> None:
    incoming = await build_message(
        message=cmd.body,
        subject=cmd.destination,
        headers=cmd.headers,
        correlation_id=cmd.correlation_id,
        reply_to=cmd.reply_to,
        serializer=self.broker.config.fd_config._serializer,
        codec=self.codec,
        id_generator=self.broker.config.id_generator,
    )

    for handler in _find_handler(
        self.subscribers,
        cmd.destination,
        cmd.stream,
    ):
        msg: list[PatchedMessage] | PatchedMessage

        if (pull := getattr(handler, "pull_sub", None)) and pull.batch:
            msg = [incoming]
        else:
            msg = incoming

        await self._execute_handler(msg, cmd.destination, handler)

request async #

request(cmd: NatsPublishCommand) -> PatchedMessage
Source code in faststream/nats/testing.py 
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
@override
async def request(self, cmd: "NatsPublishCommand") -> "PatchedMessage":
    incoming = await build_message(
        message=cmd.body,
        subject=cmd.destination,
        headers=cmd.headers,
        correlation_id=cmd.correlation_id,
        serializer=self.broker.config.fd_config._serializer,
        codec=self.codec,
        id_generator=self.broker.config.id_generator,
    )

    for handler in _find_handler(
        self.subscribers,
        cmd.destination,
        cmd.stream,
    ):
        msg: list[PatchedMessage] | PatchedMessage

        if (pull := getattr(handler, "pull_sub", None)) and pull.batch:
            msg = [incoming]
        else:
            msg = incoming

        with anyio.fail_after(cmd.timeout):
            return await self._execute_handler(msg, cmd.destination, handler)

    raise SubscriberNotFound

publish_batch async #

publish_batch(cmd: NatsPublishCommand) -> None
Source code in faststream/nats/publisher/producer.py 
56
57
58
59
@override
async def publish_batch(self, cmd: "NatsPublishCommand") -> None:
    msg = "NATS doesn't support publishing in batches."
    raise FeatureNotSupportedException(msg)

connect #

connect(
    connection: Any,
    serializer: Optional[SerializerProto],
    codec: Optional[CodecProto] = None,
) -> None
Source code in faststream/nats/publisher/producer.py 
41
42
43
44
45
46
def connect(
    self,
    connection: Any,
    serializer: Optional["SerializerProto"],
    codec: Optional["CodecProto"] = None,
) -> None: ...

disconnect #

disconnect() -> None
Source code in faststream/nats/publisher/producer.py 
48
def disconnect(self) -> None: ...