LogicPublisher

@override
async def publish(
    self,
    message: "SendableMessage",
    subject: str = "",
    headers: dict[str, str] | None = None,
    reply_to: str = "",
    correlation_id: str | None = None,
    stream: str | None = None,
    timeout: float | None = None,
) -> Optional["PubAck"]:
    """Publish message directly.

    Args:
        message:
            Message body to send.
            Can be any encodable object (native python types or `pydantic.BaseModel`).
        subject:
            NATS subject to send message.
        headers:
            Message headers to store metainformation.
            **content-type** and **correlation_id** will be set automatically by framework anyway.
        reply_to:
            NATS subject name to send response.
        correlation_id:
            Manual message **correlation_id** setter.
            **correlation_id** is a useful option to trace messages.
        stream:
            This option validates that the target subject is in presented stream.
            Can be omitted without any effect if you doesn't want PubAck frame.
        timeout:
            Timeout to send message to NATS.

    Returns:
        `None` if you publishes a regular message.
        `faststream.nats.PubAck` if you publishes a message to stream.
    """
    cmd = NatsPublishCommand(
        message,
        subject=subject or self.subject,
        headers=self.headers | (headers or {}),
        reply_to=reply_to or self.reply_to,
        correlation_id=correlation_id or gen_cor_id(),
        stream=stream or getattr(self.stream, "name", None),
        timeout=timeout or self.timeout,
        _publish_type=PublishType.PUBLISH,
    )

    response: PubAck | None
    if cmd.stream:
        response = cast(
            "PubAck",
            await self._basic_publish(
                cmd,
                producer=self._outer_config.js_producer,
                _extra_middlewares=(),
            ),
        )
    else:
        response = await self._basic_publish(
            cmd,
            producer=self._outer_config.producer,
            _extra_middlewares=(),
        )

    return response

@override
async def request(
    self,
    message: "SendableMessage",
    subject: str = "",
    headers: dict[str, str] | None = None,
    correlation_id: str | None = None,
    stream: str | None = None,
    timeout: float = 0.5,
) -> "NatsMessage":
    """Make a synchronous request to outer subscriber.

    If out subscriber listens subject by stream, you should setup the same **stream** explicitly.
    Another way you will reseave confirmation frame as a response.

    Args:
        message:
            Message body to send.
            Can be any encodable object (native python types or `pydantic.BaseModel`).
        subject:
            NATS subject to send message.
        headers:
            Message headers to store metainformation.
            **content-type** and **correlation_id** will be set automatically by framework anyway.
        correlation_id:
            Manual message **correlation_id** setter.
            **correlation_id** is a useful option to trace messages.
        stream:
            This allows to make RPC calls over JetStream subjects.
        timeout:
            Timeout to send message to NATS.

    Returns:
        `faststream.nats.message.NatsMessage` object as an outer subscriber response.
    """
    cmd = NatsPublishCommand(
        message=message,
        subject=subject or self.subject,
        headers=self.headers | (headers or {}),
        timeout=timeout or self.timeout,
        correlation_id=correlation_id or gen_cor_id(),
        stream=stream or getattr(self.stream, "name", None),
        _publish_type=PublishType.REQUEST,
    )

    if cmd.stream:
        producer: ProducerProto[Any] = self._outer_config.js_producer
    else:
        producer = self._outer_config.producer

    msg: NatsMessage = await self._basic_request(cmd, producer=producer)
    return msg

async def start(self) -> None:
    pass

def set_test(
    self,
    *,
    mock: MagicMock,
    with_fake: bool,
) -> None:
    """Turn publisher to testing mode."""
    self.mock = mock
    self._fake_handler = with_fake

def reset_test(self) -> None:
    """Turn off publisher's testing mode."""
    self._fake_handler = False
    self.mock.reset_mock()

def schema(self) -> dict[str, "PublisherSpec"]:
    return self.specification.get_schema()