use crate::call::Call;
use crate::pipeline::{Endpoint, Middleware, Next, Phase};
use crate::response::Response;
use crate::router::{Match, RouteBuilder, Router};
use crate::state::StateMap;
use bytes::Bytes;
use futures_util::FutureExt;
use http::header::ALLOW;
use http::{HeaderMap, HeaderValue, Method, StatusCode, Uri};
use std::collections::VecDeque;
use std::sync::Arc;
pub trait Plugin {
fn install(self: Box<Self>, app: &mut AppBuilder);
}
#[derive(Clone, Debug)]
pub struct ServerConfig {
pub host: String,
pub port: u16,
pub max_body_bytes: usize,
pub request_timeout_ms: u64,
pub header_read_timeout_ms: u64,
pub max_headers: usize,
pub max_path_segments: usize,
pub ws_max_frame_bytes: usize,
pub ws_max_message_bytes: usize,
pub keep_alive_ms: u64,
pub backlog: u32,
pub shutdown_timeout_ms: u64,
pub path_policy: crate::path::PathPolicy,
pub h2_max_header_list_size: u32,
pub h2_max_concurrent_streams: u32,
pub ws_idle_timeout_ms: u64,
pub max_connections: usize,
pub max_tls_handshakes: usize,
pub tls_handshake_timeout_ms: u64,
pub tls: Option<crate::config::TlsSection>,
}
impl Default for ServerConfig {
fn default() -> Self {
Self {
host: "127.0.0.1".into(),
port: 8080,
max_body_bytes: 1 << 20,
request_timeout_ms: 30_000,
header_read_timeout_ms: 10_000,
max_headers: 100,
max_path_segments: 64,
ws_max_frame_bytes: 1 << 20,
ws_max_message_bytes: 4 << 20,
keep_alive_ms: 75_000,
backlog: 1024,
shutdown_timeout_ms: 30_000,
path_policy: crate::path::PathPolicy::Strict,
h2_max_header_list_size: 16 << 10,
h2_max_concurrent_streams: 200,
ws_idle_timeout_ms: 300_000,
max_connections: 25_000,
max_tls_handshakes: 256,
tls_handshake_timeout_ms: 10_000,
tls: None,
}
}
}
pub struct AppBuilder {
router: Router,
middleware: Vec<(Phase, Arc<dyn Middleware>)>,
config: ServerConfig,
state: StateMap,
security: Option<crate::security::SecurityHeaders>,
extra_binds: Vec<String>,
on_error: Option<ErrorRenderer>,
}
type ErrorRenderer = Arc<dyn Fn(StatusCode, &Call) -> Option<Response> + Send + Sync>;
impl AppBuilder {
fn new() -> Self {
Self {
router: Router::new(),
middleware: Vec::new(),
config: ServerConfig::default(),
state: StateMap::default(),
security: Some(crate::security::SecurityHeaders::default()),
on_error: None,
extra_binds: Vec::new(),
}
}
pub fn host(mut self, host: impl Into<String>) -> Self {
self.config.host = host.into();
self
}
pub fn port(mut self, port: u16) -> Self {
self.config.port = port;
self
}
pub fn max_body_bytes(mut self, n: usize) -> Self {
self.config.max_body_bytes = n;
self
}
pub fn with_config(mut self, cfg: crate::config::Config) -> Self {
self.config.host = cfg.server.host;
self.config.port = cfg.server.port;
self.config.max_body_bytes = cfg.server.max_body_bytes;
self.config.request_timeout_ms = cfg.server.request_timeout_ms;
self.config.header_read_timeout_ms = cfg.server.header_read_timeout_ms;
self.config.max_headers = cfg.server.max_headers;
self.config.max_path_segments = cfg.server.max_path_segments;
self.config.ws_max_frame_bytes = cfg.server.ws_max_frame_bytes;
self.config.ws_max_message_bytes = cfg.server.ws_max_message_bytes;
self.config.keep_alive_ms = cfg.server.keep_alive_ms;
self.config.backlog = cfg.server.backlog;
self.config.shutdown_timeout_ms = cfg.server.shutdown_timeout_ms;
self.config.path_policy = cfg.server.path_policy;
self.config.h2_max_header_list_size = cfg.server.h2_max_header_list_size;
self.config.h2_max_concurrent_streams = cfg.server.h2_max_concurrent_streams;
self.config.ws_idle_timeout_ms = cfg.server.ws_idle_timeout_ms;
self.config.max_connections = cfg.server.max_connections;
self.config.max_tls_handshakes = cfg.server.max_tls_handshakes;
self.config.tls_handshake_timeout_ms = cfg.server.tls_handshake_timeout_ms;
self.config.tls = cfg.tls;
self
}
pub fn request_timeout_ms(mut self, ms: u64) -> Self {
self.config.request_timeout_ms = ms;
self
}
pub fn header_read_timeout_ms(mut self, ms: u64) -> Self {
self.config.header_read_timeout_ms = ms;
self
}
pub fn max_headers(mut self, n: usize) -> Self {
self.config.max_headers = n;
self
}
pub fn keep_alive_ms(mut self, ms: u64) -> Self {
self.config.keep_alive_ms = ms;
self
}
pub fn backlog(mut self, n: u32) -> Self {
self.config.backlog = n;
self
}
pub fn shutdown_timeout_ms(mut self, ms: u64) -> Self {
self.config.shutdown_timeout_ms = ms;
self
}
pub fn path_policy(mut self, policy: crate::path::PathPolicy) -> Self {
self.config.path_policy = policy;
self
}
pub fn h2_max_header_list_size(mut self, n: u32) -> Self {
self.config.h2_max_header_list_size = n;
self
}
pub fn h2_max_concurrent_streams(mut self, n: u32) -> Self {
self.config.h2_max_concurrent_streams = n;
self
}
pub fn ws_idle_timeout_ms(mut self, ms: u64) -> Self {
self.config.ws_idle_timeout_ms = ms;
self
}
pub fn max_connections(mut self, n: usize) -> Self {
self.config.max_connections = n;
self
}
pub fn max_tls_handshakes(mut self, n: usize) -> Self {
self.config.max_tls_handshakes = n;
self
}
pub fn tls_handshake_timeout_ms(mut self, ms: u64) -> Self {
self.config.tls_handshake_timeout_ms = ms;
self
}
pub fn max_path_segments(mut self, n: usize) -> Self {
self.config.max_path_segments = n;
self
}
pub fn tls(mut self, cert_path: impl Into<String>, key_path: impl Into<String>) -> Self {
self.config.tls = Some(crate::config::TlsSection {
cert: cert_path.into(),
key: key_path.into(),
});
self
}
pub fn state<T: Send + Sync + 'static>(mut self, value: T) -> Self {
self.insert_state(value);
self
}
pub fn advertise_http3(mut self, port: u16) -> Self {
self.add_middleware_in(Phase::Setup, Arc::new(AltSvc(port)));
self
}
pub fn insert_state<T: Send + Sync + 'static>(&mut self, value: T) {
self.state.insert(value);
}
pub fn install<P: Plugin + 'static>(mut self, plugin: P) -> Self {
Box::new(plugin).install(&mut self);
self
}
pub fn add_middleware_in(&mut self, phase: Phase, mw: Arc<dyn Middleware>) {
self.middleware.push((phase, mw));
}
pub fn add_middleware(&mut self, mw: Arc<dyn Middleware>) {
self.add_middleware_in(Phase::Plugins, mw);
}
pub fn routing(mut self, f: impl FnOnce(&mut RouteBuilder)) -> Self {
let mut b = RouteBuilder::new(&mut self.router);
f(&mut b);
self
}
pub fn routes(&self) -> &[(Method, String)] {
self.router.routes()
}
pub async fn start(self) -> std::io::Result<()> {
self.build().start().await
}
pub fn on_error<F>(mut self, f: F) -> Self
where
F: Fn(StatusCode, &Call) -> Option<Response> + Send + Sync + 'static,
{
self.on_error = Some(Arc::new(f));
self
}
pub fn install_middleware<M: Middleware>(mut self, mw: M) -> Self {
self.middleware.push((Phase::Plugins, Arc::new(mw)));
self
}
pub fn security_headers(mut self, headers: crate::security::SecurityHeaders) -> Self {
self.security = Some(headers);
self
}
pub fn without_security_headers(mut self) -> Self {
self.security = None;
self
}
pub fn bind(mut self, addr: impl Into<String>) -> Self {
self.extra_binds.push(addr.into());
self
}
#[cfg(unix)]
pub async fn start_unix(self, path: impl AsRef<std::path::Path>) -> std::io::Result<()> {
self.build().start_unix(path).await
}
pub fn build(self) -> App {
let mut mw = self.middleware;
if let Some(render) = self.on_error {
mw.push((
Phase::Monitoring,
Arc::new(ErrorPages { render }) as Arc<dyn Middleware>,
));
}
let security = self.security.clone();
if let Some(headers) = self.security {
let tls_enabled = self.config.tls.is_some();
mw.push((
Phase::Setup,
Arc::new(headers.into_middleware(tls_enabled)) as Arc<dyn Middleware>,
));
}
mw.sort_by_key(|(phase, _)| *phase); let middleware: Vec<Arc<dyn Middleware>> = mw.into_iter().map(|(_, m)| m).collect();
App {
inner: Arc::new(AppInner {
router: self.router,
middleware,
config: self.config,
state: Arc::new(self.state),
extra_binds: self.extra_binds,
security,
}),
}
}
}
struct AppInner {
router: Router,
middleware: Vec<Arc<dyn Middleware>>,
config: ServerConfig,
state: Arc<StateMap>,
extra_binds: Vec<String>,
security: Option<crate::security::SecurityHeaders>,
}
#[derive(Clone)]
pub struct App {
inner: Arc<AppInner>,
}
impl App {
pub fn config(&self) -> &ServerConfig {
&self.inner.config
}
pub(crate) fn apply_security_headers(&self, headers: &mut HeaderMap, over_tls: bool) {
if let Some(security) = &self.inner.security {
security.apply_to(headers, over_tls || self.inner.config.tls.is_some());
}
}
pub async fn process(
&self,
method: Method,
uri: Uri,
headers: HeaderMap,
body: Bytes,
) -> Response {
self.process_with_extensions(method, uri, headers, body, http::Extensions::new())
.await
}
pub async fn process_with_extensions(
&self,
method: Method,
uri: Uri,
headers: HeaderMap,
body: Bytes,
extensions: http::Extensions,
) -> Response {
self.process_call(Call::new(method, uri, headers, body), extensions)
.await
}
pub(crate) async fn process_call(&self, call: Call, extensions: http::Extensions) -> Response {
let app = self.clone();
let fut = async move {
let mut call = call;
call.seed_extensions(extensions);
call.set_state(app.inner.state.clone());
app.run_pipeline(call).await
};
match std::panic::AssertUnwindSafe(fut).catch_unwind().await {
Ok(res) => res,
Err(_) => Response::text("Internal Server Error")
.with_status(StatusCode::INTERNAL_SERVER_ERROR),
}
}
fn bind_addrs(&self) -> std::io::Result<Vec<std::net::SocketAddr>> {
std::iter::once(format!(
"{}:{}",
self.inner.config.host, self.inner.config.port
))
.chain(self.inner.extra_binds.iter().cloned())
.map(|a| {
a.parse::<std::net::SocketAddr>()
.map_err(|e| std::io::Error::new(std::io::ErrorKind::InvalidInput, e))
})
.collect()
}
pub async fn start(self) -> std::io::Result<()> {
self.start_with_shutdown(async {
let _ = tokio::signal::ctrl_c().await;
})
.await
}
pub async fn start_with_shutdown<F>(self, shutdown: F) -> std::io::Result<()>
where
F: std::future::Future<Output = ()> + Send + 'static,
{
let mut addrs = self.bind_addrs()?;
if addrs.len() == 1 {
return crate::engine::serve(self, addrs.remove(0), shutdown).await;
}
crate::engine::serve_many(self, addrs, shutdown).await
}
pub async fn start_on<F>(
self,
listener: tokio::net::TcpListener,
shutdown: F,
) -> std::io::Result<()>
where
F: std::future::Future<Output = ()> + Send + 'static,
{
crate::engine::serve_on(self, listener, shutdown).await
}
#[cfg(unix)]
pub async fn start_unix(self, path: impl AsRef<std::path::Path>) -> std::io::Result<()> {
crate::engine::serve_unix(self, path, async {
let _ = tokio::signal::ctrl_c().await;
})
.await
}
async fn run_pipeline(&self, call: Call) -> Response {
let inner = self.inner.clone();
let endpoint: Endpoint = Arc::new(move |mut call: Call| {
let inner = inner.clone();
Box::pin(async move {
let path = call.path().to_string();
let method = call.method().clone();
if method == Method::OPTIONS && path == "*" {
let value = crate::router::allow_header_value(inner.router.all_methods())
.unwrap_or_else(|| Method::OPTIONS.as_str().to_string());
return Response::new(StatusCode::NO_CONTENT).with_header(
ALLOW,
HeaderValue::from_str(&value).unwrap_or(HeaderValue::from_static("")),
);
}
if let Some(canonical) = crate::path::canonical_path(&path) {
match inner.config.path_policy {
crate::path::PathPolicy::Strict => {
return Response::text("Not Found").with_status(StatusCode::NOT_FOUND);
}
crate::path::PathPolicy::Redirect => {
let target = match call.uri().query() {
Some(q) if !q.is_empty() => format!("{canonical}?{q}"),
_ => canonical,
};
return Response::new(StatusCode::PERMANENT_REDIRECT).with_header(
http::header::LOCATION,
HeaderValue::from_str(&target)
.unwrap_or(HeaderValue::from_static("/")),
);
}
crate::path::PathPolicy::Collapse => {}
}
}
if inner.config.max_path_segments > 0
&& path.split('/').filter(|s| !s.is_empty()).count()
> inner.config.max_path_segments
{
return Response::text("URI Too Long").with_status(StatusCode::URI_TOO_LONG);
}
let mut lookup = inner.router.route(&method, &path, &call);
let mut synthesized_head = false;
if method == Method::HEAD && !matches!(lookup, Match::Found { .. }) {
if let m @ Match::Found { .. } = inner.router.route(&Method::GET, &path, &call)
{
lookup = m;
synthesized_head = true;
}
}
if method == Method::OPTIONS && !matches!(lookup, Match::Found { .. }) {
if let Some(value) =
crate::router::allow_header_value(inner.router.methods_for(&path))
{
return Response::new(StatusCode::NO_CONTENT).with_header(
ALLOW,
HeaderValue::from_str(&value).unwrap_or(HeaderValue::from_static("")),
);
}
}
match lookup {
Match::Found { handler, params } => {
call.set_params(params);
let res = handler.handle(call).await;
if synthesized_head {
strip_body(res)
} else {
res
}
}
Match::MethodNotAllowed { allow } => {
match crate::router::allow_header_value(allow) {
Some(value) => Response::new(StatusCode::METHOD_NOT_ALLOWED)
.with_header(
ALLOW,
HeaderValue::from_str(&value)
.unwrap_or(HeaderValue::from_static("")),
),
None => Response::text("Not Found").with_status(StatusCode::NOT_FOUND),
}
}
Match::NotFound => {
Response::text("Not Found").with_status(StatusCode::NOT_FOUND)
}
Match::BadPath => {
Response::text("Bad Request").with_status(StatusCode::BAD_REQUEST)
}
}
}) as _
});
let chain: VecDeque<Arc<dyn Middleware>> = self.inner.middleware.iter().cloned().collect();
Next::new(chain, endpoint).run(call).await
}
}
struct AltSvc(u16);
#[async_trait::async_trait]
impl Middleware for AltSvc {
async fn handle(&self, call: Call, next: Next) -> Response {
let mut res = next.run(call).await;
if let Ok(value) = http::HeaderValue::from_str(&format!("h3=\":{}\"; ma=86400", self.0)) {
res.headers
.insert(http::header::HeaderName::from_static("alt-svc"), value);
}
res
}
}
struct ErrorPages {
render: ErrorRenderer,
}
#[async_trait::async_trait]
impl Middleware for ErrorPages {
async fn handle(&self, call: Call, next: Next) -> Response {
let snapshot = call.snapshot_for_error();
let res = next.run(call).await;
if res.status.is_client_error() || res.status.is_server_error() {
if let Some(mut replacement) = (self.render)(res.status, &snapshot) {
for name in res.headers.keys() {
if replacement.headers.contains_key(name) {
continue;
}
if name == http::header::CONTENT_LENGTH
|| name == http::header::CONTENT_TYPE
|| name == http::header::CONTENT_ENCODING
{
continue;
}
for value in res.headers.get_all(name) {
replacement.headers.append(name.clone(), value.clone());
}
}
return replacement;
}
}
res
}
}
fn strip_body(mut res: Response) -> Response {
if let Some(bytes) = res.body.as_bytes() {
if let Ok(value) = HeaderValue::from_str(&bytes.len().to_string()) {
res.headers.insert(http::header::CONTENT_LENGTH, value);
}
}
res.body = crate::body::Body::empty();
res
}
pub struct Churust;
impl Churust {
pub fn server() -> AppBuilder {
AppBuilder::new()
}
pub fn from_config() -> AppBuilder {
AppBuilder::new().with_config(crate::config::Config::load_default())
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::pipeline::Next;
use async_trait::async_trait;
#[tokio::test]
async fn middleware_runs_in_phase_order() {
use crate::pipeline::{Next, Phase};
use async_trait::async_trait;
use std::sync::{Arc, Mutex};
#[derive(Clone)]
struct Recorder {
log: Arc<Mutex<Vec<&'static str>>>,
tag: &'static str,
}
#[async_trait]
impl Middleware for Recorder {
async fn handle(&self, call: Call, next: Next) -> Response {
self.log.lock().unwrap().push(self.tag);
next.run(call).await
}
}
let log = Arc::new(Mutex::new(Vec::new()));
let mut builder = Churust::server();
builder.add_middleware_in(
Phase::Fallback,
Arc::new(Recorder {
log: log.clone(),
tag: "fallback",
}),
);
builder.add_middleware_in(
Phase::Setup,
Arc::new(Recorder {
log: log.clone(),
tag: "setup",
}),
);
builder.add_middleware_in(
Phase::Monitoring,
Arc::new(Recorder {
log: log.clone(),
tag: "monitoring",
}),
);
let app = builder
.routing(|r| {
r.get("/", |_c: Call| async { "ok" });
})
.build();
let _ = get(&app, "/").await;
assert_eq!(
*log.lock().unwrap(),
vec!["setup", "monitoring", "fallback"]
);
}
fn app() -> App {
Churust::server()
.routing(|r| {
r.get("/", |_c: Call| async { "home" });
r.get("/boom", |_c: Call| async {
panic!("handler exploded");
#[allow(unreachable_code)]
""
});
})
.build()
}
async fn get(app: &App, path: &str) -> Response {
app.process(
Method::GET,
path.parse::<Uri>().unwrap(),
HeaderMap::new(),
Bytes::new(),
)
.await
}
#[tokio::test]
async fn routes_to_handler() {
let res = get(&app(), "/").await;
assert_eq!(res.status, StatusCode::OK);
assert_eq!(res.body, Bytes::from("home"));
}
#[tokio::test]
async fn unknown_path_is_404() {
let res = get(&app(), "/missing").await;
assert_eq!(res.status, StatusCode::NOT_FOUND);
}
#[tokio::test]
async fn panicking_handler_yields_500_not_crash() {
let res = get(&app(), "/boom").await;
assert_eq!(res.status, StatusCode::INTERNAL_SERVER_ERROR);
}
#[tokio::test]
async fn process_with_extensions_seeds_call() {
#[derive(Clone)]
struct Marker(u32);
let app = Churust::server()
.routing(|r| {
r.get("/", |c: Call| async move {
format!("{}", c.get::<Marker>().map(|m| m.0).unwrap_or(0))
});
})
.build();
let mut ext = http::Extensions::new();
ext.insert(Marker(7));
let res = app
.process_with_extensions(
Method::GET,
"/".parse().unwrap(),
HeaderMap::new(),
Bytes::new(),
ext,
)
.await;
assert_eq!(res.body, Bytes::from("7"));
}
#[tokio::test]
async fn state_extractor_end_to_end() {
use crate::extract::State;
#[derive(Clone)]
struct Counter(u32);
let app = Churust::server()
.state(Counter(5))
.routing(|r| {
r.get(
"/n",
|s: State<Counter>| async move { format!("n={}", s.0 .0) },
);
})
.build();
let res = get(&app, "/n").await;
assert_eq!(res.body, Bytes::from("n=5"));
}
#[tokio::test]
async fn middleware_installed_via_plugin_runs() {
struct MarkPlugin;
struct Mark;
#[async_trait]
impl Middleware for Mark {
async fn handle(&self, call: Call, next: Next) -> Response {
let mut res = next.run(call).await;
res.headers.insert(
http::header::HeaderName::from_static("x-plugin"),
HeaderValue::from_static("on"),
);
res
}
}
impl Plugin for MarkPlugin {
fn install(self: Box<Self>, app: &mut AppBuilder) {
app.add_middleware(Arc::new(Mark));
}
}
let app = Churust::server()
.install(MarkPlugin)
.routing(|r| {
r.get("/", |_c: Call| async { "ok" });
})
.build();
let res = get(&app, "/").await;
assert_eq!(res.headers.get("x-plugin").unwrap(), "on");
}
}