RabbitBroker

async def publish_batch(
    self,
    *messages: "SendableMessage",
    queue: str,
) -> Any:
    msg = f"{self.__class__.__name__} doesn't support publishing in batches."
    raise FeatureNotSupportedException(msg)

def add_middleware(self, middleware: "BrokerMiddleware[Any, Any]") -> None:
    """Append BrokerMiddleware to the end of middlewares list.

    Current middleware will be used as a most inner of the stack.
    """
    self.config.add_middleware(middleware)

def insert_middleware(self, middleware: "BrokerMiddleware[Any, Any]") -> None:
    """Insert BrokerMiddleware to the start of middlewares list.

    Current middleware will be used as a most outer of the stack.
    """
    self.config.insert_middleware(middleware)

@override
def subscriber(  # type: ignore[override]
    self,
    queue: Union[str, "RabbitQueue"],
    exchange: Union[str, "RabbitExchange", None] = None,
    *,
    channel: Optional["Channel"] = None,
    consume_args: dict[str, Any] | None = None,
    ack_policy: AckPolicy = EMPTY,
    # broker arguments
    dependencies: Sequence["Dependant"] = (),
    parser: Optional["CustomCallable"] = None,
    decoder: Optional["CustomCallable"] = None,
    codec: Optional["CodecProto"] = None,
    no_reply: bool = False,
    persistent: bool = True,
    # AsyncAPI information
    title: str | None = None,
    description: str | None = None,
    include_in_schema: bool = True,
) -> "RabbitSubscriber":
    """Subscribe a handler to a RabbitMQ queue.

    Args:
        queue (Union[str, RabbitQueue]): RabbitMQ queue to listen. **FastStream** declares and binds queue object to `exchange` automatically by default.
        exchange (Union[str, RabbitExchange, None], optional): RabbitMQ exchange to bind queue to. Uses default exchange if not presented. **FastStream** declares exchange object automatically by default.
        channel (Optional[Channel], optional): Channel to use for consuming messages.
        consume_args (dict[str, Any] | None, optional): Extra consumer arguments to use in `queue.consume(...)` method.
        ack_policy (AckPolicy, optional): Acknowledgement policy for message processing.
        dependencies (Sequence[Dependant], optional): Dependencies list (`[Dependant(),]`) to apply to the subscriber.
        parser (Optional[CustomCallable], optional): Parser to map original **IncomingMessage** Msg to FastStream one.
        decoder (Optional[CustomCallable], optional): Function to decode FastStream msg bytes body to python objects.
        codec (Optional[CodecProto], optional): Custom codec object.
        no_reply (bool, optional): Whether to disable **FastStream** RPC and Reply To auto responses or not.
        title (Optional[str], optional): AsyncAPI subscriber object title.
        description (Optional[str], optional): AsyncAPI subscriber object description. Uses decorated docstring as default.
        include_in_schema (bool, optional): Whether to include operation in AsyncAPI schema or not.
        persistent (bool): Whether to make the subscriber persistent or not.

    Returns:
        RabbitSubscriber: The subscriber specification object.
    """
    subscriber = create_subscriber(
        queue=RabbitQueue.validate(queue),
        exchange=RabbitExchange.validate(exchange),
        consume_args=consume_args,
        channel=channel,
        # subscriber args
        ack_policy=ack_policy,
        no_reply=no_reply,
        # broker args
        config=cast("RabbitBrokerConfig", self.config),
        # specification args
        title_=title,
        description_=description,
        include_in_schema=include_in_schema,
    )

    super().subscriber(subscriber, persistent=persistent)

    return subscriber.add_call(
        parser_=parser,
        decoder_=decoder,
        codec_=codec,
        dependencies_=dependencies,
    )

