Skip to content

FakeNatsFastProducer

faststream.nats.publisher.producer.FakeNatsFastProducer #

Bases: NatsFastProducer

connect #

connect(
    connection: Any, serializer: Optional[SerializerProto]
) -> None
Source code in faststream/nats/publisher/producer.py 
208
209
def connect(self, connection: Any, serializer: Optional["SerializerProto"]) -> None:
    raise NotImplementedError

disconnect #

disconnect() -> None
Source code in faststream/nats/publisher/producer.py 
211
212
def disconnect(self) -> None:
    raise NotImplementedError

publish async #

publish(cmd: NatsPublishCommand) -> None
Source code in faststream/nats/publisher/producer.py 
214
215
216
@override
async def publish(self, cmd: "NatsPublishCommand") -> None:
    raise NotImplementedError

request async #

request(cmd: NatsPublishCommand) -> Msg
Source code in faststream/nats/publisher/producer.py 
218
219
220
@override
async def request(self, cmd: "NatsPublishCommand") -> "Msg":
    raise NotImplementedError

publish_batch async #

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