BrokerAsyncUsecase

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) -> ConnectionType:
    """Connect to a remote server.

    Args:
        *args: Variable length argument list.
        **kwargs: Arbitrary keyword arguments.

    Returns:
        The connection object.
    !!! note

        The above docstring is autogenerated by docstring-gen library (https://docstring-gen.airt.ai)
    """
    if self._connection is None:
        _kwargs = self._resolve_connection_kwargs(*args, **kwargs)
        self._connection = await self._connect(**_kwargs)
    return self._connection

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)

@abstractmethod
async def publish(
    self,
    message: SendableMessage,
    *args: Any,
    reply_to: str = "",
    rpc: bool = False,
    rpc_timeout: Optional[float] = None,
    raise_timeout: bool = False,
    **kwargs: Any,
) -> Optional[SendableMessage]:
    """Publish a message.

    Args:
        message: The message to be published.
        *args: Additional arguments.
        reply_to: The reply-to address for the message.
        rpc: Whether the message is for RPC.
        rpc_timeout: The timeout for RPC.
        raise_timeout: Whether to raise an exception on timeout.
        **kwargs: Additional keyword arguments.

    Returns:
        The published message.

    Raises:
        NotImplementedError: If the method is not implemented.
    !!! note

        The above docstring is autogenerated by docstring-gen library (https://docstring-gen.airt.ai)
    """
    raise NotImplementedError()

@abstractmethod
def publisher(
    self,
    key: Any,
    publisher: BasePublisher[MsgType],
) -> BasePublisher[MsgType]:
    """Publishes a publisher.

    Args:
        key: The key associated with the publisher.
        publisher: The publisher to be published.

    Returns:
        The published publisher.

    Raises:
        NotImplementedError: If the method is not implemented.
    !!! note

        The above docstring is autogenerated by docstring-gen library (https://docstring-gen.airt.ai)
    """
    self._publishers[key] = publisher
    return publisher

@abstractmethod
async def start(self) -> None:
    super()._abc_start()
    for h in self.handlers.values():
        for f, _, _, _, _, _ in h.calls:
            f.refresh(with_mock=False)
    await self.connect()

@override
@abstractmethod
def subscriber(  # type: ignore[override,return]
    self,
    *broker_args: Any,
    retry: Union[bool, int] = False,
    dependencies: Sequence[Depends] = (),
    decoder: Optional[CustomDecoder[StreamMessage[MsgType]]] = None,
    parser: Optional[CustomParser[MsgType, StreamMessage[MsgType]]] = None,
    middlewares: Optional[Sequence[Callable[[MsgType], BaseMiddleware]]] = None,
    filter: Filter[StreamMessage[MsgType]] = default_filter,
    _raw: bool = False,
    _get_dependant: Optional[Any] = None,
    **broker_kwargs: Any,
) -> Callable[
    [
        Union[
            Callable[P_HandlerParams, T_HandlerReturn],
            HandlerCallWrapper[MsgType, P_HandlerParams, T_HandlerReturn],
        ]
    ],
    HandlerCallWrapper[MsgType, P_HandlerParams, T_HandlerReturn],
]:
    """A function decorator for subscribing to a message broker.

    Args:
        *broker_args: Positional arguments to be passed to the message broker.
        retry: Whether to retry the subscription if it fails. Can be a boolean or an integer specifying the number of retries.
        dependencies: Sequence of dependencies to be injected into the decorated function.
        decoder: Custom decoder function for decoding the message.
        parser: Custom parser function for parsing the decoded message.
        middlewares: Sequence of middleware functions to be applied to the message.
        filter: Filter function for filtering the messages to be processed.
        _raw: Whether to return the raw message instead of the processed result.
        _get_dependant: Optional argument to get the dependant object.

    Returns:
        A callable decorator that wraps the decorated function and handles the subscription.

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

        The above docstring is autogenerated by docstring-gen library (https://docstring-gen.airt.ai)
    """
    super().subscriber()