@override
def publisher(  # type: ignore[override]
    self,
    queue: Union["RabbitQueue", str] = "",
    exchange: Union["RabbitExchange", str, None] = None,
    *,
    routing_key: str = "",
    mandatory: bool = True,
    immediate: bool = False,
    timeout: "TimeoutType" = None,
    persist: bool = False,
    reply_to: str | None = None,
    priority: int | None = None,
    persistent: bool = True,
    # AsyncAPI information
    title: str | None = None,
    description: str | None = None,
    schema: Any | None = None,
    include_in_schema: bool = True,
    # message args
    headers: Optional["HeadersType"] = None,
    content_type: str | None = None,
    content_encoding: str | None = None,
    expiration: Optional["DateType"] = None,
    message_type: str | None = None,
    user_id: str | None = None,
) -> "RabbitPublisher":
    """Creates long-living and AsyncAPI-documented publisher object.

    You can use it as a handler decorator (handler should be decorated by `@broker.subscriber(...)` too) - `@broker.publisher(...)`.
    In such case publisher will publish your handler return value.

    Or you can create a publisher object to call it lately - `broker.publisher(...).publish(...)`.

    Args:
        queue: Default message routing key to publish with.
        exchange: Target exchange to publish message to.
        routing_key: Default message routing key to publish with.
        Overrides `queue` option if presented.
        mandatory: Client waits for confirmation that the message is placed
            to some queue. RabbitMQ returns message to client if there is no suitable queue.
        immediate: Client expects that there is a consumer ready to take the message to work.
            RabbitMQ returns message to client if there is no suitable consumer.
        timeout: Send confirmation time from RabbitMQ.
        persist: Restore the message on RabbitMQ reboot.
        reply_to: Reply message routing key to send with (always sending to default exchange).
        priority: The message priority (0 by default).
        title: AsyncAPI publisher object title.
        description: AsyncAPI publisher object description.
        schema: AsyncAPI publishing message type. Should be any python-native
            object annotation or `pydantic.BaseModel`.
        include_in_schema: Whether to include operation in AsyncAPI schema or not.
        headers: Message headers to store meta-information. Can be overridden
            by `publish.headers` if specified.
        content_type: Message **content-type** header. Used by application, not core RabbitMQ.
            Will be set automatically if not specified.
        content_encoding: Message body content encoding, e.g. **gzip**.
        expiration: Message expiration (lifetime) in seconds (or datetime or timedelta).
        message_type: Application-specific message type, e.g. **orders.created**.
        user_id: Publisher connection User ID, validated if set.
        persistent: Whether to make the publisher persistent or not.
    """
    message_kwargs = PublishKwargs(
        mandatory=mandatory,
        immediate=immediate,
        timeout=timeout,
        persist=persist,
        reply_to=reply_to,
        headers=headers,
        priority=priority,
        content_type=content_type,
        content_encoding=content_encoding,
        message_type=message_type,
        user_id=user_id,
        expiration=expiration,
    )

    publisher = create_publisher(
        routing_key=routing_key,
        queue=RabbitQueue.validate(queue),
        exchange=RabbitExchange.validate(exchange),
        message_kwargs=message_kwargs,
        # broker args
        config=cast("RabbitBrokerConfig", self.config),
        # specification args
        title_=title,
        description_=description,
        schema_=schema,
        include_in_schema=include_in_schema,
    )

    super().publisher(publisher, persistent=persistent)

    return publisher

@override
def include_router(
    self,
    router: "RabbitRegistrator",  # type: ignore[override]
    *,
    prefix: str = "",
    dependencies: Sequence["Dependant"] = (),
    middlewares: Sequence["BrokerMiddleware[Any, Any]"] = (),
    include_in_schema: bool | None = None,
) -> None:
    if not isinstance(router, RabbitRegistrator):
        msg = (
            f"Router must be an instance of RabbitRegistrator, "
            f"got {type(router).__name__} instead"
        )
        raise SetupError(msg)

    super().include_router(
        router,
        prefix=prefix,
        dependencies=dependencies,
        middlewares=middlewares,
        include_in_schema=include_in_schema,
    )

def include_routers(
    self,
    *routers: "Registrator[MsgType, Any]",
) -> None:
    """Includes routers in the object."""
    for r in routers:
        self.include_router(r)

