rws_server/async_state/
mod.rs

1//! Async-capable state-aware application — requires the `http2` feature (tokio).
2//!
3//! [`AsyncAppWithState<S>`] is the async counterpart to [`AppWithState<S>`]:
4//! handlers are `async fn` closures that can `await` database queries, HTTP
5//! clients, or any other async I/O without blocking the OS thread.
6//!
7//! Where a handler runs:
8//!
9//! - The async servers (HTTP/1.1, HTTP/2, HTTP/3) await it through
10//!   [`Application::execute_async`] when nothing synchronous wraps the app
11//!   (every middleware layer [`is_passthrough`]): on the connection's own
12//!   runtime when that's the shared `multi_thread` one, on the process-wide
13//!   `rws-async` runtime under per-core (`current_thread`) runtimes.
14//! - Otherwise — a middleware layer in front, the HTTP/1.1-only build,
15//!   [`TestClient::send`], any direct [`Application::execute`] call — the
16//!   calling thread waits while it runs on `rws-async`.
17//!
18//! Either way the runtime outlives the request, so tasks a handler leaves
19//! behind run: a `DbPool` gets its connection back (sqlx returns it from a
20//! spawned task) instead of closing it. Handlers must not block their
21//! thread — they share worker threads with other requests.
22//!
23//! Unmatched routes fall through to the built-in [`App`] controller chain,
24//! synchronously, without touching a runtime.
25//!
26//! [`is_passthrough`]: crate::middleware::Middleware::is_passthrough
27//! [`TestClient::send`]: crate::test_client::TestRequest::send
28//!
29//! # Example
30//!
31//! ```rust,no_run
32//! # extern crate rws_server as rust_web_server;
33//! use std::sync::Arc;
34//! use rust_web_server::async_state::AsyncAppWithState;
35//! use rust_web_server::core::New;
36//! use rust_web_server::response::{Response, STATUS_CODE_REASON_PHRASE};
37//! use rust_web_server::range::Range;
38//! use rust_web_server::mime_type::MimeType;
39//! use rust_web_server::router::PathParams;
40//! use rust_web_server::request::Request;
41//! use rust_web_server::server::ConnectionInfo;
42//!
43//! struct AppState {
44//!     greeting: String,
45//! }
46//!
47//! let app = AsyncAppWithState::new(AppState { greeting: "Hello".to_string() })
48//!     .get("/greet/:name", |_req, params, _conn, state| async move {
49//!         let name = params.get("name").unwrap_or("world");
50//!         let body = format!("{}, {}!", state.greeting, name);
51//!         let mut r = Response::new();
52//!         r.status_code = *STATUS_CODE_REASON_PHRASE.n200_ok.status_code;
53//!         r.reason_phrase = STATUS_CODE_REASON_PHRASE.n200_ok.reason_phrase.to_string();
54//!         r.content_range_list = vec![
55//!             Range::get_content_range(body.into_bytes(), MimeType::TEXT_PLAIN.to_string())
56//!         ];
57//!         r
58//!     });
59//! ```
60//!
61//! [`AppWithState<S>`]: crate::state::AppWithState
62
63#[cfg(test)]
64mod tests;
65
66use std::future::Future;
67use std::pin::Pin;
68use std::sync::Arc;
69
70use crate::app::App;
71use crate::application::Application;
72use crate::core::New;
73use crate::request::Request;
74use crate::response::Response;
75use crate::router::matcher::{self, Segment};
76use crate::router::PathParams;
77use crate::server::ConnectionInfo;
78#[cfg(feature = "openapi")]
79use crate::mime_type::MimeType;
80#[cfg(feature = "openapi")]
81use crate::range::Range;
82#[cfg(feature = "openapi")]
83use crate::response::STATUS_CODE_REASON_PHRASE;
84
85type BoxFuture<T> = Pin<Box<dyn Future<Output = T> + Send + 'static>>;
86
87type AsyncHandlerFn<S> = Arc<
88    dyn Fn(Request, PathParams, ConnectionInfo, Arc<S>) -> BoxFuture<Response> + Send + Sync,
89>;
90
91// ── AsyncRoute ────────────────────────────────────────────────────────────────
92
93#[derive(Clone)]
94struct AsyncRoute<S> {
95    method: String,
96    segments: Vec<Segment>,
97    handler: AsyncHandlerFn<S>,
98}
99
100// ── AsyncAppWithState ─────────────────────────────────────────────────────────
101
102/// An [`Application`] whose route handlers are `async` functions.
103///
104/// State is stored as `Arc<S>` and passed by value (cheap clone) to each
105/// handler invocation. Handlers receive owned `Request`, `PathParams`, and
106/// `ConnectionInfo` values so the returned future is `'static`.
107#[derive(Clone)]
108pub struct AsyncAppWithState<S> {
109    state: Arc<S>,
110    routes: Vec<AsyncRoute<S>>,
111    /// When `Some`, the fallback `App` is pinned to this config (see
112    /// [`App::with_config`]); when `None`, the fallback reads
113    /// `RWS_CONFIG_*` env vars per request via `App::new()`, same as `App`'s
114    /// own default.
115    config: Option<Arc<crate::server_config::ServerConfig>>,
116}
117
118impl<S: Send + Sync + 'static> AsyncAppWithState<S> {
119    /// Create a new `AsyncAppWithState` wrapping `state`.
120    pub fn new(state: S) -> Self {
121        AsyncAppWithState { state: Arc::new(state), routes: Vec::new(), config: None }
122    }
123
124    /// Pin the fallback [`App`] (used for any request this app's own routes
125    /// don't match) to an explicit [`crate::server_config::ServerConfig`],
126    /// instead of reading `RWS_CONFIG_*` environment variables per request.
127    ///
128    /// Mirrors [`App::with_config`] / [`crate::state::AppWithState::with_config`].
129    pub fn with_config(mut self, config: crate::server_config::ServerConfig) -> Self {
130        self.config = Some(Arc::new(config));
131        self
132    }
133
134    /// Return a reference to the shared state.
135    pub fn state(&self) -> &S {
136        &self.state
137    }
138
139    /// Return a snapshot of all registered routes as `(method, pattern)` pairs.
140    pub fn route_entries(&self) -> Vec<crate::router::RouteInfo> {
141        self.routes
142            .iter()
143            .map(|r| crate::router::RouteInfo {
144                method: r.method.clone(),
145                pattern: matcher::segments_to_pattern(&r.segments),
146            })
147            .collect()
148    }
149
150    /// The fallback `App` for requests this app's own routes don't match —
151    /// pinned to `self.config` if set, otherwise `App::new()`'s default
152    /// per-request env read.
153    fn fallback_app(&self) -> App {
154        match &self.config {
155            Some(c) => App::with_config((**c).clone()),
156            None => App::new(),
157        }
158    }
159
160    fn add<F, Fut>(mut self, method: &str, pattern: &str, handler: F) -> Self
161    where
162        F: Fn(Request, PathParams, ConnectionInfo, Arc<S>) -> Fut + Send + Sync + 'static,
163        Fut: Future<Output = Response> + Send + 'static,
164    {
165        self.routes.push(AsyncRoute {
166            method: method.to_string(),
167            segments: matcher::parse_pattern(pattern),
168            handler: Arc::new(move |req, params, conn, state| Box::pin(handler(req, params, conn, state))),
169        });
170        if let Some((last, earlier)) = self.routes.split_last() {
171            let earlier = earlier.iter().map(|r| (r.method.as_str(), r.segments.as_slice()));
172            if let Some(conflict) = crate::router::conflict_with(earlier, &last.method, &last.segments) {
173                conflict.warn_once();
174            }
175        }
176        self
177    }
178
179    /// Routes that never run because an earlier one with the same method
180    /// matches every path they do (18.28.0). See
181    /// [`Router::route_conflicts`](crate::router::Router::route_conflicts);
182    /// each is also printed once to stderr when it's registered.
183    pub fn route_conflicts(&self) -> Vec<crate::router::RouteConflict> {
184        crate::router::conflicts(self.routes.iter().map(|r| (r.method.as_str(), r.segments.as_slice())))
185    }
186
187    /// Register a typed handler (18.28.0): an `async fn` (or a sync
188    /// function) with extractor arguments annotated `#[get("/path")]` &
189    /// co.; the method and pattern come from the attribute.
190    ///
191    /// ```rust,ignore
192    /// #[get("/posts/{id}")]
193    /// async fn show_post(Path(id): Path<i64>, State(pool): State<DbPool>) -> Result<Json<Post>, AppError> {
194    ///     // … .await …
195    /// }
196    ///
197    /// let app = App::with_async_state(pool).route(show_post);
198    /// ```
199    pub fn route<H: crate::handler::AsyncHandler<S>>(self, handler: H) -> Self {
200        let handler = Arc::new(handler);
201        self.add(H::METHOD, H::PATTERN, move |request, params, connection, state| {
202            handler.handle_async(request, params, connection, state)
203        })
204    }
205
206    /// Register routes under `prefix` that share guards (18.35.0, #94);
207    /// see [`group`](crate::group).
208    pub fn group(mut self, prefix: &str, build: impl FnOnce(crate::group::AsyncGroup<S>) -> crate::group::AsyncGroup<S>) -> Self {
209        for route in build(crate::group::AsyncGroup::new(prefix)).finish() {
210            self = self.add_group_route(route);
211        }
212        self
213    }
214
215    /// Register the conventional routes of a resource at `path` (18.35.0,
216    /// #94); see [`group`](crate::group).
217    pub fn resource(mut self, path: &str, build: impl FnOnce(crate::group::AsyncResource<S>) -> crate::group::AsyncResource<S>) -> Self {
218        for route in crate::group::AsyncGroup::new("").resource(path, build).finish() {
219            self = self.add_group_route(route);
220        }
221        self
222    }
223
224    /// Name the route registered last, for
225    /// [`url_for`](crate::router::url_for) (18.35.0, #94).
226    pub fn name(self, name: &str) -> Self {
227        if let Some(route) = self.routes.last() {
228            crate::router::name_route(name, &matcher::segments_to_pattern(&route.segments));
229        }
230        self
231    }
232
233    fn add_group_route(self, route: crate::group::GroupRoute<crate::group::AsyncHandlerFn<S>>) -> Self {
234        if let Some(name) = &route.name {
235            crate::router::name_route(name, &route.pattern);
236        }
237        let handler = route.handler;
238        self.add(route.method, &route.pattern, move |request, params, connection, state| handler(request, params, connection, state))
239    }
240
241    /// Wrap this application in a middleware layer (18.35.0), as
242    /// [`AppWithState::wrap`](crate::state::AppWithState::wrap): the first
243    /// layer wrapped is the outermost.
244    pub fn wrap<M: crate::middleware::Middleware + 'static>(self, layer: M) -> crate::middleware::WithMiddleware<AsyncAppWithState<S>> {
245        crate::middleware::WithMiddleware::new(self).wrap(layer)
246    }
247
248    /// Register an async `GET` handler for `pattern`.
249    pub fn get<F, Fut>(self, pattern: &str, handler: F) -> Self
250    where
251        F: Fn(Request, PathParams, ConnectionInfo, Arc<S>) -> Fut + Send + Sync + 'static,
252        Fut: Future<Output = Response> + Send + 'static,
253    {
254        self.add("GET", pattern, handler)
255    }
256
257    /// Register an async `POST` handler for `pattern`.
258    pub fn post<F, Fut>(self, pattern: &str, handler: F) -> Self
259    where
260        F: Fn(Request, PathParams, ConnectionInfo, Arc<S>) -> Fut + Send + Sync + 'static,
261        Fut: Future<Output = Response> + Send + 'static,
262    {
263        self.add("POST", pattern, handler)
264    }
265
266    /// Register an async `PUT` handler for `pattern`.
267    pub fn put<F, Fut>(self, pattern: &str, handler: F) -> Self
268    where
269        F: Fn(Request, PathParams, ConnectionInfo, Arc<S>) -> Fut + Send + Sync + 'static,
270        Fut: Future<Output = Response> + Send + 'static,
271    {
272        self.add("PUT", pattern, handler)
273    }
274
275    /// Register an async `PATCH` handler for `pattern`.
276    pub fn patch<F, Fut>(self, pattern: &str, handler: F) -> Self
277    where
278        F: Fn(Request, PathParams, ConnectionInfo, Arc<S>) -> Fut + Send + Sync + 'static,
279        Fut: Future<Output = Response> + Send + 'static,
280    {
281        self.add("PATCH", pattern, handler)
282    }
283
284    /// Register an async `DELETE` handler for `pattern`.
285    pub fn delete<F, Fut>(self, pattern: &str, handler: F) -> Self
286    where
287        F: Fn(Request, PathParams, ConnectionInfo, Arc<S>) -> Fut + Send + Sync + 'static,
288        Fut: Future<Output = Response> + Send + 'static,
289    {
290        self.add("DELETE", pattern, handler)
291    }
292
293    /// Add `GET /openapi.json` (a generated OpenAPI 3.1 document covering
294    /// every route registered so far) and `GET /docs` (Swagger UI, loaded
295    /// from a CDN, pointed at `/openapi.json`).
296    ///
297    /// Call this *after* registering your routes — routes added afterward
298    /// still work but won't appear in the generated spec, since it's built
299    /// once at this call rather than read dynamically per request.
300    ///
301    /// Requires the `openapi` feature (and `http2`, since `AsyncAppWithState`
302    /// already requires it).
303    #[cfg(feature = "openapi")]
304    pub fn openapi(self, config: crate::openapi::OpenApiConfig) -> Self {
305        let spec_json = Arc::new(crate::openapi::build_spec(&config, &self.route_entries()));
306        let html = Arc::new(crate::openapi::swagger_ui_html("/openapi.json"));
307
308        let spec_for_route = Arc::clone(&spec_json);
309        self.get("/openapi.json", move |_req, _params, _conn, _state| {
310            let spec_for_route = Arc::clone(&spec_for_route);
311            async move {
312                let mut r = Response::new();
313                r.status_code = *STATUS_CODE_REASON_PHRASE.n200_ok.status_code;
314                r.reason_phrase = STATUS_CODE_REASON_PHRASE.n200_ok.reason_phrase.to_string();
315                r.content_range_list = vec![Range::get_content_range(
316                    spec_for_route.as_bytes().to_vec(),
317                    MimeType::APPLICATION_JSON.to_string(),
318                )];
319                r
320            }
321        })
322        .get("/docs", move |_req, _params, _conn, _state| {
323            let html = Arc::clone(&html);
324            async move {
325                let mut r = Response::new();
326                r.status_code = *STATUS_CODE_REASON_PHRASE.n200_ok.status_code;
327                r.reason_phrase = STATUS_CODE_REASON_PHRASE.n200_ok.reason_phrase.to_string();
328                r.content_range_list = vec![Range::get_content_range(
329                    html.as_bytes().to_vec(),
330                    MimeType::TEXT_HTML.to_string(),
331                )];
332                r
333            }
334        })
335    }
336
337    /// The matching route's handler future, or `None` when no route matches.
338    fn dispatch(&self, request: &Request, connection: &ConnectionInfo) -> Option<BoxFuture<Response>> {
339        let path = request.request_uri.split('?').next().unwrap_or(&request.request_uri);
340        let path_segs: Vec<&str> = path.split('/').filter(|s| !s.is_empty()).collect();
341
342        for route in &self.routes {
343            if route.method != request.method {
344                continue;
345            }
346            if let Some(params_map) = matcher::try_match(&route.segments, &path_segs) {
347                let params = PathParams::from_map(params_map);
348                return Some((route.handler)(request.clone(), params, connection.clone(), Arc::clone(&self.state)));
349            }
350        }
351        None
352    }
353
354    /// Await the response on the caller's runtime (see the module docs for
355    /// where the handler runs); unmatched routes go to the fallback `App`.
356    pub(crate) async fn respond(&self, request: &Request, connection: &ConnectionInfo) -> Result<Response, String> {
357        match self.dispatch(request, connection) {
358            Some(handler) => Ok(crate::async_bridge::run_async(handler).await),
359            None => self.fallback_app().execute(request, connection),
360        }
361    }
362}
363
364impl<S: Send + Sync + 'static> Application for AsyncAppWithState<S> {
365    /// Runs a matching handler on the shared `rws-async` runtime and waits
366    /// for it; anything else goes to the fallback `App` on this thread.
367    fn execute(&self, request: &Request, connection: &ConnectionInfo) -> Result<Response, String> {
368        match self.dispatch(request, connection) {
369            Some(handler) => Ok(crate::async_bridge::block_on_shared(handler)),
370            None => self.fallback_app().execute(request, connection),
371        }
372    }
373
374    /// A matching route's handler, awaited by the async servers; `None` for
375    /// the fallback `App` (it's synchronous).
376    fn execute_async<'a>(&'a self, request: &'a Request, connection: &'a ConnectionInfo) -> Option<crate::application::AsyncResponse<'a>> {
377        let handler = self.dispatch(request, connection)?;
378        Some(Box::pin(async move { Ok(crate::async_bridge::run_async(handler).await) }))
379    }
380
381    fn registered_routes(&self) -> Vec<crate::router::RouteInfo> {
382        self.route_entries()
383    }
384}
rws_server/async_state/mod.rs

rws_server/async_state/
mod.rs
