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.
    !!! note

        The above docstring is autogenerated by docstring-gen library (https://docstring-gen.airt.ai)
    """
    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 (  # nosec B101
        self.declarer
    ), "Declarer should be initialized in `connect` method"
    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 (  # nosec B101
        self.declarer
    ), "Declarer should be initialized in `connect` method"
    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
    !!! note

        The above docstring is autogenerated by docstring-gen library (https://docstring-gen.airt.ai)
    """
    for r in router._handlers:
        self.subscriber(*r.args, **r.kwargs)(r.call)

    self._publishers.update(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
    !!! note

        The above docstring is autogenerated by docstring-gen library (https://docstring-gen.airt.ai)
    """
    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, "RabbitBroker channel is not started 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,
    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.
        **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,
    )

    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_local(
        "log_context",
        self._get_log_context(None, RabbitQueue(""), RabbitExchange("")),
    )

    await super().start()
    assert (  # nosec B101
        self.declarer
    ), "Declarer should be initialized in `connect` method"

    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,
    # 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,
    # AsyncAPI information
    title: Optional[str] = None,
    description: Optional[str] = None,
    **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.
        title (Optional[str]): Title for AsyncAPI docs.
        description (Optional[str]): Description for AsyncAPI docs.

    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,
        ),
    )

    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,
            **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