Bases: AioPikaFastProducer
A fake RabbitMQ producer for testing purposes.
This class extends AioPikaFastProducer and is used to simulate RabbitMQ message publishing during tests.
Source code in faststream/rabbit/testing.py
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259 | def __init__(
self,
broker: RabbitBroker,
brokers: Sequence[RabbitBroker],
) -> None:
self.broker = broker
self.brokers = brokers
default_parser = AioPikaParser()
self._parser = ParserComposition(broker._parser, default_parser.parse_message)
self._decoder = ParserComposition(
broker._decoder,
default_parser.decode_message,
)
self.codec = broker.config.broker_codec or DefaultCodec()
|
instance-attribute
instance-attribute
instance-attribute
codec = broker.config.broker_codec or DefaultCodec()
property
subscribers: Iterable[SubscriberUsecase[Any]]
async
Publish a message to a RabbitMQ queue or exchange.
Source code in faststream/rabbit/testing.py
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297 | @override
async def publish(
self,
cmd: "RabbitPublishCommand",
) -> None:
"""Publish a message to a RabbitMQ queue or exchange."""
incoming = await build_message(
message=cmd.body,
exchange=cmd.exchange,
routing_key=cmd.destination,
correlation_id=cmd.correlation_id,
headers=cmd.headers,
reply_to=cmd.reply_to,
serializer=self.broker.config.fd_config._serializer,
codec=self.codec,
id_generator=self.broker.config.id_generator,
**cmd.message_options,
)
called = False
for handler in self.subscribers: # pragma: no branch
handler = cast("RabbitSubscriber", handler)
if _is_handler_matches(
handler,
incoming.routing_key,
incoming.headers,
cmd.exchange,
):
called = True
await self._execute_handler(incoming, handler)
if not called:
raise SubscriberNotFound
|
async
Make a synchronous request to RabbitMQ.
Source code in faststream/rabbit/testing.py
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328 | @override
async def request(
self,
cmd: "RabbitPublishCommand",
) -> "PatchedMessage":
"""Make a synchronous request to RabbitMQ."""
incoming = await build_message(
message=cmd.body,
exchange=cmd.exchange,
routing_key=cmd.destination,
correlation_id=cmd.correlation_id,
headers=cmd.headers,
serializer=self.broker.config.fd_config._serializer,
codec=self.codec,
id_generator=self.broker.config.id_generator,
**cmd.message_options,
)
for handler in self.subscribers: # pragma: no branch
handler = cast("RabbitSubscriber", handler)
if _is_handler_matches(
handler,
incoming.routing_key,
incoming.headers,
cmd.exchange,
):
with anyio.fail_after(cmd.timeout):
return await self._execute_handler(incoming, handler)
raise SubscriberNotFound
|
async
Source code in faststream/rabbit/publisher/producer.py
| @override
async def publish_batch(self, cmd: "RabbitPublishCommand") -> None:
msg = "RabbitMQ doesn't support publishing in batches."
raise FeatureNotSupportedException(msg)
|
connect(
serializer: Optional[SerializerProto] = None,
codec: Optional[CodecProto] = None,
) -> None
Source code in faststream/rabbit/publisher/producer.py
| def connect(
self,
serializer: Optional["SerializerProto"] = None,
codec: Optional["CodecProto"] = None,
) -> None: ...
|
Source code in faststream/rabbit/publisher/producer.py
| def disconnect(self) -> None: ...
|