FakeNatsFastProducer

faststream.nats.publisher.producer.FakeNatsFastProducer

Bases: NatsFastProducer

connect

connect(connection, serializer)

Source code in faststream/nats/publisher/producer.py

def connect(self, connection: Any, serializer: Optional["SerializerProto"]) -> None:
    raise NotImplementedError

disconnect

disconnect()

Source code in faststream/nats/publisher/producer.py

def disconnect(self) -> None:
    raise NotImplementedError

publish async

publish(cmd)

Source code in faststream/nats/publisher/producer.py

@override
async def publish(self, cmd: "NatsPublishCommand") -> None:
    raise NotImplementedError

request async

request(cmd)

Source code in faststream/nats/publisher/producer.py

@override
async def request(self, cmd: "NatsPublishCommand") -> "Msg":
    raise NotImplementedError

publish_batch async

publish_batch(cmd)

Source code in faststream/nats/publisher/producer.py

@override
async def publish_batch(self, cmd: "NatsPublishCommand") -> None:
    msg = "NATS doesn't support publishing in batches."
    raise FeatureNotSupportedException(msg)