pub mod errors;
use futures::FutureExt;
use std::collections::HashMap;
use std::net::SocketAddr;
use std::sync::{
Arc, OnceLock,
atomic::{AtomicBool, AtomicUsize, Ordering},
};
use crate::doc::{DocumentableDTO, DocumentationRegistrant};
use crate::logging::CorrelationContext;
use crate::middleware::rate::{RateLimiter, create_from_env};
use crate::response::{ErrorResult, ResponseType, ServiceResult, TypedServiceResult};
use crate::validation::Validate;
use http::{HeaderMap, Method, Request, Response};
use http_body_util::BodyExt;
use hyper::body::Incoming;
use hyper::server::conn::http1;
use hyper::service::service_fn;
use hyper_util::rt::TokioIo;
use schemars::generate::SchemaSettings;
use serde_json::Value;
use tokio::net::TcpListener;
use tracing::{info, warn};
#[derive(Clone)]
struct Holder {
path: String,
limit: u32,
}
#[derive(Clone, Debug)]
pub struct DtoEntity {
pub(crate) schema: Value,
pub(crate) example: Value,
pub(crate) name: &'static str,
}
#[derive(Clone, Debug, PartialEq, Eq)]
pub enum FileParameterType {
Image,
Document,
Spreadsheet,
Video,
Audio,
Archive,
Any,
}
impl FileParameterType {
pub fn allowed_extensions_example(&self) -> &'static str {
match self {
Self::Image => ".png, .jpg, .jpeg, .gif, .webp",
Self::Document => ".pdf, .doc, .docx, .txt",
Self::Spreadsheet => ".xls, .xlsx, .csv",
Self::Video => ".mp4, .mov, .avi",
Self::Audio => ".mp3, .wav, .aac",
Self::Archive => ".zip, .tar, .gz",
Self::Any => "Any extension",
}
}
}
#[derive(Clone, Debug)]
pub struct FileParameter {
pub name: String,
pub description: String,
pub file_type: FileParameterType,
pub count: Option<u32>,
pub limit: Option<u32>,
}
#[derive(Debug, Clone)]
pub struct RouteDescription {
pub group: String,
pub name: String,
pub description: String,
pub request_body: Option<DtoEntity>,
pub response_examples: HashMap<u16, DtoEntity>,
pub headers: HashMap<String, String>,
pub path_parameters: HashMap<String, String>,
pub path_parameter_defaults: HashMap<String, String>,
pub query_parameters: HashMap<String, String>,
pub query_parameter_defaults: HashMap<String, String>,
pub file_parameters: HashMap<String, FileParameter>,
pub rate_limit: Option<u32>,
pub authentication_required: bool,
pub authentication_comment: Option<String>,
pub summary: String,
pub tags: Vec<String>,
pub is_deprecated: bool,
}
impl RouteDescription {
pub fn new(summary: impl Into<String>) -> Self {
let s = summary.into();
Self {
group: "Default".to_string(),
name: s.clone(),
description: String::new(),
request_body: None,
response_examples: Default::default(),
headers: Default::default(),
path_parameters: Default::default(),
path_parameter_defaults: Default::default(),
query_parameters: Default::default(),
query_parameter_defaults: Default::default(),
file_parameters: Default::default(),
rate_limit: None,
authentication_required: false,
authentication_comment: None,
summary: s.clone(),
tags: vec!["Default".to_string()],
is_deprecated: false,
}
}
pub fn group(mut self, group: impl Into<String>) -> Self {
let g = group.into();
self.group = g.clone();
self.tags = vec![g];
self
}
pub fn name(mut self, name: impl Into<String>) -> Self {
let n = name.into();
self.name = n.clone();
self.summary = n;
self
}
pub fn description(mut self, desc: impl Into<String>) -> Self {
self.description = desc.into();
self
}
pub fn rate_limit(mut self, limit: u32) -> Self {
self.rate_limit = Some(limit);
self
}
pub fn with_body<T>(mut self) -> Self
where
T: DocumentableDTO + Validate,
{
let settings = SchemaSettings::openapi3();
let generator = settings.into_generator();
let schema = generator.into_root_schema_for::<T>();
let example = T::make_example();
self.request_body = Some(DtoEntity {
schema: schema.to_value(),
example: example.unwrap(),
name: std::any::type_name::<T>(),
});
self
}
pub fn authentication(mut self, required: bool) -> Self {
self.authentication_required = required;
self
}
pub fn response<T>(mut self, code: u16) -> Self
where
T: DocumentableDTO,
{
let settings = SchemaSettings::openapi3();
let generator = settings.into_generator();
let schema = generator.into_root_schema_for::<T>();
let example = T::make_example();
let response = DtoEntity {
schema: schema.to_value(),
example: example.unwrap(),
name: std::any::type_name::<T>(),
};
self.response_examples.insert(code, response);
self
}
pub fn authentication_comment(mut self, comment: impl Into<String>) -> Self {
self.authentication_comment = Some(comment.into());
self
}
pub fn header(mut self, k: impl Into<String>, v: impl Into<String>) -> Self {
self.headers.insert(k.into(), v.into());
self
}
pub fn path_param(mut self, k: impl Into<String>, v: impl Into<String>) -> Self {
self.path_parameters.insert(k.into(), v.into());
self
}
pub fn path_param_with_default(
mut self,
k: impl Into<String>,
v: impl Into<String>,
default: impl Into<String>,
) -> Self {
let k = k.into();
self.path_parameters.insert(k.clone(), v.into());
self.path_parameter_defaults.insert(k, default.into());
self
}
pub fn query_param(mut self, k: impl Into<String>, v: impl Into<String>) -> Self {
self.query_parameters.insert(k.into(), v.into());
self
}
pub fn query_param_with_default(
mut self,
k: impl Into<String>,
v: impl Into<String>,
default: impl Into<String>,
) -> Self {
let k = k.into();
self.query_parameters.insert(k.clone(), v.into());
self.query_parameter_defaults.insert(k, default.into());
self
}
pub fn file_param(
mut self,
name: impl Into<String>,
description: impl Into<String>,
file_type: FileParameterType,
) -> Self {
let name = name.into();
self.file_parameters.insert(
name.clone(),
FileParameter {
name,
description: description.into(),
file_type,
count: None,
limit: None,
},
);
self
}
pub fn file_param_with_limit(
mut self,
name: impl Into<String>,
description: impl Into<String>,
file_type: FileParameterType,
limit: u32,
count: u32,
) -> Self {
let name = name.into();
assert!(limit >= 1, "File parameter limit must be greater than 0");
assert!(count >= 1, "File parameter count must be greater than 0");
self.file_parameters.insert(
name.clone(),
FileParameter {
name,
description: description.into(),
file_type,
count: Some(count),
limit: Some(limit),
},
);
self
}
pub fn deprecated(mut self, deprecated: bool) -> Self {
self.is_deprecated = deprecated;
self
}
pub fn tag(mut self, tag: impl Into<String>) -> Self {
let t = tag.into();
if !self.tags.contains(&t) {
self.tags.push(t);
}
self
}
pub fn effective_rate_limit(&self, global: Option<u32>) -> Option<u32> {
self.rate_limit.or(global)
}
}
impl Default for RouteDescription {
fn default() -> Self {
Self::new("No description")
}
}
static TOTAL_COUNT: AtomicUsize = AtomicUsize::new(0);
static LIMITER: OnceLock<Arc<dyn RateLimiter>> = OnceLock::new();
static MOUNTED_HANDLERS: AtomicBool = AtomicBool::new(false);
pub type Middleware = Arc<
dyn for<'a> Fn(
&'a mut CorrelationContext,
) -> futures::future::BoxFuture<'a, Result<(), ErrorResult>>
+ Send
+ Sync,
>;
pub type BoxHandler = Arc<
dyn Fn(
CorrelationContext,
HeaderMap,
Method,
String,
Vec<u8>,
) -> futures::future::BoxFuture<'static, Response<String>>
+ Send
+ Sync,
>;
pub(crate) struct RouteEntry {
method: Method,
path: String,
handler: BoxHandler,
middlewares: Vec<Middleware>,
}
#[derive(Default)]
struct TrieNode {
static_children: HashMap<String, TrieNode>,
param_child: Option<(String, Box<TrieNode>)>, wildcard_routes: HashMap<Method, RouteEntry>, exact_routes: HashMap<Method, RouteEntry>, }
pub struct Router {
root: TrieNode,
global_middlewares: Vec<Middleware>,
test_client_handler: Option<BoxHandler>,
}
impl Router {
pub fn new() -> Self {
Self {
root: TrieNode::default(),
global_middlewares: Vec::new(),
test_client_handler: None,
}
}
pub(crate) fn insert(&mut self, route: RouteEntry) {
let segments: Vec<String> = route
.path
.trim_matches('/')
.split('/')
.filter(|s| !s.is_empty())
.map(String::from)
.collect();
Self::insert_at(&mut self.root, &segments, route);
}
fn insert_at(node: &mut TrieNode, segments: &[String], route: RouteEntry) {
match segments.split_first() {
None => {
node.exact_routes.insert(route.method.clone(), route);
}
Some((seg, _rest)) if seg == "*" => {
node.wildcard_routes.insert(route.method.clone(), route);
}
Some((seg, rest)) if seg.starts_with(':') => {
let name = seg[1..].to_string();
let (existing_name, child) = node
.param_child
.get_or_insert_with(|| (name.clone(), Box::new(TrieNode::default())));
debug_assert_eq!(
existing_name, &name,
"conflicting param names at same position: {existing_name} vs {name}"
);
Self::insert_at(child, rest, route);
}
Some((seg, rest)) => {
let child = node.static_children.entry(seg.clone()).or_default();
Self::insert_at(child, rest, route);
}
}
}
pub(crate) fn resolve(
&self,
method: &Method,
path: &str,
) -> Option<(&RouteEntry, HashMap<String, String>)> {
let normal_path = normalize_path(path);
let segments: Vec<&str> = normal_path
.trim_matches('/')
.split('/')
.filter(|s| !s.is_empty())
.collect();
let mut params = Vec::new();
let route = Self::find(&self.root, &segments, method, &mut params)?;
Some((route, params.into_iter().collect()))
}
pub fn set_test_client_handler(&mut self, handler: BoxHandler) {
self.test_client_handler = Some(handler);
}
fn ensure_global_middleware(&mut self) {
if MOUNTED_HANDLERS
.compare_exchange(false, true, Ordering::SeqCst, Ordering::SeqCst)
.is_ok()
{
self.global_middlewares.push(Arc::new(|ctx| {
async move {
if ctx.request_id().is_empty() {
let rid = hex::encode(rand::random::<[u8; 8]>());
ctx.set_request_id(&rid);
}
Ok(())
}
.boxed()
}));
self.global_middlewares.push(Arc::new(|ctx| {
async move {
let headers = ctx.headers();
if let Some(v) = headers
.get("x-correlation-id")
.and_then(|h| h.to_str().ok())
{
if !v.is_empty() {
ctx.set_correlation_id(v);
}
}
if let Some(v) = headers
.get("x-correlation-flow")
.and_then(|h| h.to_str().ok())
{
if let Ok(flow) = v.parse::<crate::logging::CorrelationFlow>() {
ctx.set_flow(flow);
}
}
Ok(())
}
.boxed()
}));
}
}
pub fn mount<T, F, Fut>(
&mut self,
base_path: &str,
version: u32,
path: &str,
has_body: bool,
method: Method,
description: RouteDescription,
handler: F,
middlewares: Vec<Middleware>,
) where
T: serde::Serialize + Send + Sync + 'static,
F: Fn(CorrelationContext) -> Fut + Send + Sync + 'static,
Fut: Future<Output = Result<ServiceResult<T>, ErrorResult>> + Send + 'static,
{
self.ensure_global_middleware();
let full_path = calculate_full_path(base_path, version, path);
let meta_path = Holder {
path: full_path.to_string(),
limit: description.rate_limit.unwrap_or(0),
};
if let Ok(mut reg) = DocumentationRegistrant::global().write() {
reg.register_route(
&full_path,
method.as_str(),
"Controller",
description.clone(),
);
}
if has_body {
let expected = description.request_body.clone();
if expected.is_none() && description.file_parameters.is_empty() {
panic!(
"Request body class must be specified in the route description for {} {}",
method.as_str(),
full_path
);
}
}
let mut all_middlewares = Vec::new();
if let Some(limit) = description.rate_limit {
if limit > 0 {
let path_clone = meta_path.clone();
let m: Middleware = Arc::new(move |ctx: &mut CorrelationContext| {
let headers = ctx.headers();
let path_inner = path_clone.clone();
let fut = async move {
let limiter = LIMITER.get_or_init(|| create_from_env()).clone();
let key = ctx.user_id().unwrap_or_else(|| {
headers
.get("x-forwarded-for")
.and_then(|v| v.to_str().ok())
.unwrap_or("anonymous")
.to_string()
}) + ":"
+ &path_inner.path;
if limiter.is_allowed(&key, path_inner.limit).await {
return Ok(());
}
Err(ErrorResult::from_error(
"Too many requests. Please try again later.",
429,
))
};
FutureExt::boxed(fut)
});
all_middlewares.push(m);
}
}
all_middlewares.extend(middlewares);
let file_counts: Vec<(String, u32)> = description
.file_parameters
.iter()
.filter_map(|(name, p)| p.count.map(|c| (name.clone(), c)))
.collect();
let handler = Arc::new(handler);
let boxed: BoxHandler = Arc::new(move |ctx, _headers, _method, _path, _body| {
let file_counts = file_counts.clone();
let handler = handler.clone();
Box::pin(async move {
let ctx = Arc::new(ctx);
if let Some(err) = file_count_violation(ctx.clone(), &file_counts) {
return crate::response::error_response(&err, ctx.clone());
}
match handler((*ctx).clone()).await {
Ok(typed) => {
let body = match typed.serialize() {
Ok(b) => b,
Err(e) => {
let err =
ErrorResult::from_error(&anyhow::anyhow!(e.to_string()), 500);
return crate::response::error_response(&err, ctx.clone());
}
};
if body.is_empty() {
let err = ErrorResult::new("Response body is null", None, 503);
return crate::response::error_response(&err, ctx.clone());
}
let mut builder = Response::builder()
.status(typed.code())
.header("X-Request-ID", ctx.request_id())
.header("X-Correlation-ID", ctx.correlation_id())
.header("X-Correlation-Flow", ctx.flow().to_string());
let ct = match typed.response_type() {
ResponseType::Json => "application/json",
ResponseType::File => {
let filename = body.rsplit('/').next().unwrap_or("file");
builder = builder.header(
"Content-Disposition",
format!("attachment; filename=\"{}\"", filename),
);
"application/octet-stream"
}
ResponseType::Xml => "application/xml",
ResponseType::Javascript => "application/javascript",
ResponseType::Html => "text/html",
ResponseType::Text => "text/plain",
};
builder = builder.header("Content-Type", ct);
builder.body(body).unwrap()
}
Err(e) => crate::response::error_response(&e, ctx.clone()),
}
})
});
self.insert(RouteEntry {
method: method.clone(),
path: full_path.clone(),
handler: boxed,
middlewares: all_middlewares,
});
info!(
target: "routing",
handler = "Controller",
method = method.as_str(),
path = %full_path,
"Mounted a '{}' handler which listens on '{}'",
method.as_str(),
full_path
);
TOTAL_COUNT.fetch_add(1, Ordering::SeqCst);
}
pub fn mount_typed<F, Fut>(
&mut self,
base_path: &str,
version: u32,
path: &str,
method: Method,
description: RouteDescription,
handler: F,
middlewares: Vec<Middleware>,
) where
F: Fn(CorrelationContext) -> Fut + Send + Sync + 'static,
Fut: Future<Output = Result<Box<dyn TypedServiceResult>, ErrorResult>> + Send + 'static,
{
self.ensure_global_middleware();
let full_path = calculate_full_path(base_path, version, path);
let meta_path = Holder {
path: full_path.to_string(),
limit: description.rate_limit.unwrap_or(0),
};
if let Ok(mut reg) = DocumentationRegistrant::global().write() {
reg.register_route(
&full_path,
method.as_str(),
"Controller",
description.clone(),
);
}
let mut all_middlewares = Vec::new();
if let Some(limit) = description.rate_limit {
if limit > 0 {
let path_clone = meta_path.clone();
let m: Middleware = Arc::new(move |ctx: &mut CorrelationContext| {
let headers = ctx.headers();
let path_inner = path_clone.clone();
let fut = async move {
let limiter = LIMITER.get_or_init(|| create_from_env()).clone();
let key = ctx.user_id().unwrap_or_else(|| {
headers
.get("x-forwarded-for")
.and_then(|v| v.to_str().ok())
.unwrap_or("anonymous")
.to_string()
}) + ":"
+ &path_inner.path;
if limiter.is_allowed(&key, path_inner.limit).await {
return Ok(());
}
Err(ErrorResult::from_error(
"Too many requests. Please try again later.",
429,
))
};
FutureExt::boxed(fut)
});
all_middlewares.push(m);
}
}
all_middlewares.extend(middlewares);
let file_counts: Vec<(String, u32)> = description
.file_parameters
.iter()
.filter_map(|(name, p)| p.count.map(|c| (name.clone(), c)))
.collect();
let handler = Arc::new(handler);
let boxed: BoxHandler = Arc::new(move |ctx, _headers, _method, _path, _body| {
let file_counts = file_counts.clone();
let handler = handler.clone();
Box::pin(async move {
let ctx = Arc::new(ctx);
if let Some(err) = file_count_violation(ctx.clone(), &file_counts) {
return crate::response::error_response(&err, ctx.clone());
}
match handler((*ctx).clone()).await {
Ok(typed) => crate::response::build_response(typed.as_ref(), ctx.clone()),
Err(e) => crate::response::error_response(&e, ctx.clone()),
}
})
});
self.insert(RouteEntry {
method: method.clone(),
path: full_path.clone(),
handler: boxed,
middlewares: all_middlewares,
});
info!(
handler = "Controller",
method = method.as_str(),
path = %full_path,
"Mounted a '{}' handler which listens on '{}'",
method.as_str(),
full_path
);
TOTAL_COUNT.fetch_add(1, Ordering::SeqCst);
}
pub fn mount_raw(
&mut self,
base_path: &str,
version: u32,
path: &str,
method: Method,
handler: BoxHandler,
middlewares: Vec<Middleware>,
) {
self.ensure_global_middleware();
let full_path = calculate_full_path(base_path, version, path);
self.insert(RouteEntry {
method: method.clone(),
path: full_path.clone(),
handler,
middlewares,
});
info!(
handler = "raw",
method = method.as_str(),
path = %full_path,
"Mounted a '{}' handler which listens on '{}'",
method.as_str(),
full_path
);
TOTAL_COUNT.fetch_add(1, Ordering::SeqCst);
}
pub fn mount_static(
&mut self,
base_path: &str,
version: u32,
path: &str,
fs_path: String,
middlewares: Vec<Middleware>,
) {
let full_path = if version == 0 {
no_trailing_slash(&format!(
"/{}/{}/*",
base_path.trim_matches('/'),
path.trim_matches('/')
))
} else {
no_trailing_slash(&format!(
"/v{}/{}/{}/*",
version,
base_path.trim_matches('/'),
path.trim_matches('/')
))
};
let fs_path_clone = fs_path.clone();
let full_path_no_wildcard = full_path.trim_end_matches("/*").to_string();
let handler: BoxHandler = Arc::new(move |ctx, _headers, _method, req_path, _body| {
let fs_path = fs_path_clone.clone();
let req_path = req_path.clone();
let ctx = ctx.clone();
let prefix = full_path_no_wildcard.clone();
Box::pin(async move {
let rel = req_path.trim_start_matches(&prefix).trim_start_matches('/');
let candidates = if rel.is_empty() {
vec![
format!("{}/index.html", fs_path.trim_end_matches('/')),
fs_path.clone(),
]
} else {
vec![format!("{}/{}", fs_path.trim_end_matches('/'), rel)]
};
for candidate in &candidates {
if let Ok(bytes) = std::fs::read(candidate) {
let ct = guess_content_type(candidate);
let body = String::from_utf8_lossy(&bytes).into_owned();
return Response::builder()
.status(200)
.header("X-Request-ID", ctx.request_id())
.header("Content-Type", ct)
.body(body)
.unwrap();
}
}
let file_path = if rel.is_empty() {
fs_path.clone()
} else {
format!("{}/{}", fs_path.trim_end_matches('/'), rel)
};
let typed = ServiceResult::new(
"success",
"OK",
Some(format!("Serving static file: {}", file_path)),
200,
)
.with_response_type(ResponseType::Text);
let body = TypedServiceResult::serialize(&typed).unwrap();
Response::builder()
.status(TypedServiceResult::code(&typed))
.header("X-Request-ID", ctx.request_id())
.header("Content-Type", "text/plain")
.body(body)
.unwrap()
})
});
self.insert(RouteEntry {
method: Method::GET,
path: full_path.clone(),
handler,
middlewares,
});
info!(
handler = "static",
path = %full_path,
"Mounted a '{}' handler which listens on '{}'",
"static file",
full_path
);
TOTAL_COUNT.fetch_add(1, Ordering::SeqCst);
}
fn find<'a>(
node: &'a TrieNode,
segments: &[&str],
method: &Method,
params: &mut Vec<(String, String)>,
) -> Option<&'a RouteEntry> {
match segments.split_first() {
None => node
.exact_routes
.get(method)
.or_else(|| node.wildcard_routes.get(method)),
Some((seg, rest)) => {
if let Some(child) = node.static_children.get(*seg) {
if let Some(r) = Self::find(child, rest, method, params) {
return Some(r);
}
}
if let Some((name, child)) = &node.param_child {
params.push((name.clone(), (*seg).to_string()));
if let Some(r) = Self::find(child, rest, method, params) {
return Some(r);
}
params.pop();
}
node.wildcard_routes.get(method)
}
}
}
}
impl Default for Router {
fn default() -> Self {
Self::new()
}
}
fn guess_content_type(path: &str) -> &'static str {
let lower = path.to_ascii_lowercase();
if lower.ends_with(".html") || lower.ends_with(".htm") {
"text/html"
} else if lower.ends_with(".js") {
"application/javascript"
} else if lower.ends_with(".css") {
"text/css"
} else if lower.ends_with(".json") {
"application/json"
} else if lower.ends_with(".png") {
"image/png"
} else if lower.ends_with(".jpg") || lower.ends_with(".jpeg") {
"image/jpeg"
} else if lower.ends_with(".svg") {
"image/svg+xml"
} else if lower.ends_with(".yaml") || lower.ends_with(".yml") {
"application/yaml"
} else {
"text/plain"
}
}
fn normalize_path(p: &str) -> String {
let p = p.trim().replace("//", "/");
if p.ends_with('/') && p.len() > 1 {
p[..p.len() - 1].to_string()
} else if p.is_empty() {
"/".to_string()
} else {
p
}
}
fn no_trailing_slash(path: &str) -> String {
let p = path.trim().replace("//", "/");
if p.is_empty() || p == "/" {
return "/".to_string();
}
if p.ends_with('/') {
p[..p.rfind('/').unwrap()].to_string()
} else {
p
}
}
fn file_count_violation(
ctx: Arc<CorrelationContext>,
counts: &[(String, u32)],
) -> Option<ErrorResult> {
if counts.is_empty() {
return None;
}
let mp = ctx.multipart()?;
for (name, max) in counts {
let occurrences = mp.files.iter().filter(|f| &f.field_name == name).count()
+ mp.fields.get(name).map_or(0, |v| v.len());
if occurrences as u32 > *max {
return Some(ErrorResult::bad_request(format!(
"Too many '{}' parts: got {}, maximum is {}",
name, occurrences, max
)));
}
}
None
}
fn calculate_full_path(base_path: &str, version: u32, path: &str) -> String {
let decoded = if version == 0 {
let dddd = no_trailing_slash(&format!(
"/{}/{}",
base_path.trim_matches('/'),
path.trim_start_matches('/')
));
urlencoding::decode(dddd.leak())
} else {
let dddd = no_trailing_slash(&format!(
"/v{}/{}/{}",
version,
base_path.trim_matches('/'),
path.trim_start_matches('/')
));
urlencoding::decode(dddd.leak())
};
if decoded.is_err() {
return base_path.to_string();
}
decoded.unwrap().to_string()
}
#[async_trait::async_trait]
pub trait RouteController: Send + Sync {
fn base_path(&self) -> &str;
fn version(&self) -> u32 {
0
}
fn full_path(&self, path: &str) -> String {
calculate_full_path(self.base_path(), self.version(), path)
}
async fn register_routes(&self, router: &mut Router);
async fn register(&self, router: &mut Router) {
self.register_routes(router).await;
}
fn type_name(&self) -> &'static str {
std::any::type_name::<Self>()
}
}
pub trait RouteControllerExt: RouteController {
fn mount_get<T, F, Fut>(
&self,
router: &mut Router,
path: &str,
description: RouteDescription,
handler: F,
middlewares: Vec<Middleware>,
) where
T: serde::Serialize + Send + Sync + 'static,
F: Fn(CorrelationContext) -> Fut + Send + Sync + 'static,
Fut: Future<Output = Result<ServiceResult<T>, ErrorResult>> + Send + 'static,
{
router.mount(
self.base_path(),
self.version(),
path,
false,
Method::GET,
description,
handler,
middlewares,
);
}
fn mount_post<T, F, Fut>(
&self,
router: &mut Router,
path: &str,
description: RouteDescription,
handler: F,
middlewares: Vec<Middleware>,
) where
T: serde::Serialize + Send + Sync + 'static,
F: Fn(CorrelationContext) -> Fut + Send + Sync + 'static,
Fut: Future<Output = Result<ServiceResult<T>, ErrorResult>> + Send + 'static,
{
router.mount(
self.base_path(),
self.version(),
path,
true,
Method::POST,
description,
handler,
middlewares,
);
}
fn mount_put<T, F, Fut>(
&self,
router: &mut Router,
path: &str,
description: RouteDescription,
handler: F,
middlewares: Vec<Middleware>,
) where
T: serde::Serialize + Send + Sync + 'static,
F: Fn(CorrelationContext) -> Fut + Send + Sync + 'static,
Fut: Future<Output = Result<ServiceResult<T>, ErrorResult>> + Send + 'static,
{
router.mount(
self.base_path(),
self.version(),
path,
true,
Method::PUT,
description,
handler,
middlewares,
);
}
fn mount_patch<T, F, Fut>(
&self,
router: &mut Router,
path: &str,
has_body: bool,
description: RouteDescription,
handler: F,
middlewares: Vec<Middleware>,
) where
T: serde::Serialize + Send + Sync + 'static,
F: Fn(CorrelationContext) -> Fut + Send + Sync + 'static,
Fut: Future<Output = Result<ServiceResult<T>, ErrorResult>> + Send + 'static,
{
router.mount(
self.base_path(),
self.version(),
path,
has_body,
Method::PATCH,
description,
handler,
middlewares,
);
}
fn mount_patch_with_body<T, F, Fut>(
&self,
router: &mut Router,
path: &str,
description: RouteDescription,
handler: F,
middlewares: Vec<Middleware>,
) where
T: serde::Serialize + Send + Sync + 'static,
F: Fn(CorrelationContext) -> Fut + Send + Sync + 'static,
Fut: Future<Output = Result<ServiceResult<T>, ErrorResult>> + Send + 'static,
{
self.mount_patch(router, path, true, description, handler, middlewares);
}
fn mount_patch_without_body<T, F, Fut>(
&self,
router: &mut Router,
path: &str,
description: RouteDescription,
handler: F,
middlewares: Vec<Middleware>,
) where
T: serde::Serialize + Send + Sync + 'static,
F: Fn(CorrelationContext) -> Fut + Send + Sync + 'static,
Fut: Future<Output = Result<ServiceResult<T>, ErrorResult>> + Send + 'static,
{
self.mount_patch(router, path, false, description, handler, middlewares);
}
fn mount_delete<T, F, Fut>(
&self,
router: &mut Router,
path: &str,
description: RouteDescription,
handler: F,
middlewares: Vec<Middleware>,
) where
T: serde::Serialize + Send + Sync + 'static,
F: Fn(CorrelationContext) -> Fut + Send + Sync + 'static,
Fut: Future<Output = Result<ServiceResult<T>, ErrorResult>> + Send + 'static,
{
router.mount(
self.base_path(),
self.version(),
path,
false,
Method::DELETE,
description,
handler,
middlewares,
);
}
fn mount_options<T, F, Fut>(
&self,
router: &mut Router,
path: &str,
description: RouteDescription,
handler: F,
middlewares: Vec<Middleware>,
) where
T: serde::Serialize + Send + Sync + 'static,
F: Fn(CorrelationContext) -> Fut + Send + Sync + 'static,
Fut: Future<Output = Result<ServiceResult<T>, ErrorResult>> + Send + 'static,
{
router.mount(
self.base_path(),
self.version(),
path,
false,
Method::OPTIONS,
description,
handler,
middlewares,
);
}
fn mount_head<T, F, Fut>(
&self,
router: &mut Router,
path: &str,
description: RouteDescription,
handler: F,
middlewares: Vec<Middleware>,
) where
T: serde::Serialize + Send + Sync + 'static,
F: Fn(CorrelationContext) -> Fut + Send + Sync + 'static,
Fut: Future<Output = Result<ServiceResult<T>, ErrorResult>> + Send + 'static,
{
router.mount(
self.base_path(),
self.version(),
path,
false,
Method::HEAD,
description,
handler,
middlewares,
);
}
fn mount_trace<T, F, Fut>(
&self,
router: &mut Router,
path: &str,
description: RouteDescription,
handler: F,
middlewares: Vec<Middleware>,
) where
T: serde::Serialize + Send + Sync + 'static,
F: Fn(CorrelationContext) -> Fut + Send + Sync + 'static,
Fut: Future<Output = Result<ServiceResult<T>, ErrorResult>> + Send + 'static,
{
router.mount(
self.base_path(),
self.version(),
path,
false,
Method::TRACE,
description,
handler,
middlewares,
);
}
fn mount_connect<T, F, Fut>(
&self,
router: &mut Router,
path: &str,
description: RouteDescription,
handler: F,
middlewares: Vec<Middleware>,
) where
T: serde::Serialize + Send + Sync + 'static,
F: Fn(CorrelationContext) -> Fut + Send + Sync + 'static,
Fut: Future<Output = Result<ServiceResult<T>, ErrorResult>> + Send + 'static,
{
router.mount(
self.base_path(),
self.version(),
path,
false,
Method::CONNECT,
description,
handler,
middlewares,
);
}
fn mount_copy<T, F, Fut>(
&self,
router: &mut Router,
path: &str,
description: RouteDescription,
handler: F,
middlewares: Vec<Middleware>,
) where
T: serde::Serialize + Send + Sync + 'static,
F: Fn(CorrelationContext) -> Fut + Send + Sync + 'static,
Fut: Future<Output = Result<ServiceResult<T>, ErrorResult>> + Send + 'static,
{
router.mount(
self.base_path(),
self.version(),
path,
false,
Method::from_bytes(b"COPY").unwrap(),
description,
handler,
middlewares,
);
}
fn mount_move<T, F, Fut>(
&self,
router: &mut Router,
path: &str,
description: RouteDescription,
handler: F,
middlewares: Vec<Middleware>,
) where
T: serde::Serialize + Send + Sync + 'static,
F: Fn(CorrelationContext) -> Fut + Send + Sync + 'static,
Fut: Future<Output = Result<ServiceResult<T>, ErrorResult>> + Send + 'static,
{
router.mount(
self.base_path(),
self.version(),
path,
false,
Method::from_bytes(b"MOVE").unwrap(),
description,
handler,
middlewares,
);
}
fn mount_lock<T, F, Fut>(
&self,
router: &mut Router,
path: &str,
description: RouteDescription,
handler: F,
middlewares: Vec<Middleware>,
) where
T: serde::Serialize + Send + Sync + 'static,
F: Fn(CorrelationContext) -> Fut + Send + Sync + 'static,
Fut: Future<Output = Result<ServiceResult<T>, ErrorResult>> + Send + 'static,
{
router.mount(
self.base_path(),
self.version(),
path,
true,
Method::from_bytes(b"LOCK").unwrap(),
description,
handler,
middlewares,
);
}
fn mount_unlock<T, F, Fut>(
&self,
router: &mut Router,
path: &str,
description: RouteDescription,
handler: F,
middlewares: Vec<Middleware>,
) where
T: serde::Serialize + Send + Sync + 'static,
F: Fn(CorrelationContext) -> Fut + Send + Sync + 'static,
Fut: Future<Output = Result<ServiceResult<T>, ErrorResult>> + Send + 'static,
{
router.mount(
self.base_path(),
self.version(),
path,
false,
Method::from_bytes(b"UNLOCK").unwrap(),
description,
handler,
middlewares,
);
}
fn mount_propfind<T, F, Fut>(
&self,
router: &mut Router,
path: &str,
description: RouteDescription,
handler: F,
middlewares: Vec<Middleware>,
) where
T: serde::Serialize + Send + Sync + 'static,
F: Fn(CorrelationContext) -> Fut + Send + Sync + 'static,
Fut: Future<Output = Result<ServiceResult<T>, ErrorResult>> + Send + 'static,
{
router.mount(
self.base_path(),
self.version(),
path,
true,
Method::from_bytes(b"PROPFIND").unwrap(),
description,
handler,
middlewares,
);
}
fn mount_mkcol<T, F, Fut>(
&self,
router: &mut Router,
path: &str,
description: RouteDescription,
handler: F,
middlewares: Vec<Middleware>,
) where
T: serde::Serialize + Send + Sync + 'static,
F: Fn(CorrelationContext) -> Fut + Send + Sync + 'static,
Fut: Future<Output = Result<ServiceResult<T>, ErrorResult>> + Send + 'static,
{
router.mount(
self.base_path(),
self.version(),
path,
false,
Method::from_bytes(b"MKCOL").unwrap(),
description,
handler,
middlewares,
);
}
fn mount_search<T, F, Fut>(
&self,
router: &mut Router,
path: &str,
description: RouteDescription,
handler: F,
middlewares: Vec<Middleware>,
) where
T: serde::Serialize + Send + Sync + 'static,
F: Fn(CorrelationContext) -> Fut + Send + Sync + 'static,
Fut: Future<Output = Result<ServiceResult<T>, ErrorResult>> + Send + 'static,
{
router.mount(
self.base_path(),
self.version(),
path,
true,
Method::from_bytes(b"SEARCH").unwrap(),
description,
handler,
middlewares,
);
}
fn mount_report<T, F, Fut>(
&self,
router: &mut Router,
path: &str,
description: RouteDescription,
handler: F,
middlewares: Vec<Middleware>,
) where
T: serde::Serialize + Send + Sync + 'static,
F: Fn(CorrelationContext) -> Fut + Send + Sync + 'static,
Fut: Future<Output = Result<ServiceResult<T>, ErrorResult>> + Send + 'static,
{
router.mount(
self.base_path(),
self.version(),
path,
true,
Method::from_bytes(b"REPORT").unwrap(),
description,
handler,
middlewares,
);
}
fn mount_checkin<T, F, Fut>(
&self,
router: &mut Router,
path: &str,
description: RouteDescription,
handler: F,
middlewares: Vec<Middleware>,
) where
T: serde::Serialize + Send + Sync + 'static,
F: Fn(CorrelationContext) -> Fut + Send + Sync + 'static,
Fut: Future<Output = Result<ServiceResult<T>, ErrorResult>> + Send + 'static,
{
router.mount(
self.base_path(),
self.version(),
path,
false,
Method::from_bytes(b"CHECKIN").unwrap(),
description,
handler,
middlewares,
);
}
fn mount_checkout<T, F, Fut>(
&self,
router: &mut Router,
path: &str,
description: RouteDescription,
handler: F,
middlewares: Vec<Middleware>,
) where
T: serde::Serialize + Send + Sync + 'static,
F: Fn(CorrelationContext) -> Fut + Send + Sync + 'static,
Fut: Future<Output = Result<ServiceResult<T>, ErrorResult>> + Send + 'static,
{
router.mount(
self.base_path(),
self.version(),
path,
false,
Method::from_bytes(b"CHECKOUT").unwrap(),
description,
handler,
middlewares,
);
}
fn mount_uncheckout<T, F, Fut>(
&self,
router: &mut Router,
path: &str,
description: RouteDescription,
handler: F,
middlewares: Vec<Middleware>,
) where
T: serde::Serialize + Send + Sync + 'static,
F: Fn(CorrelationContext) -> Fut + Send + Sync + 'static,
Fut: Future<Output = Result<ServiceResult<T>, ErrorResult>> + Send + 'static,
{
router.mount(
self.base_path(),
self.version(),
path,
false,
Method::from_bytes(b"UNCHECKOUT").unwrap(),
description,
handler,
middlewares,
);
}
fn mount_merge<T, F, Fut>(
&self,
router: &mut Router,
path: &str,
description: RouteDescription,
handler: F,
middlewares: Vec<Middleware>,
) where
T: serde::Serialize + Send + Sync + 'static,
F: Fn(CorrelationContext) -> Fut + Send + Sync + 'static,
Fut: Future<Output = Result<ServiceResult<T>, ErrorResult>> + Send + 'static,
{
router.mount(
self.base_path(),
self.version(),
path,
true,
Method::from_bytes(b"MERGE").unwrap(),
description,
handler,
middlewares,
);
}
fn mount_acl<T, F, Fut>(
&self,
router: &mut Router,
path: &str,
description: RouteDescription,
handler: F,
middlewares: Vec<Middleware>,
) where
T: serde::Serialize + Send + Sync + 'static,
F: Fn(CorrelationContext) -> Fut + Send + Sync + 'static,
Fut: Future<Output = Result<ServiceResult<T>, ErrorResult>> + Send + 'static,
{
router.mount(
self.base_path(),
self.version(),
path,
true,
Method::from_bytes(b"ACL").unwrap(),
description,
handler,
middlewares,
);
}
fn mount_custom<T, F, Fut>(
&self,
router: &mut Router,
path: &str,
method: Method,
description: RouteDescription,
handler: F,
middlewares: Vec<Middleware>,
) where
T: serde::Serialize + Send + Sync + 'static,
F: Fn(CorrelationContext) -> Fut + Send + Sync + 'static,
Fut: Future<Output = Result<ServiceResult<T>, ErrorResult>> + Send + 'static,
{
router.mount(
self.base_path(),
self.version(),
path,
false,
method,
description,
handler,
middlewares,
);
}
fn mount_static(
&self,
router: &mut Router,
path: &str,
fs_path: String,
middlewares: Vec<Middleware>,
) {
router.mount_static(self.base_path(), self.version(), path, fs_path, middlewares);
}
fn mount_typed<F, Fut>(
&self,
router: &mut Router,
path: &str,
method: Method,
description: RouteDescription,
handler: F,
middlewares: Vec<Middleware>,
) where
F: Fn(CorrelationContext) -> Fut + Send + Sync + 'static,
Fut: Future<Output = Result<Box<dyn TypedServiceResult>, ErrorResult>> + Send + 'static,
{
router.mount_typed(
self.base_path(),
self.version(),
path,
method,
description,
handler,
middlewares,
);
}
fn mount_get_typed<F, Fut>(
&self,
router: &mut Router,
path: &str,
description: RouteDescription,
handler: F,
middlewares: Vec<Middleware>,
) where
F: Fn(CorrelationContext) -> Fut + Send + Sync + 'static,
Fut: Future<Output = Result<Box<dyn TypedServiceResult>, ErrorResult>> + Send + 'static,
{
self.mount_typed(router, path, Method::GET, description, handler, middlewares);
}
fn mount_post_typed<F, Fut>(
&self,
router: &mut Router,
path: &str,
description: RouteDescription,
handler: F,
middlewares: Vec<Middleware>,
) where
F: Fn(CorrelationContext) -> Fut + Send + Sync + 'static,
Fut: Future<Output = Result<Box<dyn TypedServiceResult>, ErrorResult>> + Send + 'static,
{
self.mount_typed(
router,
path,
Method::POST,
description,
handler,
middlewares,
);
}
}
impl<T: RouteController> RouteControllerExt for T {}
pub struct ConfigurationRegistrant {
router: Arc<tokio::sync::RwLock<Router>>,
addr: SocketAddr,
}
impl ConfigurationRegistrant {
pub fn new(addr: SocketAddr) -> Self {
Self {
router: Arc::new(tokio::sync::RwLock::new(Router::new())),
addr,
}
}
pub async fn mount_controller<C: RouteController + 'static>(&self, controller: C) {
let mut r = self.router.write().await;
info!(
target: "routing",
handler = std::any::type_name::<C>(),
path = %controller.base_path(),
"Mounted controller '{}' at '{}'",
std::any::type_name::<C>(),
controller.base_path()
);
controller.register_routes(&mut r).await;
}
pub async fn mount_middleware(&self, _mw: Middleware) {
let mut r = self.router.write().await;
r.global_middlewares.push(_mw);
}
pub async fn serve(self: Arc<Self>) -> anyhow::Result<SocketAddr> {
let listener = TcpListener::bind(self.addr).await?;
let addr = listener.local_addr()?;
info!(address = %addr, "HTTP server listening");
let router = self.router.clone();
tokio::spawn(async move {
loop {
let (stream, remote) = match listener.accept().await {
Ok(v) => v,
Err(e) => {
warn!(error = %e, "Failed to accept HTTP connection");
continue;
}
};
let io = TokioIo::new(stream);
let router = router.clone();
tokio::spawn(async move {
let svc = service_fn(move |req: Request<Incoming>| {
let router = router.clone();
let remote = remote;
async move { handle_request(router, req, remote).await }
});
if let Err(e) = http1::Builder::new().serve_connection(io, svc).await {
tracing::debug!(remote_addr = %remote, error = %e, "HTTP connection ended with an error");
}
});
}
});
Ok(addr)
}
pub fn router_handle(&self) -> Arc<tokio::sync::RwLock<Router>> {
self.router.clone()
}
}
async fn handle_request(
router: Arc<tokio::sync::RwLock<Router>>,
req: Request<Incoming>,
remote_addr: SocketAddr,
) -> Result<Response<String>, ErrorResult> {
let request_started = std::time::Instant::now();
let method = req.method().clone();
let path = req.uri().path();
let old_path = path;
let path = urlencoding::decode(path);
if path.is_err() {
tracing::error!(
"This should not be possible. Encountered an error while processing the URL {}",
old_path
);
return Err(ErrorResult::bad_request("Invalid path parameter"));
}
let path = path.ok().unwrap().to_string();
let query = req.uri().query().unwrap_or("").to_string();
let headers = req.headers().clone();
let (_parts, body) = req.into_parts();
let body_bytes = match body.collect().await {
Ok(collected) => collected.to_bytes().to_vec(),
Err(e) => {
warn!(remote_addr = %remote_addr, error = %e, "Failed to read request body");
vec![]
}
};
let params: HashMap<String, String> = serde_urlencoded::from_str(&query).unwrap_or_default();
let mut ctx = build_correlation_context(&headers, ¶ms, &body_bytes);
if let Some(ct) = headers
.get(http::header::CONTENT_TYPE)
.and_then(|v| v.to_str().ok())
{
if ct.to_lowercase().starts_with("multipart/form-data") {
if let Some(boundary) =
crate::utils::request_parser::RequestParser::multipart_boundary(ct)
{
let mp = crate::utils::request_parser::RequestParser::parse_multipart(
&body_bytes,
&boundary,
);
ctx.set_multipart(mp);
}
}
}
{
let guard = router.read().await;
for mw in &guard.global_middlewares {
if let Err(e) = mw(&mut ctx).await {
let resp = crate::response::error_response(&e, Arc::new(ctx.clone()));
return Ok(finish_request(
&method,
&path,
request_started,
adapt_response(resp),
));
}
}
}
let limit = ctx
.query_param("limit")
.and_then(|v| v.parse().ok())
.unwrap_or(15usize);
let cursor = ctx.query_param("cursor");
ctx.set_pagination(cursor, limit);
let test_hook = {
let guard = router.read().await;
let has_header = headers.contains_key("x-moovable-test-client")
|| headers.contains_key("x-tm30-test-client");
if has_header {
guard.test_client_handler.clone()
} else {
None
}
};
if let Some(hook) = test_hook {
let resp = hook(
ctx.clone(),
headers.clone(),
method.clone(),
path.clone(),
body_bytes,
)
.await;
return Ok(finish_request(
&method,
&path,
request_started,
adapt_string_response(resp, Arc::new(ctx.clone())),
));
}
let guard = router.read().await;
if let Some((entry, params)) = guard.resolve(&method, &path) {
for mw in &entry.middlewares {
if let Err(e) = mw(&mut ctx).await {
let resp = crate::response::error_response(&e, Arc::new(ctx.clone()));
return Ok(finish_request(
&method,
&path,
request_started,
adapt_response(resp),
));
}
}
ctx.set_params(params);
let handler = entry.handler.clone();
let path_clone = path.clone();
let headers_clone = headers.clone();
let method_clone = method.clone();
drop(guard);
let resp = handler(
ctx.clone(),
headers_clone,
method_clone,
path_clone,
body_bytes,
)
.await;
Ok(finish_request(
&method,
&path,
request_started,
adapt_string_response(resp, Arc::new(ctx.clone())),
))
} else {
drop(guard);
let err = ErrorResult::not_found(format!("The requested resource was not found: {path}"));
let resp = crate::response::error_response(&err, Arc::new(ctx.clone()));
Ok(finish_request(
&method,
&path,
request_started,
adapt_response(resp),
))
}
}
fn finish_request(
method: &Method,
path: &str,
started: std::time::Instant,
response: Response<String>,
) -> Response<String> {
crate::middleware::monitoring::RequestLogger::log(
path,
method.as_str(),
response.status().as_u16(),
started.elapsed(),
);
response
}
fn build_correlation_context(
headers: &HeaderMap,
query: &HashMap<String, String>,
body: &[u8],
) -> CorrelationContext {
let corr_id = headers
.get("x-correlation-id")
.and_then(|v| v.to_str().ok())
.unwrap_or("");
let flow_str = headers
.get("x-correlation-flow")
.and_then(|v| v.to_str().ok())
.unwrap_or("ONCE");
let flow = flow_str
.parse()
.unwrap_or(crate::logging::CorrelationFlow::Once);
let ctx = if corr_id.is_empty() {
CorrelationContext::new()
} else {
CorrelationContext::with_ids(corr_id, &hex::encode(rand::random::<[u8; 8]>()))
};
ctx.set_flow(flow);
ctx.set_headers(headers.clone());
ctx.set_query_params(query.clone());
ctx.set_body(body.to_vec());
let req_id = headers.get("x-request-id").and_then(|v| v.to_str().ok());
if let Some(rid) = req_id {
ctx.set_request_id(rid);
} else {
let rid = hex::encode(rand::random::<[u8; 8]>());
ctx.set_request_id(&rid);
}
ctx
}
fn adapt_response(r: Response<String>) -> Response<String> {
r
}
fn adapt_string_response(
mut r: Response<String>,
ctx: Arc<CorrelationContext>,
) -> Response<String> {
let headers = r.headers_mut();
headers
.entry("x-request-id")
.or_insert(ctx.request_id().parse().unwrap());
headers
.entry("x-correlation-id")
.or_insert(ctx.correlation_id().parse().unwrap());
headers
.entry("x-correlation-flow")
.or_insert(ctx.flow().to_string().parse().unwrap());
r
}
pub fn json_response<T: serde::Serialize + Send + Sync>(
result: ServiceResult<T>,
ctx: Arc<CorrelationContext>,
) -> Response<String> {
crate::response::build_response(&result, ctx)
}
pub fn typed_response(
result: &dyn TypedServiceResult,
ctx: Arc<CorrelationContext>,
) -> Response<String> {
let body = result.serialize().unwrap_or_else(|e| {
ErrorResult::from_error(&anyhow::anyhow!(e.to_string()), 500)
.message
.clone()
});
let mut builder = Response::builder()
.status(result.code())
.header("X-Request-ID", ctx.request_id())
.header("X-Correlation-ID", ctx.correlation_id())
.header("X-Correlation-Flow", ctx.flow().to_string());
let ct = match result.response_type() {
ResponseType::Json => "application/json",
ResponseType::File => {
let filename = body.rsplit('/').next().unwrap_or("file");
builder = builder.header(
"Content-Disposition",
format!("attachment; filename=\"{}\"", filename),
);
"application/octet-stream"
}
ResponseType::Xml => "application/xml",
ResponseType::Javascript => "application/javascript",
ResponseType::Html => "text/html",
ResponseType::Text => "text/plain",
};
builder = builder.header("Content-Type", ct);
builder.body(body).unwrap()
}