Skip to content

NatsFastProducerImpl

faststream.nats.publisher.producer.NatsFastProducerImpl #

NatsFastProducerImpl(parser, decoder)

Bases: NatsFastProducer

A class to represent a NATS producer.

Source code in faststream/nats/publisher/producer.py 
60
61
62
63
64
65
66
67
68
69
70
71
def __init__(
    self,
    parser: Optional["CustomCallable"],
    decoder: Optional["CustomCallable"],
) -> None:
    self.serializer: SerializerProto | None = None

    default = NatsParser(pattern="", is_ack_disabled=True)
    self._parser = ParserComposition(parser, default.parse_message)
    self._decoder = ParserComposition(decoder, default.decode_message)

    self.__state: ConnectionState[Client] = EmptyConnectionState()

serializer instance-attribute #

serializer = None

connect #

connect(connection, serializer)
Source code in faststream/nats/publisher/producer.py 
73
74
75
76
77
78
79
def connect(
    self,
    connection: "Client",
    serializer: Optional["SerializerProto"],
) -> None:
    self.serializer = serializer
    self.__state = ConnectedState(connection)

disconnect #

disconnect()
Source code in faststream/nats/publisher/producer.py 
81
82
def disconnect(self) -> None:
    self.__state = EmptyConnectionState()

publish async #

publish(cmd)
Source code in faststream/nats/publisher/producer.py 
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
@override
async def publish(self, cmd: "NatsPublishCommand") -> None:
    payload, content_type = encode_message(cmd.body, self.serializer)

    headers_to_send = {
        "content-type": content_type or "",
        **cmd.headers_to_publish(),
    }

    return await self.__state.connection.publish(
        subject=cmd.destination,
        payload=payload,
        reply=cmd.reply_to,
        headers=headers_to_send,
    )

request async #

request(cmd)
Source code in faststream/nats/publisher/producer.py 
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
@override
async def request(self, cmd: "NatsPublishCommand") -> "Msg":
    payload, content_type = encode_message(cmd.body, self.serializer)

    headers_to_send = {
        "content-type": content_type or "",
        **cmd.headers_to_publish(),
    }

    return await self.__state.connection.request(
        subject=cmd.destination,
        payload=payload,
        headers=headers_to_send,
        timeout=cmd.timeout,
    )

publish_batch async #

publish_batch(cmd)
Source code in faststream/nats/publisher/producer.py 
49
50
51
async def publish_batch(self, cmd: "NatsPublishCommand") -> None:
    msg = "NATS doesn't support publishing in batches."
    raise FeatureNotSupportedException(msg)