BrokerUsecase

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)

@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.

    """
    self._publishers = {**self._publishers, key: publisher}
    return publisher

@abstractmethod
def subscriber(  # type: ignore[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[
                [StreamMessage[MsgType]],
                BaseMiddleware,
            ]
        ]
    ] = None,
    filter: Filter[StreamMessage[MsgType]] = lambda m: not m.processed,
    _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],
]:
    """This is a function decorator for subscribing to a message broker.

    Args:
        *broker_args: Positional arguments to be passed to the 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 handler function.
        decoder: Custom decoder function to decode the message.
        parser: Custom parser function to parse the decoded message.
        middlewares: Sequence of middleware functions to be applied to the message.
        filter: Filter function to filter the messages to be processed.
        _raw: Whether to return the raw message instead of the processed message.
        _get_dependant: Optional parameter to get the dependant object.
        **broker_kwargs: Keyword arguments to be passed to the broker.

    Returns:
        A callable object that can be used as a decorator for a handler function.

    Raises:
        RuntimeWarning: If the broker is already running.

    """
    if self.started and not is_test_env():  # pragma: no cover
        warnings.warn(
            "You are trying to register `handler` with already running broker\n"
            "It has no effect until broker restarting.",
            category=RuntimeWarning,
            stacklevel=1,
        )