FastStream

def after_shutdown(
    self,
    func: Callable[P_HookParams, T_HookReturn],
) -> Callable[P_HookParams, T_HookReturn]:
    """Add hook running AFTER broker disconnected.

    Args:
        func: async or sync func to call as a hook

    Returns:
        Async version of the func argument
    """
    super().after_shutdown(to_async(func))
    return func

def after_startup(
    self,
    func: Callable[P_HookParams, T_HookReturn],
) -> Callable[P_HookParams, T_HookReturn]:
    """Add hook running AFTER broker connected.

    Args:
        func: async or sync func to call as a hook

    Returns:
        Async version of the func argument
    """
    super().after_startup(to_async(func))
    return func

def on_shutdown(
    self,
    func: Callable[P_HookParams, T_HookReturn],
) -> Callable[P_HookParams, T_HookReturn]:
    """Add hook running BEFORE broker disconnected.

    Args:
        func: async or sync func to call as a hook

    Returns:
        Async version of the func argument
    """
    super().on_shutdown(to_async(func))
    return func

def on_startup(
    self,
    func: Callable[P_HookParams, T_HookReturn],
) -> Callable[P_HookParams, T_HookReturn]:
    """Add hook running BEFORE broker connected.

    This hook also takes an extra CLI options as a kwargs

    Args:
        func: async or sync func to call as a hook

    Returns:
        Async version of the func argument
    """
    super().on_startup(to_async(func))
    return func

async def run(
    self,
    log_level: int = logging.INFO,
    run_extra_options: Optional[Dict[str, SettingField]] = None,
) -> None:
    """Run FastStream Application.

    Args:
        log_level: force application log level
        run_extra_options: extra options for running the app

    Returns:
        Block an event loop until stopped
    """
    assert self.broker, "You should setup a broker"  # nosec B101

    async with self.lifespan_context(**(run_extra_options or {})):
        try:
            async with anyio.create_task_group() as tg:
                tg.start_soon(self._start, log_level, run_extra_options)
                await self._stop(log_level)
                tg.cancel_scope.cancel()
        except ExceptionGroup as e:
            for ex in e.exceptions:
                raise ex from None

def set_broker(self, broker: BrokerAsyncUsecase[Any, Any]) -> None:
    """Set already existed App object broker.

    Useful then you create/init broker in `on_startup` hook.
    """
    self.broker = broker