RabbitPublisher

async def assert_called_once_with(
    self,
    body: Any = EMPTY,
    /,
    *,
    headers: Any = EMPTY,
    correlation_id: Any = EMPTY,
    reply_to: Any = EMPTY,
    content_type: Any = EMPTY,
    path: Any = EMPTY,
    context: Mapping[str, Any] = EMPTY,
) -> None:
    """Assert the endpoint was called once, with the message described here.

    Args:
        body: The body as a dict, a model or a matcher; it goes through the codec.
        headers: Headers the message must carry; the rest may carry more.
        correlation_id: The exact correlation id.
        reply_to: The exact reply-to destination.
        content_type: The exact content type.
        path: The exact path parameters the subject template matched.
        context: Context paths, as given to `Context()`, mapped to their values.
    """
    await self._assert_called_once_with(
        ExpectedCall(
            body=body,
            headers=headers,
            correlation_id=correlation_id,
            reply_to=reply_to,
            content_type=content_type,
            path=path,
            context=context,
        )
    )

async def assert_called_with(
    self,
    body: Any = EMPTY,
    /,
    *,
    headers: Any = EMPTY,
    correlation_id: Any = EMPTY,
    reply_to: Any = EMPTY,
    content_type: Any = EMPTY,
    path: Any = EMPTY,
    context: Mapping[str, Any] = EMPTY,
) -> None:
    """Assert the last message the endpoint saw is the one described here.

    Args:
        body: The body as a dict, a model or a matcher; it goes through the codec.
        headers: Headers the message must carry; the rest may carry more.
        correlation_id: The exact correlation id.
        reply_to: The exact reply-to destination.
        content_type: The exact content type.
        path: The exact path parameters the subject template matched.
        context: Context paths, as given to `Context()`, mapped to their values.
    """
    await self._assert_called_with(
        ExpectedCall(
            body=body,
            headers=headers,
            correlation_id=correlation_id,
            reply_to=reply_to,
            content_type=content_type,
            path=path,
            context=context,
        )
    )

async def assert_any_call(
    self,
    body: Any = EMPTY,
    /,
    *,
    headers: Any = EMPTY,
    correlation_id: Any = EMPTY,
    reply_to: Any = EMPTY,
    content_type: Any = EMPTY,
    path: Any = EMPTY,
    context: Mapping[str, Any] = EMPTY,
) -> None:
    """Assert one of the messages the endpoint saw is the one described here.

    Args:
        body: The body as a dict, a model or a matcher; it goes through the codec.
        headers: Headers the message must carry; the rest may carry more.
        correlation_id: The exact correlation id.
        reply_to: The exact reply-to destination.
        content_type: The exact content type.
        path: The exact path parameters the subject template matched.
        context: Context paths, as given to `Context()`, mapped to their values.
    """
    await self._assert_any_call(
        ExpectedCall(
            body=body,
            headers=headers,
            correlation_id=correlation_id,
            reply_to=reply_to,
            content_type=content_type,
            path=path,
            context=context,
        )
    )

def set_test(
    self,
    *,
    recorder: CallRecorder | None = None,
    with_fake: bool,
) -> None:
    """Turn publisher to testing mode, sharing `recorder` when one is given."""
    self.is_test = True
    if recorder is None:
        self._recorder.reset()
    else:
        self._recorder = recorder
    self._fake_handler = with_fake

def reset_test(self) -> None:
    """Turn off publisher's testing mode."""
    self.is_test = False
    self._recorder.reset()
    # A shared recorder goes back to the handler it belongs to
    self._recorder = CallRecorder(self.specification.name, self._outer_config)
    self._fake_handler = False

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

def routing(
    self,
    *,
    queue: Union["RabbitQueue", str, None] = None,
    routing_key: str = "",
) -> str:
    if not routing_key:
        if q := RabbitQueue.validate(queue):
            routing_key = q.routing()
        else:
            r = self.routing_key or self.queue.routing()
            routing_key = f"{self._outer_config.prefix}{r}"

    return routing_key

async def start(self) -> None:
    if self.exchange is not None:
        await self._outer_config.declarer.declare_exchange(self.exchange)
    return await super().start()

@override
async def publish(
    self,
    message: "AioPikaSendableMessage",
    queue: Union["RabbitQueue", str, None] = None,
    exchange: Union["RabbitExchange", str, None] = None,
    *,
    routing_key: str = "",
    **publish_kwargs: "Unpack[PublishKwargs]",
) -> Optional["aiormq.abc.ConfirmationFrameType"]:
    if "headers" in publish_kwargs:
        headers = self.headers | (publish_kwargs.pop("headers") or {})
    else:
        headers = self.headers

    correlation_id = (
        publish_kwargs.pop("correlation_id", None)
        or self._outer_config.id_generator()
    )

    cmd = RabbitPublishCommand(
        message,
        routing_key=self.routing(queue=queue, routing_key=routing_key),
        exchange=RabbitExchange.validate(exchange or self.exchange),
        headers=headers,
        correlation_id=correlation_id,
        _publish_type=PublishType.PUBLISH,
        **(self.publish_options | self.message_options | publish_kwargs),  # type: ignore[operator]
    )

    frame: aiormq.abc.ConfirmationFrameType | None = await self._basic_publish(
        cmd,
        producer=self._outer_config.producer,
        _extra_middlewares=(),
    )
    return frame

@override
async def request(
    self,
    message: "AioPikaSendableMessage",
    queue: Union["RabbitQueue", str, None] = None,
    exchange: Union["RabbitExchange", str, None] = None,
    *,
    routing_key: str = "",
    **publish_kwargs: "Unpack[PublishKwargs]",
) -> "RabbitMessage":
    if "headers" in publish_kwargs:
        headers = self.headers | (publish_kwargs.pop("headers") or {})
    else:
        headers = self.headers

    correlation_id = (
        publish_kwargs.pop("correlation_id", None)
        or self._outer_config.id_generator()
    )

    cmd = RabbitPublishCommand(
        message,
        routing_key=self.routing(queue=queue, routing_key=routing_key),
        exchange=RabbitExchange.validate(exchange or self.exchange),
        correlation_id=correlation_id,
        headers=headers,
        _publish_type=PublishType.PUBLISH,
        **(self.publish_options | self.message_options | publish_kwargs),  # type: ignore[operator]
    )

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