AsyncAPIPublisher

@override
async def publish(
    self,
    message: "AioPikaSendableMessage",
    queue: Annotated[
        Union["RabbitQueue", str, None],
        Doc("Message routing key to publish with."),
    ] = None,
    exchange: Annotated[
        Union["RabbitExchange", str, None],
        Doc("Target exchange to publish message to."),
    ] = None,
    *,
    routing_key: Annotated[
        str,
        Doc(
            "Message routing key to publish with. "
            "Overrides `queue` option if presented."
        ),
    ] = "",
    # message args
    correlation_id: Annotated[
        Optional[str],
        Doc(
            "Manual message **correlation_id** setter. "
            "**correlation_id** is a useful option to trace messages."
        ),
    ] = None,
    message_id: Annotated[
        Optional[str],
        Doc("Arbitrary message id. Generated automatically if not presented."),
    ] = None,
    timestamp: Annotated[
        Optional["DateType"],
        Doc("Message publish timestamp. Generated automatically if not presented."),
    ] = None,
    # rpc args
    rpc: Annotated[
        bool,
        Doc("Whether to wait for reply in blocking mode."),
        deprecated(
            "Deprecated in **FastStream 0.5.17**. "
            "Please, use `request` method instead. "
            "Argument will be removed in **FastStream 0.6.0**."
        ),
    ] = False,
    rpc_timeout: Annotated[
        Optional[float],
        Doc("RPC reply waiting time."),
        deprecated(
            "Deprecated in **FastStream 0.5.17**. "
            "Please, use `request` method with `timeout` instead. "
            "Argument will be removed in **FastStream 0.6.0**."
        ),
    ] = 30.0,
    raise_timeout: Annotated[
        bool,
        Doc(
            "Whetever to raise `TimeoutError` or return `None` at **rpc_timeout**. "
            "RPC request returns `None` at timeout by default."
        ),
        deprecated(
            "Deprecated in **FastStream 0.5.17**. "
            "`request` always raises TimeoutError instead. "
            "Argument will be removed in **FastStream 0.6.0**."
        ),
    ] = False,
    # publisher specific
    _extra_middlewares: Annotated[
        Iterable["PublisherMiddleware"],
        Doc("Extra middlewares to wrap publishing process."),
    ] = (),
    **publish_kwargs: "Unpack[PublishKwargs]",
) -> Optional[Any]:
    assert self._producer, NOT_CONNECTED_YET  # nosec B101

    kwargs: AnyDict = {
        "routing_key": routing_key
        or self.routing_key
        or RabbitQueue.validate(queue or self.queue).routing,
        "exchange": exchange or self.exchange.name,
        "app_id": self.app_id,
        "correlation_id": correlation_id or gen_cor_id(),
        "message_id": message_id,
        "timestamp": timestamp,
        # specific args
        "rpc": rpc,
        "rpc_timeout": rpc_timeout,
        "raise_timeout": raise_timeout,
        "reply_to": self.reply_to,
        **self.message_kwargs,
        **publish_kwargs,
    }

    call: AsyncFunc = self._producer.publish

    for m in chain(
        (
            _extra_middlewares
            or (m(None).publish_scope for m in self._broker_middlewares)
        ),
        self._middlewares,
    ):
        call = partial(m, call)

    return await call(message, **kwargs)

@override
async def request(
    self,
    message: "AioPikaSendableMessage",
    queue: Annotated[
        Union["RabbitQueue", str, None],
        Doc("Message routing key to publish with."),
    ] = None,
    exchange: Annotated[
        Union["RabbitExchange", str, None],
        Doc("Target exchange to publish message to."),
    ] = None,
    *,
    routing_key: Annotated[
        str,
        Doc(
            "Message routing key to publish with. "
            "Overrides `queue` option if presented."
        ),
    ] = "",
    # message args
    correlation_id: Annotated[
        Optional[str],
        Doc(
            "Manual message **correlation_id** setter. "
            "**correlation_id** is a useful option to trace messages."
        ),
    ] = None,
    message_id: Annotated[
        Optional[str],
        Doc("Arbitrary message id. Generated automatically if not presented."),
    ] = None,
    timestamp: Annotated[
        Optional["DateType"],
        Doc("Message publish timestamp. Generated automatically if not presented."),
    ] = None,
    # publisher specific
    _extra_middlewares: Annotated[
        Iterable["PublisherMiddleware"],
        Doc("Extra middlewares to wrap publishing process."),
    ] = (),
    **publish_kwargs: "Unpack[RequestPublishKwargs]",
) -> "RabbitMessage":
    assert self._producer, NOT_CONNECTED_YET  # nosec B101

    kwargs: AnyDict = {
        "routing_key": routing_key
        or self.routing_key
        or RabbitQueue.validate(queue or self.queue).routing,
        "exchange": exchange or self.exchange.name,
        "app_id": self.app_id,
        "correlation_id": correlation_id or gen_cor_id(),
        "message_id": message_id,
        "timestamp": timestamp,
        # specific args
        **self.message_kwargs,
        **publish_kwargs,
    }

    request: AsyncFunc = self._producer.request

    for pub_m in chain(
        (
            _extra_middlewares
            or (m(None).publish_scope for m in self._broker_middlewares)
        ),
        self._middlewares,
    ):
        request = partial(pub_m, request)

    published_msg = await request(
        message,
        **kwargs,
    )

    async with AsyncExitStack() as stack:
        return_msg: Callable[[RabbitMessage], Awaitable[RabbitMessage]] = (
            return_input
        )
        for m in self._broker_middlewares:
            mid = m(published_msg)
            await stack.enter_async_context(mid)
            return_msg = partial(mid.consume_scope, return_msg)

        parsed_msg = await self._producer._parser(published_msg)
        parsed_msg._decoded_body = await self._producer._decoder(parsed_msg)
        parsed_msg._source_type = SourceType.Response
        return await return_msg(parsed_msg)

    raise AssertionError("unreachable")

