RabbitBroker

async def close(
    self,
    exc_type: Optional[Type[BaseException]] = None,
    exc_val: Optional[BaseException] = None,
    exec_tb: Optional[TracebackType] = None,
) -> None:
    """Closes the object.

    Args:
        exc_type: The type of the exception being handled, if any.
        exc_val: The exception instance being handled, if any.
        exec_tb: The traceback of the exception being handled, if any.

    Returns:
        None

    Raises:
        NotImplementedError: If the method is not implemented.

    """
    super()._abc_close(exc_type, exc_val, exec_tb)

    for h in self.handlers.values():
        await h.close()

    if self._connection is not None:
        await self._close(exc_type, exc_val, exec_tb)

async def connect(self, *args: Any, **kwargs: Any) -> aio_pika.RobustConnection:
    """Connect to the RabbitMQ server.

    Args:
        *args: Additional positional arguments.
        **kwargs: Additional keyword arguments.

    Returns:
        aio_pika.RobustConnection: The RabbitMQ connection instance.
    """
    connection = await super().connect(*args, **kwargs)
    for p in self._publishers.values():
        p._producer = self._producer
    return connection

async def declare_exchange(
    self,
    exchange: RabbitExchange,
) -> aio_pika.RobustExchange:
    """Declare a RabbitMQ exchange.

    Args:
        exchange (RabbitExchange): The RabbitMQ exchange to declare.

    Returns:
        aio_pika.RobustExchange: The declared RabbitMQ exchange.

    Raises:
        RuntimeError: If the declarer is not initialized in the `connect` method.
    """
    assert self.declarer, NOT_CONNECTED_YET  # nosec B101
    return await self.declarer.declare_exchange(exchange)

async def declare_queue(
    self,
    queue: RabbitQueue,
) -> aio_pika.RobustQueue:
    """Declare a RabbitMQ queue.

    Args:
        queue (RabbitQueue): The RabbitMQ queue to declare.

    Returns:
        aio_pika.RobustQueue: The declared RabbitMQ queue.

    Raises:
        RuntimeError: If the declarer is not initialized in the `connect` method.
    """
    assert self.declarer, NOT_CONNECTED_YET  # nosec B101
    return await self.declarer.declare_queue(queue)

def include_router(self, router: BrokerRouter[Any, MsgType]) -> None:
    """Includes a router in the current object.

    Args:
        router: The router to be included.

    Returns:
        None

    """
    for r in router._handlers:
        self.subscriber(*r.args, **r.kwargs)(r.call)

    self._publishers = {**self._publishers, **router._publishers}

def include_routers(self, *routers: BrokerRouter[Any, MsgType]) -> None:
    """Includes routers in the current object.

    Args:
        *routers: Variable length argument list of routers to include.

    Returns:
        None

    """
    for r in routers:
        self.include_router(r)

@override
async def publish(  # type: ignore[override]
    self,
    *args: Any,
    **kwargs: Any,
) -> Union[aiormq.abc.ConfirmationFrameType, SendableMessage]:
    """Publish a message to the RabbitMQ broker.

    Args:
        *args: Additional positional arguments.
        **kwargs: Additional keyword arguments.

    Returns:
        Union[aiormq.abc.ConfirmationFrameType, SendableMessage]: The confirmation frame or the response message.
    """
    assert self._producer, NOT_CONNECTED_YET  # nosec B101
    return await self._producer.publish(*args, **kwargs)

@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: Optional[str] = None,
    # AsyncAPI information
    title: Optional[str] = None,
    description: Optional[str] = None,
    schema: Optional[Any] = None,
    include_in_schema: bool = True,
    priority: Optional[int] = None,
    **message_kwargs: Any,
) -> Publisher:
    """Define a message publisher.

    Args:
        queue (Union[RabbitQueue, str], optional): The name of the RabbitMQ queue. Defaults to "".
        exchange (Union[RabbitExchange, str, None], optional): The name of the RabbitMQ exchange. Defaults to None.
        routing_key (str, optional): The routing key for messages. Defaults to "".
        mandatory (bool, optional): Whether the message is mandatory. Defaults to True.
        immediate (bool, optional): Whether the message should be sent immediately. Defaults to False.
        timeout (TimeoutType, optional): Timeout for message publishing. Defaults to None.
        persist (bool, optional): Whether to persist messages. Defaults to False.
        reply_to (Optional[str], optional): The reply-to queue name. Defaults to None.
        title (Optional[str]): Title for AsyncAPI docs.
        description (Optional[str]): Description for AsyncAPI docs.
        schema (Optional[Any]): Schema for AsyncAPI docs.
        include_in_schema (bool): Whether to include the publisher in AsyncAPI docs.
        priority (Optional[int]): Priority for the message.
        **message_kwargs (Any): Additional message properties and content.

    Returns:
        Publisher: A message publisher instance.
    """
    q, ex = RabbitQueue.validate(queue), RabbitExchange.validate(exchange)

    publisher = Publisher(
        title=title,
        queue=q,
        exchange=ex,
        routing_key=routing_key,
        mandatory=mandatory,
        immediate=immediate,
        timeout=timeout,
        persist=persist,
        reply_to=reply_to,
        priority=priority,
        message_kwargs=message_kwargs,
        _description=description,
        _schema=schema,
        virtual_host=self.virtual_host,
        include_in_schema=include_in_schema,
    )

    key = publisher._get_routing_hash()
    publisher = self._publishers.get(key, publisher)
    super().publisher(key, publisher)
    if self._producer is not None:
        publisher._producer = self._producer
    return publisher