async def connect(self) -> ConnectionType:
    """Connect to a remote server."""
    if self._connection is None:
        self._connection = await self._connect()
        self._setup_logger()

    return self._connection

async def stop(
    self,
    exc_type: type[BaseException] | None = None,
    exc_val: BaseException | None = None,
    exc_tb: Optional["TracebackType"] = None,
) -> None:
    await super().stop(exc_type, exc_val, exc_tb)

    if self._channel is not None:
        if not self._channel.is_closed:
            await self._channel.close()

        self._channel = None

    if self._connection is not None:
        await self._connection.close()
        self._connection = None

    self.config.disconnect()

async def start(self) -> None:
    """Connect broker to RabbitMQ and startup all subscribers."""
    await self.connect()
    await self.declare_queue(RABBIT_REPLY)
    await super().start()

@override
async def publish(
    self,
    message: "AioPikaSendableMessage" = None,
    queue: Union["RabbitQueue", str] = "",
    exchange: Union["RabbitExchange", str, None] = None,
    *,
    routing_key: str = "",
    # publish options
    mandatory: bool = True,
    immediate: bool = False,
    timeout: "TimeoutType" = None,
    persist: bool = False,
    reply_to: str | None = None,
    correlation_id: str | None = None,
    # message options
    headers: Optional["HeadersType"] = None,
    content_type: str | None = None,
    content_encoding: str | None = None,
    expiration: Optional["DateType"] = None,
    message_id: str | None = None,
    timestamp: Optional["DateType"] = None,
    message_type: str | None = None,
    user_id: str | None = None,
    priority: int | None = None,
) -> Optional["aiormq.abc.ConfirmationFrameType"]:
    """Publish message directly.

    This method allows you to publish message in not AsyncAPI-documented way. You can use it in another frameworks
    applications or to publish messages from time to time.

    Please, use `@broker.publisher(...)` or `broker.publisher(...).publish(...)` instead in a regular way.

    Args:
        message:
            Message body to send.
        queue:
            Message routing key to publish with.
        exchange:
            Target exchange to publish message to.
        routing_key:
            Message routing key to publish with. Overrides `queue` option if presented.
        mandatory:
            Client waits for confirmation that the message is placed to some queue. RabbitMQ returns message to client if there is no suitable queue.
        immediate:
            Client expects that there is consumer ready to take the message to work. RabbitMQ returns message to client if there is no suitable consumer.
        timeout:
            Send confirmation time from RabbitMQ.
        persist:
            Restore the message on RabbitMQ reboot.
        reply_to:
            Reply message routing key to send with (always sending to default exchange).
        correlation_id:
            Manual message **correlation_id** setter. **correlation_id** is a useful option to trace messages.
        headers:
            Message headers to store metainformation.
        content_type:
            Message **content-type** header. Used by application, not core RabbitMQ. Will be set automatically if not specified.
        content_encoding:
            Message body content encoding, e.g. **gzip**.
        expiration:
            Message expiration (lifetime) in seconds (or datetime or timedelta).
        message_id:
            Arbitrary message id. Generated automatically if not present.
        timestamp:
            Message publish timestamp. Generated automatically if not presented.
        message_type:
            Application-specific message type, e.g. **orders.created**.
        user_id:
            Publisher connection User ID, validated if set.
        priority:
            The message priority (0 by default).

    Returns:
        An optional `aiormq.abc.ConfirmationFrameType` representing the confirmation frame if RabbitMQ is configured to send confirmations.
    """
    cmd = RabbitPublishCommand(
        message,
        routing_key=routing_key or RabbitQueue.validate(queue).routing(),
        exchange=RabbitExchange.validate(exchange),
        correlation_id=correlation_id or self.config.id_generator(),
        app_id=self.config.app_id,
        mandatory=mandatory,
        immediate=immediate,
        persist=persist,
        reply_to=reply_to,
        headers=headers,
        content_type=content_type,
        content_encoding=content_encoding,
        expiration=expiration,
        message_id=message_id,
        message_type=message_type,
        timestamp=timestamp,
        user_id=user_id,
        timeout=timeout,
        priority=priority,
        _publish_type=PublishType.PUBLISH,
    )

    result: aiormq.abc.ConfirmationFrameType | None = await super()._basic_publish(
        cmd,
        producer=self._producer,
    )
    return result

