churust_core/
engine.rs

1//! The hyper-based HTTP/1.1 serving engine that drives an [`App`] over a real
2//! socket.
3//!
4//! Most applications never call into this module directly — [`App::start`] and
5//! [`App::start_with_shutdown`] bind a listener and delegate to [`serve`]. It is
6//! public so advanced callers can drive serving with a custom address and
7//! shutdown signal.
8
9use crate::app::App;
10use crate::body::Body;
11#[cfg(feature = "tls")]
12use crate::tls::acceptor_from_pem;
13use bytes::Bytes;
14use futures_util::StreamExt;
15use http_body_util::{combinators::UnsyncBoxBody, BodyExt, Full, Limited, StreamBody};
16use hyper::body::Frame;
17use hyper::body::Incoming;
18use hyper::service::service_fn;
19use hyper::{Request as HyperRequest, Response as HyperResponse, StatusCode};
20use hyper_util::rt::TokioIo;
21use std::convert::Infallible;
22use std::future::Future;
23use std::net::SocketAddr;
24use tokio::net::TcpListener;
25
26/// Convert a [`Body`] into the boxed hyper body the engine writes to the wire:
27/// `Full` for a buffered payload, `StreamBody` for a lazy stream.
28///
29/// The boxed body is the `!Sync` [`UnsyncBoxBody`] rather than the plan's
30/// `BoxBody`: `Body::Stream` wraps a `Pin<Box<dyn Stream + Send>>` (intentionally
31/// `Send` but not `Sync`), and the resolved `http-body-util` 0.1.3 `BodyExt::boxed`
32/// requires `Self: Send + Sync`. `boxed_unsync` only requires `Send`, and hyper
33/// accepts any `http_body::Body` response body, so behavior is unchanged.
34fn into_boxed_body(body: Body) -> UnsyncBoxBody<Bytes, std::io::Error> {
35    match body {
36        Body::Bytes(bytes) => Full::new(bytes)
37            .map_err(|never| match never {})
38            .boxed_unsync(),
39        Body::Stream(stream) => {
40            let frames = stream.map(|chunk| {
41                chunk
42                    .map(Frame::data)
43                    .map_err(|e| std::io::Error::other(e.to_string()))
44            });
45            StreamBody::new(frames).boxed_unsync()
46        }
47    }
48}
49
50/// Serve `app` on `addr` until `shutdown` resolves (graceful drain).
51///
52/// Uses HTTP/1.1 (`hyper::server::conn::http1::Builder`).  The plan
53/// referenced `auto::Builder` but that type's `serve_connection` returns a
54/// `Connection<'_, …>` that borrows the builder, which cannot be moved into a
55/// `tokio::spawn` closure.  Using `http1::Builder` directly returns an owned
56/// `Connection<I, S>` that is `'static`, compiles correctly, and still
57/// satisfies the HTTP/1.1-only requirement of Plan 1.
58///
59/// Prefer [`App::start`] / [`App::start_with_shutdown`] unless you need to
60/// control the bind address and shutdown future yourself.
61///
62/// [`App::start`]: crate::App::start
63/// [`App::start_with_shutdown`]: crate::App::start_with_shutdown
64///
65/// ```no_run
66/// use churust_core::{Churust, Call, engine};
67/// # tokio::runtime::Runtime::new().unwrap().block_on(async {
68/// let app = Churust::server()
69///     .routing(|r| { r.get("/", |_c: Call| async { "hi" }); })
70///     .build();
71/// let addr = "127.0.0.1:8080".parse().unwrap();
72/// // Serve until Ctrl-C.
73/// engine::serve(app, addr, async { let _ = tokio::signal::ctrl_c().await; }).await?;
74/// # Ok::<(), std::io::Error>(())
75/// # });
76/// ```
77pub async fn serve<F>(app: App, addr: SocketAddr, shutdown: F) -> std::io::Result<()>
78where
79    F: Future<Output = ()> + Send + 'static,
80{
81    let listener = TcpListener::bind(addr).await?;
82    let max_body = app.config().max_body_bytes;
83    let timeout_ms = app.config().request_timeout_ms;
84    let graceful = hyper_util::server::graceful::GracefulShutdown::new();
85    tokio::pin!(shutdown);
86
87    #[cfg(feature = "tls")]
88    let tls_acceptor = match &app.config().tls {
89        Some(t) => Some(acceptor_from_pem(&t.cert, &t.key)?),
90        None => None,
91    };
92
93    loop {
94        tokio::select! {
95            accepted = listener.accept() => {
96                let (stream, _peer) = match accepted {
97                    Ok(s) => s,
98                    Err(_) => continue,
99                };
100                #[cfg(feature = "tls")]
101                {
102                    if let Some(acceptor) = tls_acceptor.clone() {
103                        let app = app.clone();
104                        // `Watcher` is the owned counterpart of `GracefulShutdown`
105                        // (`watcher()` exists precisely so it can be moved onto
106                        // another task before `watch()`). We hand it to the
107                        // spawned handshake task so the TLS connection is still
108                        // drained on graceful shutdown — without blocking the
109                        // accept loop on the (potentially slow) TLS handshake.
110                        let watcher = graceful.watcher();
111                        let conn_builder_fut = async move {
112                            // A failed TLS handshake silently drops the connection.
113                            if let Ok(tls_stream) = acceptor.accept(stream).await {
114                                serve_stream(app, TokioIo::new(tls_stream), max_body, timeout_ms, watcher).await;
115                            }
116                        };
117                        tokio::spawn(conn_builder_fut);
118                        continue;
119                    }
120                }
121                serve_stream(app.clone(), TokioIo::new(stream), max_body, timeout_ms, graceful.watcher()).await;
122            }
123            _ = &mut shutdown => {
124                break;
125            }
126        }
127    }
128
129    graceful.shutdown().await;
130    Ok(())
131}
132
133async fn serve_stream<I>(
134    app: App,
135    io: I,
136    max_body: usize,
137    timeout_ms: u64,
138    watcher: hyper_util::server::graceful::Watcher,
139) where
140    I: hyper::rt::Read + hyper::rt::Write + Unpin + Send + 'static,
141{
142    let svc = service_fn(move |req: HyperRequest<Incoming>| {
143        let app = app.clone();
144        async move { handle(app, req, max_body, timeout_ms).await }
145    });
146    let conn = hyper::server::conn::http1::Builder::new().serve_connection(io, svc);
147    // Under the `ws` feature the connection is made upgradeable. The
148    // resolved `hyper-util` (0.1.x) does not implement `GracefulConnection`
149    // for `http1::UpgradeableConnection`, so `watcher.watch(conn)` rejects it.
150    // Per the plan's implementer note we fall back to awaiting the connection
151    // directly for the ws build (dropping graceful drain only for WS
152    // connections) while keeping the non-ws path exactly as-is.
153    #[cfg(feature = "ws")]
154    {
155        let _ = &watcher;
156        let conn = conn.with_upgrades();
157        tokio::spawn(async move {
158            let _ = conn.await;
159        });
160    }
161    #[cfg(not(feature = "ws"))]
162    {
163        let fut = watcher.watch(conn);
164        tokio::spawn(async move {
165            let _ = fut.await;
166        });
167    }
168}
169
170async fn handle(
171    app: App,
172    req: HyperRequest<Incoming>,
173    max_body: usize,
174    timeout_ms: u64,
175) -> Result<HyperResponse<UnsyncBoxBody<Bytes, std::io::Error>>, Infallible> {
176    #[cfg(feature = "ws")]
177    let mut req = req;
178    #[cfg(feature = "ws")]
179    let on_upgrade = if crate::ws::is_upgrade_request(req.headers()) {
180        Some(hyper::upgrade::on(&mut req))
181    } else {
182        None
183    };
184
185    let (parts, body) = req.into_parts();
186
187    // Enforce max body size before buffering.
188    let collected = Limited::new(body, max_body).collect().await;
189    let body_bytes = match collected {
190        Ok(buf) => buf.to_bytes(),
191        Err(_) => {
192            let mut resp = HyperResponse::new(
193                Full::new(Bytes::from("Payload Too Large"))
194                    .map_err(|never| match never {})
195                    .boxed_unsync(),
196            );
197            *resp.status_mut() = StatusCode::PAYLOAD_TOO_LARGE;
198            return Ok(resp);
199        }
200    };
201
202    #[cfg(feature = "ws")]
203    let process = {
204        let mut extensions = http::Extensions::new();
205        if let Some(on_upgrade) = on_upgrade {
206            extensions.insert(crate::ws::OnUpgradeHandle::new(on_upgrade));
207        }
208        app.process_with_extensions(
209            parts.method,
210            parts.uri,
211            parts.headers,
212            body_bytes,
213            extensions,
214        )
215    };
216    #[cfg(not(feature = "ws"))]
217    let process = app.process(parts.method, parts.uri, parts.headers, body_bytes);
218
219    let res = if timeout_ms == 0 {
220        process.await
221    } else {
222        match tokio::time::timeout(std::time::Duration::from_millis(timeout_ms), process).await {
223            Ok(res) => res,
224            Err(_) => crate::response::Response::text("Request Timeout")
225                .with_status(StatusCode::REQUEST_TIMEOUT),
226        }
227    };
228
229    let mut builder = HyperResponse::builder().status(res.status);
230    if let Some(headers) = builder.headers_mut() {
231        *headers = res.headers;
232    }
233    Ok(builder
234        .body(into_boxed_body(res.body))
235        .expect("response build is infallible"))
236}
churust_core/engine.rs

churust_core/
engine.rs