async def start(self) -> None:
    """Start the RabbitMQ broker.

    Raises:
        RuntimeError: If the declarer is not initialized in the `connect` method.
    """
    context.set_global(
        "default_log_context",
        self._get_log_context(None, RabbitQueue(""), RabbitExchange("")),
    )

    await super().start()
    assert self.declarer, NOT_CONNECTED_YET  # nosec B101

    for publisher in self._publishers.values():
        if publisher.exchange is not None:
            await self.declare_exchange(publisher.exchange)

    for handler in self.handlers.values():
        c = self._get_log_context(None, handler.queue, handler.exchange)
        self._log(f"`{handler.call_name}` waiting for messages", extra=c)
        await handler.start(self.declarer)

@override
def subscriber(  # type: ignore[override]
    self,
    queue: Union[str, RabbitQueue],
    exchange: Union[str, RabbitExchange, None] = None,
    *,
    consume_args: Optional[AnyDict] = None,
    reply_config: Optional[ReplyConfig] = None,
    # broker arguments
    dependencies: Sequence[Depends] = (),
    parser: Optional[CustomParser[aio_pika.IncomingMessage, RabbitMessage]] = None,
    decoder: Optional[CustomDecoder[RabbitMessage]] = None,
    middlewares: Optional[
        Sequence[Callable[[aio_pika.IncomingMessage], BaseMiddleware]]
    ] = None,
    filter: Filter[RabbitMessage] = default_filter,
    no_ack: bool = False,
    # AsyncAPI information
    title: Optional[str] = None,
    description: Optional[str] = None,
    include_in_schema: bool = True,
    **original_kwargs: Any,
) -> Callable[
    [Callable[P_HandlerParams, T_HandlerReturn]],
    HandlerCallWrapper[aio_pika.IncomingMessage, P_HandlerParams, T_HandlerReturn],
]:
    """Decorator to define a message subscriber.

    Args:
        queue (Union[str, RabbitQueue]): The name of the RabbitMQ queue.
        exchange (Union[str, RabbitExchange, None], optional): The name of the RabbitMQ exchange. Defaults to None.
        consume_args (Optional[AnyDict], optional): Additional arguments for message consumption.
        reply_config (Optional[ReplyConfig], optional): The reply configuration for the message.
        dependencies (Sequence[Depends], optional): Additional dependencies for the handler function. Defaults to ().
        parser (Optional[CustomParser[aio_pika.IncomingMessage, RabbitMessage]], optional): Optional custom parser for parsing the input. Defaults to None.
        decoder (Optional[CustomDecoder[RabbitMessage]], optional): Optional custom decoder for decoding the input. Defaults to None.
        middlewares (Optional[Sequence[Callable[[aio_pika.IncomingMessage], BaseMiddleware]]], optional): Optional sequence of middlewares to be applied. Defaults to None.
        filter (Filter[RabbitMessage], optional): Optional filter for filtering messages. Defaults to default_filter.
        no_ack (bool): Whether not to ack/nack/reject messages.
        title (Optional[str]): Title for AsyncAPI docs.
        description (Optional[str]): Description for AsyncAPI docs.
        include_in_schema (bool): Whether to include the handler in AsyncAPI docs.
        **original_kwargs (Any): Additional keyword arguments.

    Returns:
        Callable: A decorator function for defining message subscribers.
    """
    super().subscriber()

    r_queue = RabbitQueue.validate(queue)
    r_exchange = RabbitExchange.validate(exchange)

    self._setup_log_context(r_queue, r_exchange)

    key = get_routing_hash(r_queue, r_exchange)
    handler = self.handlers.get(
        key,
        Handler(
            log_context_builder=partial(
                self._get_log_context, queue=r_queue, exchange=r_exchange
            ),
            queue=r_queue,
            exchange=r_exchange,
            consume_args=consume_args,
            description=description,
            title=title,
            virtual_host=self.virtual_host,
            include_in_schema=include_in_schema,
            graceful_timeout=self.graceful_timeout,
        ),
    )

    self.handlers[key] = handler

    def consumer_wrapper(
        func: Callable[P_HandlerParams, T_HandlerReturn],
    ) -> HandlerCallWrapper[
        aio_pika.IncomingMessage, P_HandlerParams, T_HandlerReturn
    ]:
        """Wraps a consumer function with additional functionality.

        Args:
            func: The consumer function to be wrapped.

        Returns:
            The wrapped consumer function.

        Raises:
            NotImplementedError: If silent animals are not supported.
        !!! note

            The above docstring is autogenerated by docstring-gen library (https://docstring-gen.airt.ai)
        """
        handler_call, dependant = self._wrap_handler(
            func,
            extra_dependencies=dependencies,
            no_ack=no_ack,
            _process_kwargs={
                "reply_config": reply_config,
            },
            **original_kwargs,
        )

        handler.add_call(
            handler=handler_call,
            filter=filter,
            middlewares=middlewares,
            parser=parser or self._global_parser,
            decoder=decoder or self._global_decoder,
            dependant=dependant,
        )

        return handler_call

    return consumer_wrapper