@override
def setup(  # type: ignore[override]
    self,
    *,
    producer: Optional["AioPikaFastProducer"],
    app_id: Optional[str],
    virtual_host: str,
) -> None:
    self.app_id = app_id
    self.virtual_host = virtual_host
    super().setup(producer=producer)

def add_prefix(self, prefix: str) -> None:
    """Include Publisher in router."""
    new_q = deepcopy(self.queue)
    new_q.name = prefix + new_q.name
    self.queue = new_q

def schema(self) -> Dict[str, Channel]:
    """Returns the schema of the API operation as a dictionary of channel names and channel objects."""
    if self.include_in_schema:
        return self.get_schema()
    else:
        return {}

def add_middleware(self, middleware: "BrokerMiddleware[MsgType]") -> None:
    self._broker_middlewares = (*self._broker_middlewares, middleware)

def get_description(self) -> Optional[str]:
    """Description property fallback."""
    return None

def get_payloads(self) -> List[Tuple["AnyDict", str]]:
    payloads: List[Tuple[AnyDict, str]] = []

    if self.schema_:
        params = {"response__": (self.schema_, ...)}

        call_model: CallModel[Any, Any] = CallModel(
            call=lambda: None,
            model=create_model("Fake"),
            response_model=create_model(  # type: ignore[call-overload]
                "",
                __config__=get_config_base(),
                **params,
            ),
            params=params,
        )

        body = get_response_schema(
            call_model,
            prefix=f"{self.name}:Message",
        )
        if body:  # pragma: no branch
            payloads.append((body, ""))

    else:
        for call in self.calls:
            call_model = build_call_model(call)
            body = get_response_schema(
                call_model,
                prefix=f"{self.name}:Message",
            )
            if body:
                payloads.append((body, to_camelcase(unwrap(call).__name__)))

    return payloads

def set_test(
    self,
    *,
    mock: Annotated[
        MagicMock,
        Doc("Mock object to check in tests."),
    ],
    with_fake: Annotated[
        bool,
        Doc("Whetevet publisher's fake subscriber created or not."),
    ],
) -> 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 = None

def get_name(self) -> str:
    routing = (
        self.routing_key
        or (self.queue.routing if is_routing_exchange(self.exchange) else None)
        or "_"
    )

    return f"{routing}:{getattr(self.exchange, 'name', None) or '_'}:Publisher"

def get_schema(self) -> Dict[str, Channel]:
    payloads = self.get_payloads()

    return {
        self.name: Channel(
            description=self.description,
            publish=Operation(
                bindings=OperationBinding(
                    amqp=amqp.OperationBinding(
                        cc=self.routing or None,
                        deliveryMode=2 if self.message_kwargs.get("persist") else 1,
                        mandatory=self.message_kwargs.get("mandatory"),
                        replyTo=self.message_kwargs.get("reply_to"),
                        priority=self.message_kwargs.get("priority"),
                    ),
                )
                if is_routing_exchange(self.exchange)
                else None,
                message=Message(
                    title=f"{self.name}:Message",
                    payload=resolve_payloads(
                        payloads,
                        "Publisher",
                        served_words=2 if self.title_ is None else 1,
                    ),
                    correlationId=CorrelationId(
                        location="$message.header#/correlation_id"
                    ),
                ),
            ),
            bindings=ChannelBinding(
                amqp=amqp.ChannelBinding(
                    **{
                        "is": "routingKey",
                        "queue": amqp.Queue(
                            name=self.queue.name,
                            durable=self.queue.durable,
                            exclusive=self.queue.exclusive,
                            autoDelete=self.queue.auto_delete,
                            vhost=self.virtual_host,
                        )
                        if is_routing_exchange(self.exchange) and self.queue.name
                        else None,
                        "exchange": (
                            amqp.Exchange(type="default", vhost=self.virtual_host)
                            if not self.exchange.name
                            else amqp.Exchange(
                                type=self.exchange.type.value,
                                name=self.exchange.name,
                                durable=self.exchange.durable,
                                autoDelete=self.exchange.auto_delete,
                                vhost=self.virtual_host,
                            )
                        ),
                    }
                )
            ),
        )
    }

@override
@classmethod
def create(  # type: ignore[override]
    cls,
    *,
    routing_key: str,
    queue: "RabbitQueue",
    exchange: "RabbitExchange",
    message_kwargs: "PublishKwargs",
    # Publisher args
    broker_middlewares: Iterable["BrokerMiddleware[IncomingMessage]"],
    middlewares: Iterable["PublisherMiddleware"],
    # AsyncAPI args
    schema_: Optional[Any],
    title_: Optional[str],
    description_: Optional[str],
    include_in_schema: bool,
) -> "AsyncAPIPublisher":
    return cls(
        routing_key=routing_key,
        queue=queue,
        exchange=exchange,
        message_kwargs=message_kwargs,
        # Publisher args
        broker_middlewares=broker_middlewares,
        middlewares=middlewares,
        # AsyncAPI args
        schema_=schema_,
        title_=title_,
        description_=description_,
        include_in_schema=include_in_schema,
    )