@override
async def request(  # type: ignore[override]
    self,
    message: "AioPikaSendableMessage" = None,
    queue: Union["RabbitQueue", str] = "",
    exchange: Union["RabbitExchange", str, None] = None,
    *,
    routing_key: str = "",
    mandatory: bool = True,
    immediate: bool = False,
    timeout: "TimeoutType" = None,
    persist: bool = False,
    # message args
    correlation_id: str | None = None,
    headers: Optional["HeadersType"] = None,
    content_type: str | None = None,
    content_encoding: str | None = None,
    expiration: Optional["DateType"] = None,
    message_id: str | None = None,
    timestamp: Optional["DateType"] = None,
    message_type: str | None = None,
    user_id: str | None = None,
    priority: int | None = None,
) -> "RabbitMessage":
    """Make a synchronous request to RabbitMQ.

    This method uses Direct Reply-To pattern to send a message and wait for a reply.
    It is a blocking call and will wait for a reply until timeout.

    Args:
        message: Message body to send.
        queue: Message routing key to publish with.
        exchange: Target exchange to publish message to.
        routing_key: Message routing key to publish with. Overrides `queue` option if presented.
        mandatory: Client waits for confirmation that the message is placed to some queue.
        RabbitMQ returns message to client if there is no suitable queue.
        immediate: Client expects that there is a consumer ready to take the message to work.
        RabbitMQ returns message to client if there is no suitable consumer.
        timeout: Send confirmation time from RabbitMQ.
        persist: Restore the message on RabbitMQ reboot.
        correlation_id: Manual message **correlation_id** setter. **correlation_id** is a useful option to trace messages.
        headers: Message headers to store metainformation.
        content_type: Message **content-type** header. Used by application, not core RabbitMQ.
        Will be set automatically if not specified.
        content_encoding: Message body content encoding, e.g. **gzip**.
        expiration: Message expiration (lifetime) in seconds (or datetime or timedelta).
        message_id: Arbitrary message id. Generated automatically if not present.
        timestamp: Message publish timestamp. Generated automatically if not present.
        message_type: Application-specific message type, e.g. **orders.created**.
        user_id: Publisher connection User ID, validated if set.
        priority: The message priority (0 by default).
    """
    cmd = RabbitPublishCommand(
        message,
        routing_key=routing_key or RabbitQueue.validate(queue).routing(),
        exchange=RabbitExchange.validate(exchange),
        correlation_id=correlation_id or self.config.id_generator(),
        app_id=self.config.app_id,
        mandatory=mandatory,
        immediate=immediate,
        persist=persist,
        headers=headers,
        content_type=content_type,
        content_encoding=content_encoding,
        expiration=expiration,
        message_id=message_id,
        message_type=message_type,
        timestamp=timestamp,
        user_id=user_id,
        timeout=timeout,
        priority=priority,
        _publish_type=PublishType.REQUEST,
    )

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

async def declare_queue(self, queue: "RabbitQueue") -> "RobustQueue":
    """Declares queue object in **RabbitMQ**."""
    declarer: RabbitDeclarer = self.config.declarer
    return await declarer.declare_queue(queue)

async def declare_exchange(self, exchange: "RabbitExchange") -> "RobustExchange":
    """Declares exchange object in **RabbitMQ**."""
    declarer: RabbitDeclarer = self.config.declarer
    return await declarer.declare_exchange(exchange)

@override
async def ping(self, timeout: float | None) -> bool:
    sleep_time = (timeout or 10) / 10

    with anyio.move_on_after(timeout) as cancel_scope:
        if self._connection is None:
            return False

        while True:
            if cancel_scope.cancel_called:
                return False

            if self._connection.connected.is_set():
                return True

            await anyio.sleep(sleep_time)

    return False