AsgiFastStream

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

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

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.
    """
    self._on_startup_calling.append(apply_types(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."""
    self._on_shutdown_calling.append(apply_types(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."""
    self._after_startup_calling.append(apply_types(to_async(func)))
    return func

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

def exit(self) -> None:
    """Stop application manually."""
    self._should_exit = True

async def start(
    self,
    **run_extra_options: "SettingField",
) -> None:
    """Executes startup hooks and start broker."""
    for func in self._on_startup_calling:
        await func(**run_extra_options)

    if self.broker is not None:
        await self.broker.start()

    for func in self._after_startup_calling:
        await func()

async def stop(self) -> None:
    """Executes shutdown hooks and stop broker."""
    for func in self._on_shutdown_calling:
        await func()

    if self.broker is not None:
        await self.broker.close()

    for func in self._after_shutdown_calling:
        await func()

@classmethod
def from_app(
    cls,
    app: Application,
    asgi_routes: Sequence[Tuple[str, "ASGIApp"]],
    asyncapi_path: Optional[str] = None,
) -> "AsgiFastStream":
    asgi_app = cls(
        app.broker,
        asgi_routes=asgi_routes,
        asyncapi_path=asyncapi_path,
        logger=app.logger,
        lifespan=None,
        title=app.title,
        version=app.version,
        description=app.description,
        terms_of_service=app.terms_of_service,
        license=app.license,
        contact=app.contact,
        tags=app.asyncapi_tags,
        external_docs=app.external_docs,
        identifier=app.identifier,
    )
    asgi_app.lifespan_context = app.lifespan_context
    asgi_app._on_startup_calling = app._on_startup_calling
    asgi_app._after_startup_calling = app._after_startup_calling
    asgi_app._on_shutdown_calling = app._on_shutdown_calling
    asgi_app._after_shutdown_calling = app._after_shutdown_calling
    return asgi_app

def mount(self, path: str, route: "ASGIApp") -> None:
    self.routes.append((path, route))

async def run(
    self,
    log_level: int = logging.INFO,
    run_extra_options: Optional[Dict[str, "SettingField"]] = None,
    sleep_time: float = 0.1,
) -> None:
    try:
        import uvicorn
    except ImportError as e:
        raise RuntimeError(
            "You need uvicorn to run FastStream ASGI App via CLI. pip install uvicorn"
        ) from e

    run_extra_options = cast_uvicorn_params(run_extra_options or {})

    uvicorn_config_params = set(inspect.signature(uvicorn.Config).parameters.keys())

    config = uvicorn.Config(
        app=self,
        log_level=log_level,
        **{
            key: v
            for key, v in run_extra_options.items()
            if key in uvicorn_config_params
        },
    )

    server = uvicorn.Server(config)
    await server.serve()

@asynccontextmanager
async def start_lifespan_context(self) -> AsyncIterator[None]:
    async with anyio.create_task_group() as tg, self.lifespan_context():
        tg.start_soon(self._startup)

        try:
            yield
        finally:
            await self._shutdown()
            tg.cancel_scope.cancel()

async def lifespan(self, scope: "Scope", receive: "Receive", send: "Send") -> None:
    """Handle ASGI lifespan messages to start and shutdown the app."""
    started = False
    await receive()  # handle `lifespan.startup` event

    try:
        async with self.start_lifespan_context():
            await send({"type": "lifespan.startup.complete"})
            started = True
            await receive()  # handle `lifespan.shutdown` event

    except BaseException:
        exc_text = traceback.format_exc()
        if started:
            await send({"type": "lifespan.shutdown.failed", "message": exc_text})
        else:
            await send({"type": "lifespan.startup.failed", "message": exc_text})
        raise

    else:
        await send({"type": "lifespan.shutdown.complete"})

async def not_found(self, scope: "Scope", receive: "Receive", send: "Send") -> None:
    not_found_msg = "App doesn't support regular HTTP protocol."

    if scope["type"] == "websocket":
        websocket_close = WebSocketClose(
            code=1000,
            reason=not_found_msg,
        )
        await websocket_close(scope, receive, send)
        return

    response = AsgiResponse(
        body=not_found_msg.encode(),
        status_code=404,
    )

    await response(scope, receive, send)