NatsJSFastProducer

def connect(
    self,
    connection: "JetStreamContext",
    serializer: Optional["SerializerProto"],
    codec: Optional["CodecProto"] = None,
) -> None:
    self.serializer = serializer
    self.codec = codec or DefaultCodec()
    self.__state = ConnectedState(connection)

def disconnect(self) -> None:
    self.__state = EmptyConnectionState()

@override
async def publish(self, cmd: "NatsPublishCommand") -> "PubAck":
    payload, content_type = await self.codec.encode(cmd.body, self.serializer)

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

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

@override
async def request(self, cmd: "NatsPublishCommand") -> "Msg":
    payload, content_type = await self.codec.encode(cmd.body, self.serializer)

    reply_to = self.__state.connection._nc.new_inbox()
    future: asyncio.Future[Msg] = asyncio.Future()
    sub = await self.__state.connection._nc.subscribe(
        reply_to,
        future=future,
        max_msgs=1,
    )
    try:
        await sub.unsubscribe(limit=1)

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

        with anyio.fail_after(cmd.timeout):
            await self.__state.connection.publish(
                subject=cmd.destination,
                payload=payload,
                headers=headers_to_send,
                stream=cmd.stream,
                timeout=cmd.timeout,
            )

            msg = await future

            if (  # pragma: no cover
                msg.headers
                and (msg.headers.get(Header.STATUS) == NO_RESPONDERS_STATUS)
            ):
                raise nats.errors.NoRespondersError

            return msg
    finally:
        future.cancel()
        # The one-message limit cannot remove an inbox that never receives a reply.
        with (
            anyio.CancelScope(shield=True),
            suppress(
                nats.errors.ConnectionClosedError,
                nats.errors.ConnectionDrainingError,
            ),
        ):
            await sub.unsubscribe()

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