sova_core/app/
hooks.rs

1use super::App;
2use crate::error::Result;
3use crate::handler::BoxFuture;
4use crate::request::Request;
5use crate::response::Response;
6use crate::router::{compile_router, CompiledRouter};
7use crate::service::BoxedService;
8use crate::state::StateMap;
9use crate::upgrade::UpgradeBudget;
10use bytes::Bytes;
11use http::Method;
12use std::sync::Arc;
13use std::time::Duration;
14use tokio::sync::Semaphore;
15
16pub(crate) type StartupHook = Arc<dyn Fn(Arc<StateMap>) -> BoxFuture<Result<()>> + Send + Sync>;
17pub(crate) type ShutdownHook = Arc<dyn Fn() -> BoxFuture<()> + Send + Sync>;
18
19/// Compiled app ready to handle requests without recompiling the router.
20#[derive(Clone)]
21pub struct Server {
22    pub(crate) inner: Arc<AppInner>,
23    pub(crate) startups: Vec<StartupHook>,
24    pub(crate) shutdowns: Vec<ShutdownHook>,
25}
26
27impl Server {
28    /// Shared application state map (same as request `state`).
29    pub fn state(&self) -> Arc<StateMap> {
30        self.inner.state()
31    }
32
33    /// Run startup hooks without consuming them (safe to call multiple times).
34    #[cfg(any(test, feature = "testing"))]
35    pub async fn run_startup(&self) -> Result<Arc<StateMap>> {
36        let state = self.state();
37        for hook in &self.startups {
38            hook(Arc::clone(&state)).await?;
39        }
40        Ok(state)
41    }
42
43    /// Run shutdown hooks without consuming them.
44    #[cfg(any(test, feature = "testing"))]
45    pub async fn run_shutdown(&self) {
46        for hook in &self.shutdowns {
47            hook().await;
48        }
49    }
50
51    /// Handle a request (tests / embedded). Injects router state.
52    pub async fn handle(&self, mut req: Request) -> Response {
53        if let Some(d) = self.inner.state().get::<crate::dispatch::AppDispatch>() {
54            d.install_if_needed(&self.inner);
55        }
56        req.state = Arc::clone(&self.inner.compiled.state);
57        self.inner.compiled.dispatch(req).await
58    }
59
60    /// Convenience for unit tests without headers.
61    /// Custom headers: [`Request::builder`] + [`Self::handle`].
62    pub async fn handle_request(&self, method: Method, path: &str, body: &str) -> Response {
63        let req = Request::builder()
64            .method(method)
65            .path(path)
66            .body(Bytes::from(body.to_string()))
67            .build();
68        self.handle(req).await
69    }
70}
71
72pub(crate) struct ListenParts {
73    pub(crate) inner: AppInner,
74    pub(crate) startups: Vec<StartupHook>,
75    pub(crate) shutdowns: Vec<ShutdownHook>,
76    pub(crate) services: Vec<BoxedService>,
77    /// When false (CLI default), BackgroundServices are not started.
78    pub(crate) start_services: bool,
79}
80
81impl App {
82    fn validate_plugin_requires(&self) -> Result<()> {
83        if let Some((plugin, dep)) = self.missing_plugin_requires.first().copied() {
84            return Err(crate::error::Error::Internal(format!(
85                "plugin `{plugin}` requires `{dep}`; install `{dep}` before `{plugin}`"
86            )));
87        }
88        if let Some(id) = self.duplicate_plugin_ids.first().copied() {
89            return Err(crate::error::Error::Internal(format!(
90                "plugin `{id}` already installed; customize before first install or build App::new() with an explicit stack (re-install does not replace)"
91            )));
92        }
93        if let Some(msg) = self.plugin_sdk_errors.first() {
94            return Err(crate::error::Error::Internal(msg.clone()));
95        }
96        Ok(())
97    }
98
99    /// Compile routes once into a [`Server`]. Prefer this over repeated [`App::handle`].
100    pub fn build(&self) -> Result<Server> {
101        self.validate_plugin_requires()?;
102        let router = self.router.clone_for_compile();
103        router
104            .check_route_values(&self.router.state, &self.installed_plugins)
105            .map_err(crate::error::Error::Internal)?;
106        let explain = router.explain();
107        let route_count = router.route_entries().len();
108        let compiled = Arc::new(compile_router(router)?);
109        Ok(Server {
110            inner: Arc::new(AppInner::from_settings(
111                compiled,
112                route_count,
113                explain,
114                AppSettings::from(self),
115            )),
116            startups: self.on_startup.clone(),
117            shutdowns: self.on_shutdown.clone(),
118        })
119    }
120
121    pub(crate) fn into_listen_parts(mut self) -> Result<ListenParts> {
122        self.validate_plugin_requires()?;
123        let services = std::mem::take(&mut self.services);
124        let startups = self.on_startup.clone();
125        let shutdowns = self.on_shutdown.clone();
126        let start_services = !self.cli_mode || self.service_in_cli;
127
128        self.router
129            .check_route_values(&self.router.state, &self.installed_plugins)
130            .map_err(crate::error::Error::Internal)?;
131        let explain = self.router.explain();
132        let route_count = self.router.route_entries().len();
133        let settings = AppSettings::from(&self);
134        let router = self.router;
135        let compiled = Arc::new(compile_router(router)?);
136
137        Ok(ListenParts {
138            inner: AppInner::from_settings(compiled, route_count, explain, settings),
139            startups,
140            shutdowns,
141            services,
142            start_services,
143        })
144    }
145}
146
147pub(crate) struct AppSettings {
148    pub max_body_size: usize,
149    pub max_connections: usize,
150    pub max_upgraded_connections: usize,
151    pub max_concurrent_streams: usize,
152    pub max_headers: usize,
153    pub max_buf_size: Option<usize>,
154    pub request_timeout: Option<Duration>,
155    pub header_read_timeout: Duration,
156    pub idle_timeout: Duration,
157    pub drain_timeout: Duration,
158    pub keep_alive: bool,
159    pub trust_proxy: bool,
160    pub reuseport: bool,
161    pub hsts: bool,
162    pub alt_svc: Option<String>,
163}
164
165impl From<&App> for AppSettings {
166    fn from(app: &App) -> Self {
167        Self {
168            max_body_size: app.max_body_size,
169            max_connections: app.max_connections,
170            max_upgraded_connections: app.max_upgraded_connections,
171            max_concurrent_streams: app.max_concurrent_streams,
172            max_headers: app.max_headers,
173            max_buf_size: app.max_buf_size,
174            request_timeout: app.request_timeout,
175            header_read_timeout: app.header_read_timeout,
176            idle_timeout: app.idle_timeout,
177            drain_timeout: app.drain_timeout,
178            keep_alive: app.keep_alive,
179            trust_proxy: app.trust_proxy,
180            reuseport: app.reuseport,
181            hsts: app.hsts,
182            alt_svc: app.alt_svc.clone(),
183        }
184    }
185}
186
187pub(crate) struct AppInner {
188    pub(crate) compiled: Arc<CompiledRouter>,
189    pub(crate) max_body_size: usize,
190    pub(crate) max_connections: usize,
191    pub(crate) max_upgraded: UpgradeBudget,
192    pub(crate) max_concurrent_streams: usize,
193    pub(crate) max_headers: usize,
194    pub(crate) max_buf_size: Option<usize>,
195    pub(crate) request_timeout: Option<Duration>,
196    pub(crate) header_read_timeout: Duration,
197    pub(crate) idle_timeout: Duration,
198    pub(crate) drain_timeout: Duration,
199    pub(crate) keep_alive: bool,
200    pub(crate) trust_proxy: bool,
201    pub(crate) reuseport: bool,
202    pub(crate) hsts: bool,
203    pub(crate) alt_svc: Option<String>,
204    pub(crate) route_count: usize,
205    pub(crate) explain: String,
206}
207
208impl AppInner {
209    fn from_settings(
210        compiled: Arc<CompiledRouter>,
211        route_count: usize,
212        explain: String,
213        s: AppSettings,
214    ) -> Self {
215        Self {
216            compiled,
217            max_body_size: s.max_body_size,
218            max_connections: s.max_connections,
219            max_upgraded: UpgradeBudget(Arc::new(Semaphore::new(s.max_upgraded_connections))),
220            max_concurrent_streams: s.max_concurrent_streams,
221            max_headers: s.max_headers,
222            max_buf_size: s.max_buf_size,
223            request_timeout: s.request_timeout,
224            header_read_timeout: s.header_read_timeout,
225            idle_timeout: s.idle_timeout,
226            drain_timeout: s.drain_timeout,
227            keep_alive: s.keep_alive,
228            trust_proxy: s.trust_proxy,
229            reuseport: s.reuseport,
230            hsts: s.hsts,
231            alt_svc: s.alt_svc,
232            route_count,
233            explain,
234        }
235    }
236
237    pub(crate) async fn handle(&self, req: Request) -> Response {
238        self.compiled.dispatch(req).await
239    }
240
241    pub(crate) fn state(&self) -> Arc<StateMap> {
242        Arc::clone(&self.compiled.state)
243    }
244
245    pub(crate) fn conn_header_timeout(&self) -> Duration {
246        self.header_read_timeout.min(self.idle_timeout)
247    }
248}
sova_core/app/hooks.rs

sova_core/app/
hooks.rs
