#![forbid(unsafe_code)]
use blazingly_core::{
AppDefinition, BackgroundTask, BackgroundTaskError, BodyStreamError, HttpMethod, HttpUpgrade,
InputSource, OperationDescriptor, ResponseHeader, SecuritySchemeDescriptor, StreamingBody,
};
use blazingly_executor::{
DependencyError, ExecutableApp, ExecutionOutcome, FromInvocation,
HttpRequestParts as InvocationRequestParts, InputRejection, InvocationControl, InvocationInput,
};
use blazingly_json::{Value, json};
use blazingly_openapi::{OpenApiAssetResponse, OpenApiConfig, OpenApiService};
use serde::Serialize;
use serde::de::DeserializeOwned;
use std::any::{Any, TypeId};
use std::borrow::Cow;
use std::cell::{OnceCell, RefCell};
use std::collections::{BTreeMap, HashMap};
use std::fmt;
use std::future::Future;
use std::net::{IpAddr, SocketAddr};
use std::rc::Rc;
use std::str::Utf8Error;
pub const DEFAULT_MAX_BODY_BYTES: usize = 1024 * 1024;
pub const EMIT_VARIABLE: &str = "BLAZINGLY_EMIT";
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct Request {
method: HttpMethod,
target: String,
headers: BTreeMap<String, String>,
body: Vec<u8>,
peer_addr: Option<SocketAddr>,
scheme: String,
}
pub trait HttpRequestView {
fn method(&self) -> HttpMethod;
fn target(&self) -> &str;
fn header_value(&self, name: &str, index: usize) -> Option<&str>;
fn body(&self) -> &[u8];
fn take_body_stream(&self) -> Option<StreamingBody> {
None
}
fn peer_addr(&self) -> Option<SocketAddr> {
None
}
#[allow(clippy::unnecessary_literal_bound)]
fn scheme(&self) -> &str {
"http"
}
}
pub struct HttpRequestContext<'request> {
request: &'request dyn HttpRequestView,
client_ip: Option<IpAddr>,
scheme: Cow<'request, str>,
host: Option<Cow<'request, str>>,
extensions: Vec<(TypeId, Box<dyn Any>)>,
}
impl<'request> HttpRequestContext<'request> {
fn new(request: &'request dyn HttpRequestView) -> Self {
Self {
request,
client_ip: request.peer_addr().map(|address| address.ip()),
scheme: Cow::Borrowed(request.scheme()),
host: None,
extensions: Vec::new(),
}
}
#[must_use]
pub fn request(&self) -> &dyn HttpRequestView {
self.request
}
#[must_use]
pub fn client_ip(&self) -> Option<IpAddr> {
self.client_ip
}
pub fn set_client_ip(&mut self, client_ip: IpAddr) {
self.client_ip = Some(client_ip);
}
#[must_use]
pub fn scheme(&self) -> &str {
&self.scheme
}
pub fn set_scheme(&mut self, scheme: impl Into<String>) {
self.scheme = Cow::Owned(scheme.into());
}
#[must_use]
pub fn host(&self) -> Option<&str> {
self.host
.as_deref()
.or_else(|| self.request.header_value("host", 0))
}
pub fn set_host(&mut self, host: impl Into<String>) {
self.host = Some(Cow::Owned(host.into()));
}
pub fn insert_extension<T: 'static>(&mut self, value: T) {
let type_id = TypeId::of::<T>();
if let Some((_, existing)) = self
.extensions
.iter_mut()
.find(|(existing, _)| *existing == type_id)
{
*existing = Box::new(value);
} else {
self.extensions.push((type_id, Box::new(value)));
}
}
#[must_use]
pub fn extension<T: 'static>(&self) -> Option<&T> {
self.extension_by_id(TypeId::of::<T>())?.downcast_ref()
}
#[must_use]
pub fn connection_info(&self) -> ConnectionInfo {
ConnectionInfo {
client_ip: self.client_ip,
scheme: self.scheme.as_ref().to_owned(),
host: self.host().map(str::to_owned),
}
}
fn extension_by_id(&self, type_id: TypeId) -> Option<&dyn Any> {
self.extensions
.iter()
.find(|(existing, _)| *existing == type_id)
.map(|(_, value)| value.as_ref())
}
}
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct ConnectionInfo {
client_ip: Option<IpAddr>,
scheme: String,
host: Option<String>,
}
impl ConnectionInfo {
#[must_use]
pub const fn client_ip(&self) -> Option<IpAddr> {
self.client_ip
}
#[must_use]
pub fn scheme(&self) -> &str {
&self.scheme
}
#[must_use]
pub fn host(&self) -> Option<&str> {
self.host.as_deref()
}
fn from_request(request: &(impl HttpRequestView + ?Sized)) -> Self {
Self {
client_ip: request.peer_addr().map(|address| address.ip()),
scheme: request.scheme().to_owned(),
host: request.header_value("host", 0).map(str::to_owned),
}
}
}
pub trait HttpMiddleware {
fn on_request(&self, _context: &mut HttpRequestContext<'_>) -> Option<Response> {
None
}
fn on_operation(
&self,
_context: &mut HttpRequestContext<'_>,
_operation: &OperationDescriptor,
_security_schemes: &[SecuritySchemeDescriptor],
) -> Option<Response> {
None
}
fn on_response(
&self,
_context: &HttpRequestContext<'_>,
_operation: Option<&OperationDescriptor>,
_response: &mut Response,
) {
}
fn verifies_security(&self) -> bool {
true
}
}
type OperationFilter = Rc<dyn Fn(&str) -> bool>;
#[derive(Clone)]
enum OperationPredicate {
Exact(Box<str>),
Prefix(Box<str>),
Filter(OperationFilter),
}
impl OperationPredicate {
fn matches(&self, operation_id: &str) -> bool {
match self {
Self::Exact(expected) => operation_id == expected.as_ref(),
Self::Prefix(prefix) => operation_id.starts_with(prefix.as_ref()),
Self::Filter(filter) => filter(operation_id),
}
}
}
#[derive(Clone, Default)]
pub struct MiddlewareScope {
prefixes: Vec<Box<str>>,
operations: Vec<OperationPredicate>,
}
impl MiddlewareScope {
#[must_use]
pub const fn all() -> Self {
Self {
prefixes: Vec::new(),
operations: Vec::new(),
}
}
#[must_use]
pub fn prefix(prefix: &str) -> Self {
Self::all().with_prefix(prefix)
}
#[must_use]
pub fn operation(operation_id: &str) -> Self {
Self::all().with_operation(operation_id)
}
#[must_use]
pub fn with_prefix(mut self, prefix: &str) -> Self {
self.prefixes.push(normalize_prefix(prefix));
self
}
#[must_use]
pub fn with_operation(mut self, operation_id: &str) -> Self {
self.operations
.push(OperationPredicate::Exact(Box::from(operation_id)));
self
}
#[must_use]
pub fn with_operation_prefix(mut self, prefix: &str) -> Self {
self.operations
.push(OperationPredicate::Prefix(Box::from(prefix)));
self
}
#[must_use]
pub fn with_operation_filter<Filter>(mut self, filter: Filter) -> Self
where
Filter: Fn(&str) -> bool + 'static,
{
self.operations
.push(OperationPredicate::Filter(Rc::new(filter)));
self
}
#[must_use]
pub fn is_global(&self) -> bool {
self.prefixes.is_empty() && self.operations.is_empty()
}
#[must_use]
pub fn matches_request(&self, path: &str) -> bool {
self.operations.is_empty() && self.matches_path(path)
}
#[must_use]
pub fn matches_operation(&self, path: &str, operation_id: &str) -> bool {
self.matches_path(path)
&& (self.operations.is_empty()
|| self
.operations
.iter()
.any(|predicate| predicate.matches(operation_id)))
}
fn matches_path(&self, path: &str) -> bool {
self.prefixes.is_empty()
|| self
.prefixes
.iter()
.any(|prefix| path_has_prefix(path, prefix))
}
}
impl fmt::Debug for MiddlewareScope {
fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
formatter
.debug_struct("MiddlewareScope")
.field("prefixes", &self.prefixes)
.field("operations", &self.operations.len())
.finish()
}
}
fn normalize_prefix(prefix: &str) -> Box<str> {
let trimmed = prefix.trim_end_matches('/');
if trimmed.is_empty() {
Box::from("")
} else if trimmed.starts_with('/') {
Box::from(trimmed)
} else {
Box::from(format!("/{trimmed}"))
}
}
fn path_has_prefix(path: &str, prefix: &str) -> bool {
let Some(rest) = path.strip_prefix(prefix) else {
return false;
};
rest.is_empty() || rest.starts_with('/')
}
struct ScopedMiddleware {
scope: MiddlewareScope,
layer: Rc<dyn HttpMiddleware>,
}
impl ScopedMiddleware {
fn matches(&self, path: &str, operation: Option<&OperationDescriptor>) -> bool {
operation.map_or_else(
|| self.scope.matches_request(path),
|operation| {
self.scope
.matches_operation(path, operation.contract.id.as_str())
},
)
}
}
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum HttpErrorSource {
Routing,
Request,
Rejection,
Domain,
Internal,
}
pub struct HttpError<'dispatch> {
source: HttpErrorSource,
status: u16,
code: &'dispatch str,
message: &'dispatch str,
method: HttpMethod,
path: &'dispatch str,
operation: Option<&'dispatch OperationDescriptor>,
}
impl HttpError<'_> {
#[must_use]
pub const fn source(&self) -> HttpErrorSource {
self.source
}
#[must_use]
pub const fn status(&self) -> u16 {
self.status
}
#[must_use]
pub const fn code(&self) -> &str {
self.code
}
#[must_use]
pub const fn message(&self) -> &str {
self.message
}
#[must_use]
pub const fn method(&self) -> HttpMethod {
self.method
}
#[must_use]
pub const fn path(&self) -> &str {
self.path
}
#[must_use]
pub const fn operation(&self) -> Option<&OperationDescriptor> {
self.operation
}
}
pub trait HttpErrorHandler {
fn on_error(&self, error: &HttpError<'_>, response: &mut Response);
}
#[derive(Clone, Debug, Default)]
pub struct BackgroundTasks {
tasks: Rc<RefCell<Vec<BackgroundTask>>>,
}
impl BackgroundTasks {
#[must_use]
pub fn new() -> Self {
Self::default()
}
pub fn add_task(&self, task: BackgroundTask) {
self.tasks.borrow_mut().push(task);
}
pub fn add<Task, TaskFuture>(&self, task: Task)
where
Task: FnOnce() -> TaskFuture + 'static,
TaskFuture: Future<Output = Result<(), BackgroundTaskError>> + 'static,
{
self.add_task(BackgroundTask::new(task));
}
pub fn add_infallible<Task, TaskFuture>(&self, task: Task)
where
Task: FnOnce() -> TaskFuture + 'static,
TaskFuture: Future<Output = ()> + 'static,
{
self.add_task(BackgroundTask::infallible(task));
}
#[must_use]
pub fn len(&self) -> usize {
self.tasks.borrow().len()
}
#[must_use]
pub fn is_empty(&self) -> bool {
self.tasks.borrow().is_empty()
}
#[must_use]
pub fn take(&self) -> Vec<BackgroundTask> {
std::mem::take(&mut self.tasks.borrow_mut())
}
}
impl FromInvocation for BackgroundTasks {
fn from_invocation(
input: &InvocationInput<'_>,
name: &str,
_required: bool,
) -> Result<Self, InputRejection> {
let InvocationInput::Http(request) = input else {
return Err(InputRejection::new(
500,
"background_tasks_transport_mismatch",
"after-response tasks are available only through HTTP",
));
};
request
.extension(TypeId::of::<Self>())
.and_then(<dyn Any>::downcast_ref::<Self>)
.cloned()
.ok_or_else(|| {
InputRejection::new(
500,
"background_tasks_unavailable",
format!("this transport installed no after-response tasks for `{name}`"),
)
})
}
}
impl Request {
#[must_use]
pub fn new(method: HttpMethod, target: impl Into<String>) -> Self {
Self {
method,
target: target.into(),
headers: BTreeMap::new(),
body: Vec::new(),
peer_addr: None,
scheme: "http".to_owned(),
}
}
#[must_use]
pub fn get(target: impl Into<String>) -> Self {
Self::new(HttpMethod::Get, target)
}
#[must_use]
pub fn head(target: impl Into<String>) -> Self {
Self::new(HttpMethod::Head, target)
}
#[must_use]
pub fn post(target: impl Into<String>) -> Self {
Self::new(HttpMethod::Post, target)
}
#[must_use]
pub fn put(target: impl Into<String>) -> Self {
Self::new(HttpMethod::Put, target)
}
#[must_use]
pub fn patch(target: impl Into<String>) -> Self {
Self::new(HttpMethod::Patch, target)
}
#[must_use]
pub fn delete(target: impl Into<String>) -> Self {
Self::new(HttpMethod::Delete, target)
}
#[must_use]
pub fn options(target: impl Into<String>) -> Self {
Self::new(HttpMethod::Options, target)
}
#[must_use]
pub fn trace(target: impl Into<String>) -> Self {
Self::new(HttpMethod::Trace, target)
}
#[must_use]
pub fn connect(target: impl Into<String>) -> Self {
Self::new(HttpMethod::Connect, target)
}
#[must_use]
pub const fn method(&self) -> HttpMethod {
self.method
}
#[must_use]
pub fn target(&self) -> &str {
&self.target
}
#[must_use]
pub fn path(&self) -> &str {
self.target
.split_once('?')
.map_or(self.target.as_str(), |(path, _)| path)
}
#[must_use]
pub fn header(mut self, name: impl AsRef<str>, value: impl Into<String>) -> Self {
self.headers
.insert(normalize_header_name(name.as_ref()), value.into());
self
}
#[must_use]
pub fn body(mut self, body: impl Into<Vec<u8>>) -> Self {
self.body = body.into();
self
}
#[must_use]
pub const fn peer_addr(mut self, peer_addr: SocketAddr) -> Self {
self.peer_addr = Some(peer_addr);
self
}
#[must_use]
pub fn scheme(mut self, scheme: impl Into<String>) -> Self {
self.scheme = scheme.into();
self
}
pub fn json(mut self, value: &impl Serialize) -> Result<Self, blazingly_json::Error> {
self.body = blazingly_json::to_vec(value)?;
self.headers
.insert("content-type".to_owned(), "application/json".to_owned());
Ok(self)
}
#[must_use]
pub fn get_header(&self, name: &str) -> Option<&str> {
self.headers
.get(name)
.or_else(|| {
self.headers
.iter()
.find(|(header, _)| header.eq_ignore_ascii_case(name))
.map(|(_, value)| value)
})
.map(String::as_str)
}
#[must_use]
pub fn headers(&self) -> &BTreeMap<String, String> {
&self.headers
}
#[must_use]
pub fn body_bytes(&self) -> &[u8] {
&self.body
}
}
impl HttpRequestView for Request {
fn method(&self) -> HttpMethod {
self.method()
}
fn target(&self) -> &str {
self.target()
}
fn header_value(&self, name: &str, index: usize) -> Option<&str> {
self.headers
.iter()
.filter(|(header, _)| header_name_matches(header, name))
.nth(index)
.map(|(_, value)| value.as_str())
}
fn body(&self) -> &[u8] {
self.body_bytes()
}
fn peer_addr(&self) -> Option<SocketAddr> {
self.peer_addr
}
fn scheme(&self) -> &str {
&self.scheme
}
}
#[derive(Debug)]
pub struct Response {
status: u16,
headers: ResponseHeaders,
body: Vec<u8>,
stream: Option<StreamingBody>,
upgrade: Option<HttpUpgrade>,
background: Vec<BackgroundTask>,
}
impl Response {
#[must_use]
pub fn from_bytes(status: u16, body: impl Into<Vec<u8>>) -> Self {
Self {
status,
headers: ResponseHeaders::empty(),
body: body.into(),
stream: None,
upgrade: None,
background: Vec::new(),
}
}
#[must_use]
pub fn empty(status: u16) -> Self {
Self::from_bytes(status, Vec::new())
}
#[must_use]
pub fn payload_too_large(max_body_bytes: usize) -> Self {
BodyRejection::PayloadTooLarge { max_body_bytes }.into_response()
}
#[must_use]
pub const fn status(&self) -> u16 {
self.status
}
pub const fn set_status(&mut self, status: u16) {
self.status = status;
}
#[must_use]
pub fn get_header(&self, name: &str) -> Option<&str> {
self.headers.get(name)
}
pub fn headers(&self) -> impl Iterator<Item = (&str, &str)> {
self.headers.iter()
}
pub fn set_header(&mut self, name: impl AsRef<str>, value: impl Into<String>) {
self.headers.insert(
Cow::Owned(normalize_header_name(name.as_ref())),
Cow::Owned(value.into()),
);
}
pub fn remove_header(&mut self, name: &str) {
self.headers.remove(name);
}
#[must_use]
pub fn body(&self) -> &[u8] {
&self.body
}
pub fn replace_body(&mut self, body: impl Into<Vec<u8>>) -> bool {
if self.stream.is_some() {
return false;
}
self.body = body.into();
true
}
#[must_use]
pub const fn is_streaming(&self) -> bool {
self.stream.is_some()
}
#[must_use]
pub fn take_body_stream(&mut self) -> Option<StreamingBody> {
self.stream.take()
}
pub fn set_body_stream(&mut self, stream: StreamingBody) {
self.body.clear();
self.stream = Some(stream);
}
#[must_use]
pub const fn is_upgrade(&self) -> bool {
self.upgrade.is_some()
}
pub fn take_upgrade(&mut self) -> Option<HttpUpgrade> {
self.upgrade.take()
}
pub fn take_background_tasks(&mut self) -> Vec<BackgroundTask> {
std::mem::take(&mut self.background)
}
pub async fn run_background(&mut self) -> Result<(), BackgroundTaskError> {
let mut first_error = None;
for task in self.take_background_tasks() {
if let Err(error) = task.run().await
&& first_error.is_none()
{
first_error = Some(error);
}
}
first_error.map_or(Ok(()), Err)
}
#[must_use]
pub fn exact_body_length(&self) -> Option<u64> {
self.stream.as_ref().map_or_else(
|| u64::try_from(self.body.len()).ok(),
StreamingBody::exact_length,
)
}
pub async fn next_body_chunk(&mut self) -> Option<Result<Vec<u8>, BodyStreamError>> {
match self.stream.as_mut() {
Some(stream) => stream.next_chunk().await,
None => None,
}
}
pub async fn collect_body(mut self, limit: usize) -> Result<Vec<u8>, CollectBodyError> {
if self.stream.is_none() {
if self.body.len() > limit {
return Err(CollectBodyError::LimitExceeded { limit });
}
return Ok(self.body);
}
let initial_capacity = self
.exact_body_length()
.and_then(|length| usize::try_from(length).ok())
.unwrap_or(0)
.min(limit);
let mut body = Vec::with_capacity(initial_capacity);
while let Some(chunk) = self.next_body_chunk().await {
let chunk = chunk.map_err(CollectBodyError::Stream)?;
if body.len().saturating_add(chunk.len()) > limit {
return Err(CollectBodyError::LimitExceeded { limit });
}
body.extend_from_slice(&chunk);
}
Ok(body)
}
pub fn text(&self) -> Result<&str, Utf8Error> {
std::str::from_utf8(&self.body)
}
pub fn json<T: DeserializeOwned>(&self) -> Result<T, blazingly_json::Error> {
blazingly_json::from_slice(&self.body)
}
}
#[derive(Clone, Debug, Eq, PartialEq)]
pub enum CollectBodyError {
Stream(BodyStreamError),
LimitExceeded { limit: usize },
}
impl fmt::Display for CollectBodyError {
fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
match self {
Self::Stream(error) => error.fmt(formatter),
Self::LimitExceeded { limit } => {
write!(
formatter,
"response body exceeds the {limit}-byte collection limit"
)
}
}
}
}
impl std::error::Error for CollectBodyError {}
pub struct HttpApp {
app: ExecutableApp,
router: Router,
max_body_bytes: usize,
openapi: Option<OpenApiService>,
middleware: Vec<ScopedMiddleware>,
error_handlers: Vec<Rc<dyn HttpErrorHandler>>,
allow_unverified_security_schemes: bool,
}
impl HttpApp {
#[must_use]
pub fn new(app: ExecutableApp) -> Self {
emit_and_exit_if_requested(&app);
let router = Router::new(&app);
Self {
app,
router,
max_body_bytes: DEFAULT_MAX_BODY_BYTES,
openapi: None,
middleware: Vec::new(),
error_handlers: Vec::new(),
allow_unverified_security_schemes: false,
}
}
#[must_use]
pub const fn with_max_body_bytes(mut self, max_body_bytes: usize) -> Self {
self.max_body_bytes = max_body_bytes;
self
}
#[must_use]
pub const fn with_unverified_security_schemes(mut self, allow: bool) -> Self {
self.allow_unverified_security_schemes = allow;
self
}
pub async fn startup(&self) -> Result<(), DependencyError> {
self.app.startup().await
}
pub async fn shutdown(&self) -> Result<(), DependencyError> {
self.app.shutdown().await
}
#[must_use]
pub fn with_openapi(mut self, config: OpenApiConfig) -> Self {
self.openapi = Some(OpenApiService::new(self.app.definition(), config));
self
}
#[must_use]
pub fn with_middleware(mut self, middleware: impl HttpMiddleware + 'static) -> Self {
self.middleware.push(ScopedMiddleware {
scope: MiddlewareScope::all(),
layer: Rc::new(middleware),
});
self
}
#[must_use]
pub fn with_shared_middleware(mut self, middleware: Rc<dyn HttpMiddleware>) -> Self {
self.middleware.push(ScopedMiddleware {
scope: MiddlewareScope::all(),
layer: middleware,
});
self
}
#[must_use]
pub fn with_scoped_middleware(
mut self,
scope: MiddlewareScope,
middleware: impl HttpMiddleware + 'static,
) -> Self {
self.middleware.push(ScopedMiddleware {
scope,
layer: Rc::new(middleware),
});
self
}
#[must_use]
pub fn with_shared_scoped_middleware(
mut self,
scope: MiddlewareScope,
middleware: Rc<dyn HttpMiddleware>,
) -> Self {
self.middleware.push(ScopedMiddleware {
scope,
layer: middleware,
});
self
}
#[must_use]
pub fn with_error_handler(mut self, handler: impl HttpErrorHandler + 'static) -> Self {
self.error_handlers.push(Rc::new(handler));
self
}
#[must_use]
pub fn with_shared_error_handler(mut self, handler: Rc<dyn HttpErrorHandler>) -> Self {
self.error_handlers.push(handler);
self
}
pub async fn call(&self, request: Request) -> Response {
self.call_view(&request).await
}
pub async fn call_view(&self, request: &impl HttpRequestView) -> Response {
let dispatcher = self.dispatcher();
dispatcher.dispatch(request, None).await
}
pub async fn call_view_controlled(
&self,
request: &impl HttpRequestView,
control: InvocationControl,
) -> Response {
let dispatcher = self.dispatcher();
dispatcher.dispatch(request, Some(control)).await
}
fn dispatcher(&self) -> Dispatcher<'_> {
Dispatcher {
app: &self.app,
router: &self.router,
max_body_bytes: self.max_body_bytes,
openapi: self.openapi.as_ref(),
middleware: &self.middleware,
error_handlers: &self.error_handlers,
allow_unverified_security_schemes: self.allow_unverified_security_schemes,
}
}
#[must_use]
pub fn request_body_source(&self, method: HttpMethod, target: &str) -> Option<InputSource> {
let path = target.split_once('?').map_or(target, |(path, _)| path);
self.router
.recognize(method, path)
.ok()
.and_then(|route| route.body_source())
}
}
fn emit_and_exit_if_requested(app: &ExecutableApp) {
static EMITTED: std::sync::Once = std::sync::Once::new();
let Some(mode) = std::env::var_os(EMIT_VARIABLE) else {
return;
};
if mode.is_empty() {
return;
}
let mut code = 0;
EMITTED.call_once(|| code = emit(app, &mode.to_string_lossy()));
std::process::exit(code);
}
fn emit(app: &ExecutableApp, mode: &str) -> i32 {
use std::io::Write as _;
let output = match mode {
"openapi" => match openapi_document_text(app.definition()) {
Ok(document) => document,
Err(error) => {
eprintln!("error: the OpenAPI document could not be serialized: {error}");
return 1;
}
},
"routes" => routes_table_text(app.definition()),
unknown => {
eprintln!("error: {EMIT_VARIABLE} must be `openapi` or `routes`, not `{unknown}`");
return 2;
}
};
let mut stdout = std::io::stdout().lock();
if stdout.write_all(output.as_bytes()).is_err() || stdout.flush().is_err() {
return 1;
}
0
}
fn openapi_document_text(definition: &AppDefinition) -> Result<String, blazingly_json::Error> {
let document = blazingly_openapi::to_value(definition);
let mut text = blazingly_json::to_string_pretty(&document)?;
text.push('\n');
Ok(text)
}
fn routes_table_text(definition: &AppDefinition) -> String {
use std::fmt::Write as _;
let mut table = String::from("METHOD\tPATH\tOPERATION\tSUMMARY\n");
for operation in definition.operations() {
let _ = writeln!(
table,
"{}\t{}\t{}\t{}",
operation.http.method.as_str(),
operation.http.path,
operation.contract.id.as_str(),
operation.contract.summary
);
}
table
}
pub struct TestApp<'app> {
app: &'app ExecutableApp,
router: Router,
max_body_bytes: usize,
openapi: Option<OpenApiService>,
middleware: Vec<ScopedMiddleware>,
error_handlers: Vec<Rc<dyn HttpErrorHandler>>,
allow_unverified_security_schemes: bool,
}
impl<'app> TestApp<'app> {
#[must_use]
pub fn new(app: &'app ExecutableApp) -> Self {
Self {
app,
router: Router::new(app),
max_body_bytes: DEFAULT_MAX_BODY_BYTES,
openapi: None,
middleware: Vec::new(),
error_handlers: Vec::new(),
allow_unverified_security_schemes: false,
}
}
#[must_use]
pub const fn with_max_body_bytes(mut self, max_body_bytes: usize) -> Self {
self.max_body_bytes = max_body_bytes;
self
}
#[must_use]
pub const fn with_unverified_security_schemes(mut self, allow: bool) -> Self {
self.allow_unverified_security_schemes = allow;
self
}
pub async fn startup(&self) -> Result<(), DependencyError> {
self.app.startup().await
}
pub async fn shutdown(&self) -> Result<(), DependencyError> {
self.app.shutdown().await
}
#[must_use]
pub fn with_openapi(mut self, config: OpenApiConfig) -> Self {
self.openapi = Some(OpenApiService::new(self.app.definition(), config));
self
}
#[must_use]
pub fn with_middleware(mut self, middleware: impl HttpMiddleware + 'static) -> Self {
self.middleware.push(ScopedMiddleware {
scope: MiddlewareScope::all(),
layer: Rc::new(middleware),
});
self
}
#[must_use]
pub fn with_shared_middleware(mut self, middleware: Rc<dyn HttpMiddleware>) -> Self {
self.middleware.push(ScopedMiddleware {
scope: MiddlewareScope::all(),
layer: middleware,
});
self
}
#[must_use]
pub fn with_scoped_middleware(
mut self,
scope: MiddlewareScope,
middleware: impl HttpMiddleware + 'static,
) -> Self {
self.middleware.push(ScopedMiddleware {
scope,
layer: Rc::new(middleware),
});
self
}
#[must_use]
pub fn with_shared_scoped_middleware(
mut self,
scope: MiddlewareScope,
middleware: Rc<dyn HttpMiddleware>,
) -> Self {
self.middleware.push(ScopedMiddleware {
scope,
layer: middleware,
});
self
}
#[must_use]
pub fn with_error_handler(mut self, handler: impl HttpErrorHandler + 'static) -> Self {
self.error_handlers.push(Rc::new(handler));
self
}
#[must_use]
pub fn with_shared_error_handler(mut self, handler: Rc<dyn HttpErrorHandler>) -> Self {
self.error_handlers.push(handler);
self
}
pub async fn call(&self, request: Request) -> Response {
let dispatcher = self.dispatcher();
dispatcher.dispatch(&request, None).await
}
pub async fn call_controlled(&self, request: Request, control: InvocationControl) -> Response {
let dispatcher = self.dispatcher();
dispatcher.dispatch(&request, Some(control)).await
}
fn dispatcher(&self) -> Dispatcher<'_> {
Dispatcher {
app: self.app,
router: &self.router,
max_body_bytes: self.max_body_bytes,
openapi: self.openapi.as_ref(),
middleware: &self.middleware,
error_handlers: &self.error_handlers,
allow_unverified_security_schemes: self.allow_unverified_security_schemes,
}
}
}
struct Dispatcher<'app> {
app: &'app ExecutableApp,
router: &'app Router,
max_body_bytes: usize,
openapi: Option<&'app OpenApiService>,
middleware: &'app [ScopedMiddleware],
error_handlers: &'app [Rc<dyn HttpErrorHandler>],
allow_unverified_security_schemes: bool,
}
struct DispatchSite<'dispatch> {
method: HttpMethod,
path: &'dispatch str,
operation: Option<&'dispatch OperationDescriptor>,
}
impl Dispatcher<'_> {
async fn dispatch<RequestView>(
&self,
request: &RequestView,
control: Option<InvocationControl>,
) -> Response
where
RequestView: HttpRequestView,
{
if self.middleware.is_empty() {
return self.dispatch_unlayered(request, control).await;
}
let middleware = self.middleware;
let target = request.target();
let mut site = DispatchSite {
method: request.method(),
path: target.split_once('?').map_or(target, |(path, _)| path),
operation: None,
};
let mut context = HttpRequestContext::new(request);
for layer in middleware {
if layer.scope.matches_request(site.path)
&& let Some(response) = layer.layer.on_request(&mut context)
{
return complete_response(middleware, &context, &site, response);
}
}
if validate_url_encoding(target).is_err() {
let response = self.fail(&site, invalid_url_encoding_failure());
return complete_response(middleware, &context, &site, response);
}
if let Some(response) = self
.openapi
.and_then(|service| service.handle(site.method, site.path))
{
return complete_response(middleware, &context, &site, openapi_response(response));
}
let recognized = match self.router.recognize(site.method, site.path) {
Ok(recognized) => recognized,
Err(error) => {
let response = self.fail(&site, route_miss_failure(&error));
return complete_response(middleware, &context, &site, response);
}
};
let Some(operation) = self.app.operation_at(recognized.operation_index()) else {
let response = self.fail(&site, internal_failure());
return complete_response(middleware, &context, &site, response);
};
let descriptor = operation.descriptor();
site.operation = Some(descriptor);
for layer in middleware {
if layer.matches(site.path, site.operation)
&& let Some(response) = layer.layer.on_operation(
&mut context,
descriptor,
self.app.definition().security_schemes(),
)
{
return complete_response(middleware, &context, &site, response);
}
}
if let Some(failure) = self.security_guard(site.path, descriptor) {
let response = self.fail(&site, failure);
return complete_response(middleware, &context, &site, response);
}
if let Some(body_source) = recognized.body_source() {
match validate_body(request, self.max_body_bytes, body_source) {
Ok(()) => {}
Err(rejection) => {
let response = self.fail(&site, rejection.into_failure());
return complete_response(middleware, &context, &site, response);
}
}
}
let request_parts = RoutedRequestParts {
request,
route: &recognized,
context: Some(&context),
connection: OnceCell::new(),
background: OnceCell::new(),
};
let outcome = if let Some(control) = control {
operation
.invoke_http_controlled(&request_parts, control)
.await
} else {
operation.invoke_http(&request_parts).await
};
let mut response = match outcome_result(outcome) {
Ok(response) => response,
Err(failure) => self.fail(&site, failure),
};
response.background.extend(request_parts.scheduled_tasks());
complete_response(middleware, &context, &site, response)
}
async fn dispatch_unlayered<RequestView>(
&self,
request: &RequestView,
control: Option<InvocationControl>,
) -> Response
where
RequestView: HttpRequestView + ?Sized,
{
let target = request.target();
let mut site = DispatchSite {
method: request.method(),
path: target.split_once('?').map_or(target, |(path, _)| path),
operation: None,
};
if validate_url_encoding(target).is_err() {
return self.fail(&site, invalid_url_encoding_failure());
}
if let Some(response) = self
.openapi
.and_then(|service| service.handle(site.method, site.path))
{
return openapi_response(response);
}
let recognized = match self.router.recognize(site.method, site.path) {
Ok(recognized) => recognized,
Err(error) => return self.fail(&site, route_miss_failure(&error)),
};
let Some(operation) = self.app.operation_at(recognized.operation_index()) else {
return self.fail(&site, internal_failure());
};
site.operation = Some(operation.descriptor());
if let Some(failure) = self.security_guard(site.path, operation.descriptor()) {
return self.fail(&site, failure);
}
if let Some(body_source) = recognized.body_source() {
match validate_body(request, self.max_body_bytes, body_source) {
Ok(()) => {}
Err(rejection) => return self.fail(&site, rejection.into_failure()),
}
}
let request_parts = RoutedRequestParts {
request,
route: &recognized,
context: None,
connection: OnceCell::new(),
background: OnceCell::new(),
};
let outcome = if let Some(control) = control {
operation
.invoke_http_controlled(&request_parts, control)
.await
} else {
operation.invoke_http(&request_parts).await
};
let mut response = match outcome_result(outcome) {
Ok(response) => response,
Err(failure) => self.fail(&site, failure),
};
response.background.extend(request_parts.scheduled_tasks());
response
}
fn fail(&self, site: &DispatchSite<'_>, mut failure: Failure) -> Response {
let mut response = failure.response();
if self.error_handlers.is_empty() {
return response;
}
let error = HttpError {
source: failure.source,
status: failure.status,
code: &failure.code,
message: &failure.message,
method: site.method,
path: site.path,
operation: site.operation,
};
for handler in self.error_handlers {
handler.on_error(&error, &mut response);
}
if failure.source == HttpErrorSource::Domain {
response.status = failure.status;
}
response
}
fn security_guard(&self, path: &str, descriptor: &OperationDescriptor) -> Option<Failure> {
if self.allow_unverified_security_schemes || descriptor.contract.security.is_empty() {
return None;
}
if self
.middleware
.iter()
.any(|layer| layer.layer.verifies_security() && layer.matches(path, Some(descriptor)))
{
return None;
}
Some(Failure::new(
HttpErrorSource::Internal,
500,
"security_verifier_missing",
"the operation declares a security scheme with no registered verifier",
))
}
}
struct Failure {
source: HttpErrorSource,
status: u16,
code: Cow<'static, str>,
message: Cow<'static, str>,
details: Option<Value>,
headers: Vec<ResponseHeader>,
}
impl Failure {
fn new(
source: HttpErrorSource,
status: u16,
code: impl Into<Cow<'static, str>>,
message: impl Into<Cow<'static, str>>,
) -> Self {
Self {
source,
status,
code: code.into(),
message: message.into(),
details: None,
headers: Vec::new(),
}
}
fn with_details(mut self, details: Value) -> Self {
self.details = Some(details);
self
}
fn with_headers(mut self, headers: Vec<ResponseHeader>) -> Self {
self.headers = headers;
self
}
fn response(&mut self) -> Response {
let response = error_response(self.status, &self.code, &self.message, self.details.take());
with_outcome_headers(response, std::mem::take(&mut self.headers))
}
fn into_response(mut self) -> Response {
self.response()
}
}
fn invalid_url_encoding_failure() -> Failure {
Failure::new(
HttpErrorSource::Request,
400,
"invalid_url_encoding",
"request target contains invalid percent encoding",
)
}
fn route_miss_failure(error: &RouteError) -> Failure {
match error {
RouteError::MethodNotAllowed { allowed } => {
let allow = allowed
.iter()
.map(|method| method.as_str())
.collect::<Vec<_>>()
.join(", ");
Failure::new(
HttpErrorSource::Routing,
405,
"method_not_allowed",
"HTTP method not allowed",
)
.with_headers(vec![ResponseHeader::new("allow", allow)])
}
RouteError::NotFound => Failure::new(
HttpErrorSource::Routing,
404,
"not_found",
"HTTP route not found",
),
}
}
fn complete_response(
middleware: &[ScopedMiddleware],
context: &HttpRequestContext<'_>,
site: &DispatchSite<'_>,
mut response: Response,
) -> Response {
for layer in middleware.iter().rev() {
if layer.matches(site.path, site.operation) {
layer
.layer
.on_response(context, site.operation, &mut response);
}
}
response
}
fn openapi_response(asset: OpenApiAssetResponse) -> Response {
let mut response = Response {
status: asset.status,
headers: ResponseHeaders::empty(),
body: asset.body,
stream: None,
upgrade: None,
background: Vec::new(),
};
for (name, value) in asset.headers {
response = response.with_header(name, value);
}
response
}
#[derive(Default)]
struct RouteNode {
static_children: HashMap<String, usize>,
parameter_child: Option<usize>,
endpoints: BTreeMap<HttpMethod, CompiledEndpoint>,
}
#[derive(Clone)]
struct CompiledEndpoint {
operation_index: usize,
parameter_names: Vec<String>,
body_source: Option<InputSource>,
}
pub struct Router {
nodes: Vec<RouteNode>,
static_routes: HashMap<HttpMethod, HashMap<String, CompiledEndpoint>>,
}
impl Router {
#[must_use]
pub fn new(app: &ExecutableApp) -> Self {
let mut router = Self {
nodes: vec![RouteNode::default()],
static_routes: HashMap::new(),
};
for descriptor in app.definition().operations() {
let Some(operation_index) = app.operation_index(&descriptor.contract.id) else {
continue;
};
router.insert(descriptor, operation_index);
}
router
}
fn insert(&mut self, descriptor: &OperationDescriptor, operation_index: usize) {
let endpoint = CompiledEndpoint {
operation_index,
parameter_names: route_segments(&descriptor.http.path)
.filter_map(path_parameter_name)
.map(str::to_owned)
.collect(),
body_source: body_source(descriptor),
};
if endpoint.parameter_names.is_empty() {
self.static_routes
.entry(descriptor.http.method)
.or_default()
.insert(descriptor.http.path.clone(), endpoint);
return;
}
let mut node_index = 0;
for segment in route_segments(&descriptor.http.path) {
if path_parameter_name(segment).is_some() {
node_index = if let Some(child) = self.nodes[node_index].parameter_child {
child
} else {
let child = self.nodes.len();
self.nodes.push(RouteNode::default());
self.nodes[node_index].parameter_child = Some(child);
child
};
} else if let Some(child) = self.nodes[node_index].static_children.get(segment) {
node_index = *child;
} else {
let child = self.nodes.len();
self.nodes.push(RouteNode::default());
self.nodes[node_index]
.static_children
.insert(segment.to_owned(), child);
node_index = child;
}
}
self.nodes[node_index]
.endpoints
.insert(descriptor.http.method, endpoint);
}
pub fn recognize<'router, 'path>(
&'router self,
method: HttpMethod,
path: &'path str,
) -> Result<RouteMatch<'router, 'path>, RouteError> {
if let Some(endpoint) = self
.static_routes
.get(&method)
.and_then(|routes| routes.get(path))
{
return Ok(RouteMatch {
endpoint,
captures: CapturedSegments::new(),
});
}
if let Some((endpoint, captures)) =
self.find_dynamic(0, route_segments(path), method, CapturedSegments::new())
{
return Ok(RouteMatch { endpoint, captures });
}
let mut allowed = self
.static_routes
.iter()
.filter_map(|(allowed_method, routes)| {
routes.contains_key(path).then_some(*allowed_method)
})
.collect::<Vec<_>>();
self.collect_dynamic_methods(0, route_segments(path), &mut allowed);
allowed.sort_unstable();
allowed.dedup();
if allowed.is_empty() {
Err(RouteError::NotFound)
} else {
Err(RouteError::MethodNotAllowed { allowed })
}
}
fn find_dynamic<'router, 'path, I>(
&'router self,
node_index: usize,
mut segments: I,
method: HttpMethod,
captures: CapturedSegments<'path>,
) -> Option<(&'router CompiledEndpoint, CapturedSegments<'path>)>
where
I: Iterator<Item = &'path str> + Clone,
{
let Some(segment) = segments.next() else {
return self.nodes[node_index]
.endpoints
.get(&method)
.map(|endpoint| (endpoint, captures));
};
if let Some(child) = self.nodes[node_index].static_children.get(segment)
&& let Some(found) =
self.find_dynamic(*child, segments.clone(), method, captures.clone())
{
return Some(found);
}
let child = self.nodes[node_index].parameter_child?;
let mut captures = captures;
captures.push(segment);
self.find_dynamic(child, segments, method, captures)
}
fn collect_dynamic_methods<'path, I>(
&self,
node_index: usize,
mut segments: I,
methods: &mut Vec<HttpMethod>,
) where
I: Iterator<Item = &'path str> + Clone,
{
let Some(segment) = segments.next() else {
methods.extend(self.nodes[node_index].endpoints.keys().copied());
return;
};
if let Some(child) = self.nodes[node_index].static_children.get(segment) {
self.collect_dynamic_methods(*child, segments.clone(), methods);
}
if let Some(child) = self.nodes[node_index].parameter_child {
self.collect_dynamic_methods(child, segments, methods);
}
}
}
fn route_segments(path: &str) -> std::str::Split<'_, char> {
path.strip_prefix('/').unwrap_or(path).split('/')
}
#[derive(Clone, Debug, Eq, PartialEq)]
pub enum RouteError {
MethodNotAllowed { allowed: Vec<HttpMethod> },
NotFound,
}
pub struct RouteMatch<'router, 'path> {
endpoint: &'router CompiledEndpoint,
captures: CapturedSegments<'path>,
}
impl RouteMatch<'_, '_> {
#[must_use]
pub const fn operation_index(&self) -> usize {
self.endpoint.operation_index
}
#[must_use]
pub const fn requires_json(&self) -> bool {
matches!(self.endpoint.body_source, Some(InputSource::Json))
}
#[must_use]
pub const fn body_source(&self) -> Option<InputSource> {
self.endpoint.body_source
}
#[must_use]
pub fn path_parameter(&self, name: &str) -> Option<Cow<'_, str>> {
self.endpoint
.parameter_names
.iter()
.position(|parameter| parameter == name)
.and_then(|position| self.captures.get(position))
.and_then(|value| decode_url_component(value, false).ok())
}
}
const INLINE_PATH_PARAMETERS: usize = 8;
#[derive(Clone)]
enum CapturedSegments<'path> {
Inline {
values: [Option<&'path str>; INLINE_PATH_PARAMETERS],
len: usize,
},
Heap(Vec<&'path str>),
}
impl<'path> CapturedSegments<'path> {
const fn new() -> Self {
Self::Inline {
values: [None; INLINE_PATH_PARAMETERS],
len: 0,
}
}
fn push(&mut self, value: &'path str) {
match self {
Self::Inline { values, len } if *len < INLINE_PATH_PARAMETERS => {
values[*len] = Some(value);
*len += 1;
}
Self::Inline { values, len } => {
let mut heap = values[..*len]
.iter()
.filter_map(|value| *value)
.collect::<Vec<_>>();
heap.push(value);
*self = Self::Heap(heap);
}
Self::Heap(values) => values.push(value),
}
}
fn get(&self, index: usize) -> Option<&'path str> {
match self {
Self::Inline { values, len } if index < *len => values[index],
Self::Inline { .. } => None,
Self::Heap(values) => values.get(index).copied(),
}
}
}
struct RoutedRequestParts<'request, 'router, 'path, 'context, RequestView: ?Sized> {
request: &'request RequestView,
route: &'request RouteMatch<'router, 'path>,
context: Option<&'context HttpRequestContext<'request>>,
connection: OnceCell<ConnectionInfo>,
background: OnceCell<BackgroundTasks>,
}
impl<RequestView> RoutedRequestParts<'_, '_, '_, '_, RequestView>
where
RequestView: HttpRequestView + ?Sized,
{
fn connection_info(&self) -> &ConnectionInfo {
self.connection.get_or_init(|| {
self.context.map_or_else(
|| ConnectionInfo::from_request(self.request),
HttpRequestContext::connection_info,
)
})
}
fn background_tasks(&self) -> &BackgroundTasks {
self.background.get_or_init(BackgroundTasks::new)
}
fn scheduled_tasks(&self) -> Vec<BackgroundTask> {
self.background
.get()
.map_or_else(Vec::new, BackgroundTasks::take)
}
}
impl<RequestView> InvocationRequestParts for RoutedRequestParts<'_, '_, '_, '_, RequestView>
where
RequestView: HttpRequestView + ?Sized,
{
fn value(&self, source: InputSource, name: &str, index: usize) -> Option<Cow<'_, str>> {
match source {
InputSource::Path if index == 0 => self.route.path_parameter(name),
InputSource::Query => query_value(self.request.target(), name, index),
InputSource::Header => self.request.header_value(name, index).map(Cow::Borrowed),
InputSource::Cookie => cookie_value(self.request, name, index),
InputSource::Form => form_value(self.request.body(), name, index),
InputSource::Path
| InputSource::Json
| InputSource::Multipart
| InputSource::File
| InputSource::Stream => None,
}
}
fn body(&self) -> &[u8] {
self.request.body()
}
fn take_body_stream(&self) -> Option<StreamingBody> {
self.request.take_body_stream()
}
fn extension(&self, type_id: TypeId) -> Option<&dyn Any> {
if let Some(value) = self
.context
.and_then(|context| context.extension_by_id(type_id))
{
return Some(value);
}
if type_id == TypeId::of::<ConnectionInfo>() {
return Some(self.connection_info());
}
if type_id == TypeId::of::<BackgroundTasks>() {
return Some(self.background_tasks());
}
None
}
}
fn body_source(descriptor: &OperationDescriptor) -> Option<InputSource> {
descriptor
.contract
.inputs
.iter()
.map(|input| input.source)
.find(|source| {
matches!(
source,
InputSource::Json
| InputSource::Form
| InputSource::Multipart
| InputSource::File
| InputSource::Stream
)
})
}
fn cookie_value<'request>(
request: &'request (impl HttpRequestView + ?Sized),
name: &str,
index: usize,
) -> Option<Cow<'request, str>> {
let mut header_index = 0;
let mut found = 0;
while let Some(header) = request.header_value("cookie", header_index) {
for cookie in header.split(';') {
let (cookie_name, value) = cookie.trim().split_once('=').unwrap_or((cookie.trim(), ""));
if cookie_name == name {
if found == index {
return Some(Cow::Borrowed(value));
}
found += 1;
}
}
header_index += 1;
}
None
}
fn form_value<'body>(body: &'body [u8], name: &str, index: usize) -> Option<Cow<'body, str>> {
let body = std::str::from_utf8(body).ok()?;
let mut found = 0;
for pair in body.split('&').filter(|pair| !pair.is_empty()) {
let (raw_name, raw_value) = pair.split_once('=').unwrap_or((pair, ""));
let decoded_name = decode_url_component(raw_name, true).ok()?;
if decoded_name == name {
if found == index {
return decode_url_component(raw_value, true).ok();
}
found += 1;
}
}
None
}
fn path_parameter_name(segment: &str) -> Option<&str> {
segment
.strip_prefix('{')
.and_then(|segment| segment.strip_suffix('}'))
.filter(|name| !name.is_empty())
}
fn query_value<'target>(
target: &'target str,
name: &str,
index: usize,
) -> Option<Cow<'target, str>> {
let (_, query) = target.split_once('?')?;
let mut found = 0;
for pair in query.split('&').filter(|pair| !pair.is_empty()) {
let (raw_name, raw_value) = pair.split_once('=').unwrap_or((pair, ""));
let decoded_name = decode_url_component(raw_name, true).ok()?;
if decoded_name == name {
if found == index {
return decode_url_component(raw_value, true).ok();
}
found += 1;
}
}
None
}
fn header_name_matches(header: &str, argument: &str) -> bool {
header.bytes().eq(argument.bytes().map(|byte| {
if byte == b'_' {
b'-'
} else {
byte.to_ascii_lowercase()
}
}))
}
fn validate_url_encoding(target: &str) -> Result<(), ()> {
let path = target.split_once('?').map_or(target, |(path, _)| path);
decode_url_component(path, false)?;
if let Some((_, query)) = target.split_once('?') {
for pair in query.split('&') {
let (name, value) = pair.split_once('=').unwrap_or((pair, ""));
decode_url_component(name, true)?;
decode_url_component(value, true)?;
}
}
Ok(())
}
fn decode_url_component(value: &str, plus_as_space: bool) -> Result<Cow<'_, str>, ()> {
if !value.as_bytes().contains(&b'%') && (!plus_as_space || !value.as_bytes().contains(&b'+')) {
return Ok(Cow::Borrowed(value));
}
let bytes = value.as_bytes();
let mut decoded = Vec::with_capacity(bytes.len());
let mut index = 0;
while index < bytes.len() {
match bytes[index] {
b'%' => {
let high = bytes
.get(index + 1)
.copied()
.and_then(hex_value)
.ok_or(())?;
let low = bytes
.get(index + 2)
.copied()
.and_then(hex_value)
.ok_or(())?;
decoded.push((high << 4) | low);
index += 3;
}
b'+' if plus_as_space => {
decoded.push(b' ');
index += 1;
}
byte => {
decoded.push(byte);
index += 1;
}
}
}
String::from_utf8(decoded).map(Cow::Owned).map_err(|_| ())
}
const fn hex_value(byte: u8) -> Option<u8> {
match byte {
b'0'..=b'9' => Some(byte - b'0'),
b'a'..=b'f' => Some(byte - b'a' + 10),
b'A'..=b'F' => Some(byte - b'A' + 10),
_ => None,
}
}
enum BodyRejection {
PayloadTooLarge { max_body_bytes: usize },
UnsupportedMediaType,
}
impl BodyRejection {
fn into_failure(self) -> Failure {
match self {
Self::PayloadTooLarge { max_body_bytes } => Failure::new(
HttpErrorSource::Request,
413,
"payload_too_large",
"request body exceeds the configured limit",
)
.with_details(json!({ "maxBytes": max_body_bytes })),
Self::UnsupportedMediaType => Failure::new(
HttpErrorSource::Request,
415,
"unsupported_media_type",
"request body media type does not match the operation input",
),
}
}
fn into_response(self) -> Response {
self.into_failure().into_response()
}
}
fn validate_body(
request: &(impl HttpRequestView + ?Sized),
max_body_bytes: usize,
source: InputSource,
) -> Result<(), BodyRejection> {
if request.body().len() > max_body_bytes {
return Err(BodyRejection::PayloadTooLarge { max_body_bytes });
}
if source == InputSource::Stream {
return Ok(());
}
let valid_media_type = request
.header_value("content-type", 0)
.is_some_and(|content_type| match source {
InputSource::Json => is_json_media_type(content_type),
InputSource::Form => media_type_is(content_type, "application/x-www-form-urlencoded"),
InputSource::Multipart | InputSource::File => {
media_type_is(content_type, "multipart/form-data")
}
InputSource::Stream => unreachable!("streaming bodies do not require a media type"),
InputSource::Path | InputSource::Query | InputSource::Header | InputSource::Cookie => {
true
}
});
if !valid_media_type {
return Err(BodyRejection::UnsupportedMediaType);
}
Ok(())
}
fn media_type_is(value: &str, expected: &str) -> bool {
value
.split(';')
.next()
.is_some_and(|media_type| media_type.trim().eq_ignore_ascii_case(expected))
}
fn outcome_result(outcome: ExecutionOutcome) -> Result<Response, Failure> {
match outcome {
ExecutionOutcome::Success {
status,
headers,
body,
background,
} => {
let response = match body {
Some(body) => Response {
status,
headers: json_headers(),
body,
stream: None,
upgrade: None,
background: Vec::new(),
},
None => Response {
status,
headers: ResponseHeaders::empty(),
body: Vec::new(),
stream: None,
upgrade: None,
background: Vec::new(),
},
};
let mut response = with_outcome_headers(response, headers);
response.background = background;
Ok(response)
}
ExecutionOutcome::StreamingSuccess {
status,
headers,
body,
background,
} => Ok(with_outcome_headers(
Response {
status,
headers: ResponseHeaders::empty(),
body: Vec::new(),
stream: Some(body),
upgrade: None,
background,
},
headers,
)),
ExecutionOutcome::Upgrade {
upgrade,
background,
} => {
let headers = upgrade.headers().to_vec();
Ok(with_outcome_headers(
Response {
status: 101,
headers: ResponseHeaders::empty(),
body: Vec::new(),
stream: None,
upgrade: Some(upgrade),
background,
},
headers,
))
}
ExecutionOutcome::Rejected {
status,
code,
message,
details,
} => Err(Failure {
source: HttpErrorSource::Rejection,
status,
code: Cow::Owned(code),
message: Cow::Owned(message),
details,
headers: Vec::new(),
}),
ExecutionOutcome::DomainError(error) => {
let details = match error.details {
Some(details) => match blazingly_json::from_slice(&details) {
Ok(details) => Some(details),
Err(_) => return Err(internal_failure()),
},
None => None,
};
Err(Failure {
source: HttpErrorSource::Domain,
status: error.status,
code: Cow::Owned(error.code),
message: Cow::Owned(error.message),
details,
headers: error.headers,
})
}
ExecutionOutcome::InternalError { .. } => Err(internal_failure()),
}
}
fn with_outcome_headers(mut response: Response, headers: Vec<ResponseHeader>) -> Response {
for header in headers {
response = response.with_header(header.name, header.value);
}
response
}
fn error_response(status: u16, code: &str, message: &str, details: Option<Value>) -> Response {
let mut error = json!({
"error": {
"code": code,
"message": message,
}
});
if let Some(details) = details {
error["error"]["details"] = details;
}
json_response(status, &error)
}
fn internal_failure() -> Failure {
Failure::new(
HttpErrorSource::Internal,
500,
"internal_error",
"the operation could not be completed",
)
}
fn json_response(status: u16, value: &Value) -> Response {
let Ok(body) = blazingly_json::to_vec(value) else {
return Response {
status: 500,
headers: json_headers(),
body: br#"{"error":{"code":"internal_error","message":"the operation could not be completed"}}"#
.to_vec(),
stream: None,
upgrade: None,
background: Vec::new(),
};
};
Response {
status,
headers: json_headers(),
body,
stream: None,
upgrade: None,
background: Vec::new(),
}
}
fn json_headers() -> ResponseHeaders {
ResponseHeaders::one("content-type", "application/json")
}
fn is_json_media_type(value: &str) -> bool {
value
.split(';')
.next()
.is_some_and(|media_type| media_type.trim().eq_ignore_ascii_case("application/json"))
}
fn normalize_header_name(name: &str) -> String {
name.to_ascii_lowercase()
}
impl Response {
#[must_use]
pub fn with_header(mut self, name: impl AsRef<str>, value: impl Into<String>) -> Self {
self.set_header(name, value);
self
}
}
const INLINE_RESPONSE_HEADERS: usize = 4;
type OwnedHeader = (Cow<'static, str>, Cow<'static, str>);
#[derive(Clone, Debug, Eq, PartialEq)]
struct ResponseHeaders {
inline: [Option<OwnedHeader>; INLINE_RESPONSE_HEADERS],
overflow: Vec<OwnedHeader>,
}
impl ResponseHeaders {
fn empty() -> Self {
Self {
inline: std::array::from_fn(|_| None),
overflow: Vec::new(),
}
}
fn one(name: &'static str, value: &'static str) -> Self {
let mut headers = Self::empty();
headers.inline[0] = Some((Cow::Borrowed(name), Cow::Borrowed(value)));
headers
}
fn insert(&mut self, name: Cow<'static, str>, value: Cow<'static, str>) {
if !name.eq_ignore_ascii_case("set-cookie") {
if let Some((_, existing)) = self
.inline
.iter_mut()
.filter_map(Option::as_mut)
.chain(self.overflow.iter_mut())
.find(|(existing, _)| existing.eq_ignore_ascii_case(&name))
{
*existing = value;
return;
}
}
if let Some(slot) = self.inline.iter_mut().find(|slot| slot.is_none()) {
*slot = Some((name, value));
} else {
self.overflow.push((name, value));
}
}
fn remove(&mut self, name: &str) {
for slot in &mut self.inline {
if slot
.as_ref()
.is_some_and(|(existing, _)| existing.eq_ignore_ascii_case(name))
{
*slot = None;
}
}
self.overflow
.retain(|(existing, _)| !existing.eq_ignore_ascii_case(name));
}
fn get(&self, name: &str) -> Option<&str> {
self.inline
.iter()
.filter_map(Option::as_ref)
.chain(&self.overflow)
.find(|(header, _)| header.eq_ignore_ascii_case(name))
.map(|(_, value)| value.as_ref())
}
fn iter(&self) -> impl Iterator<Item = (&str, &str)> {
self.inline
.iter()
.filter_map(Option::as_ref)
.chain(&self.overflow)
.map(|(name, value)| (name.as_ref(), value.as_ref()))
}
}
#[cfg(test)]
mod tests {
use super::{
BackgroundTasks, ConnectionInfo, HttpApp, HttpError, HttpErrorHandler, HttpErrorSource,
HttpMiddleware, HttpRequestContext, MiddlewareScope, Request, Response, TestApp,
};
use blazingly_core::{
HttpMethod, OperationDescriptor, OperationFailure, PreparedJson, ResponseDescriptor,
SecurityLocation, SecurityRequirement, SecuritySchemeDescriptor, SecuritySchemeKind,
TypeDescriptor,
};
use blazingly_executor::{
ExecutableApp, ExecutableOperation, ExecutionOutcome, Extension, FromInvocation,
InputRejection, InvocationControl, InvocationInput, OperationFuture, OperationOutput,
};
use blazingly_json::{Value, json};
use futures_lite::future;
use std::cell::{Cell, RefCell};
use std::future::Future;
use std::net::{IpAddr, Ipv4Addr, SocketAddrV4};
use std::pin::Pin;
use std::rc::Rc;
use std::task::{Context, Poll};
struct PassthroughLayer;
impl HttpMiddleware for PassthroughLayer {}
struct AuditLayer;
impl HttpMiddleware for AuditLayer {
fn verifies_security(&self) -> bool {
false
}
}
struct NormalizingProxy;
impl HttpMiddleware for NormalizingProxy {
fn on_request(&self, context: &mut HttpRequestContext<'_>) -> Option<Response> {
context.set_client_ip(IpAddr::V4(Ipv4Addr::new(198, 51, 100, 7)));
context.set_scheme("https");
context.set_host("api.example");
None
}
}
fn connection_operation(
path: &str,
id: &str,
security: Vec<SecurityRequirement>,
) -> ExecutableOperation {
let descriptor = OperationDescriptor::new(
HttpMethod::Get,
path,
id,
"Reports the normalized connection",
None,
vec![ResponseDescriptor::success(
200,
Some(TypeDescriptor::new("Connection")),
)],
)
.expect("test operation id should be valid")
.with_security(security);
ExecutableOperation::typed(descriptor, |input| {
let Extension(connection) =
Extension::<ConnectionInfo>::from_invocation(&input, "connection", true)?;
let body = json!({
"scheme": connection.scheme(),
"host": connection.host(),
"clientIp": connection.client_ip().map(|address| address.to_string()),
});
Ok(Box::pin(async move {
ExecutionOutcome::Success {
status: 200,
headers: Vec::new(),
body: Some(blazingly_json::to_vec(&body).expect("connection body")),
background: Vec::new(),
}
}) as OperationFuture)
})
}
fn executable() -> ExecutableApp {
ExecutableApp::with_security_schemes(
[
connection_operation(
"/secure",
"http.secure",
vec![SecurityRequirement::new("api_key")],
),
connection_operation("/plain", "http.plain", Vec::new()),
],
[SecuritySchemeDescriptor::new(
"api_key",
SecuritySchemeKind::ApiKey {
location: SecurityLocation::Header,
name: "x-api-key".to_owned(),
},
)],
)
.expect("secured operation graph should compile")
}
fn error_code(response: &Response) -> String {
let body: Value = response.json().expect("json error body");
body["error"]["code"]
.as_str()
.expect("stable error code")
.to_owned()
}
fn prepared_operation() -> ExecutableOperation {
let descriptor = OperationDescriptor::new(
HttpMethod::Get,
"/prepared",
"http.prepared",
"Encodes a borrowed view inside the operation",
None,
vec![ResponseDescriptor::success(
200,
Some(TypeDescriptor::new("Titles")),
)],
)
.expect("test operation id should be valid");
ExecutableOperation::typed(descriptor, |_| {
Ok(Box::pin(async move {
let owned = [String::from("first"), String::from("second")];
let borrowed: Vec<&str> = owned.iter().map(String::as_str).collect();
let body = PreparedJson::<TitlesSchema>::encode(&borrowed)
.expect("the borrowed view encodes");
OperationOutput::into_execution_outcome(body)
}) as OperationFuture)
})
}
struct TitlesSchema;
impl blazingly_core::ApiSchema for TitlesSchema {
fn type_descriptor() -> TypeDescriptor {
TypeDescriptor::new("Titles")
}
}
#[test]
fn a_prepared_body_reaches_the_wire_verbatim_as_json() {
let executable = ExecutableApp::new([prepared_operation()])
.expect("prepared operation graph should compile");
let response = future::block_on(TestApp::new(&executable).call(Request::get("/prepared")));
assert_eq!(response.status(), 200);
assert_eq!(
response.get_header("content-type"),
Some("application/json")
);
assert_eq!(response.body(), br#"["first","second"]"#);
}
#[test]
fn unlayered_dispatch_fails_closed_for_a_declared_scheme() {
let executable = executable();
let response = future::block_on(TestApp::new(&executable).call(Request::get("/secure")));
assert_eq!(response.status(), 500);
assert_eq!(error_code(&response), "security_verifier_missing");
}
#[test]
fn owned_adapter_fails_closed_for_a_declared_scheme() {
let app = HttpApp::new(executable());
let response = future::block_on(app.call(Request::get("/secure")));
assert_eq!(response.status(), 500);
assert_eq!(error_code(&response), "security_verifier_missing");
}
#[test]
fn unverified_scheme_opt_out_executes_the_operation() {
let executable = executable();
let response = future::block_on(
TestApp::new(&executable)
.with_unverified_security_schemes(true)
.call(Request::get("/secure")),
);
assert_eq!(response.status(), 200);
let owned = HttpApp::new(executable).with_unverified_security_schemes(true);
let response = future::block_on(owned.call(Request::get("/secure")));
assert_eq!(response.status(), 200);
}
#[test]
fn routes_emission_lists_every_operation() {
let executable = executable();
let table = super::routes_table_text(executable.definition());
assert!(table.starts_with("METHOD\tPATH\tOPERATION\tSUMMARY\n"));
assert!(table.contains("GET\t/secure\thttp.secure\tReports the normalized connection\n"));
assert!(table.contains("GET\t/plain\thttp.plain\tReports the normalized connection\n"));
}
#[test]
fn openapi_emission_serializes_the_default_document() {
let executable = executable();
let text = super::openapi_document_text(executable.definition())
.expect("the OpenAPI document serializes");
assert!(text.ends_with('\n'));
let document: Value = blazingly_json::from_str(&text).expect("emitted JSON parses");
assert_eq!(document["openapi"].as_str(), Some("3.1.0"));
assert_eq!(
document["paths"]["/secure"]["get"]["operationId"].as_str(),
Some("http.secure")
);
assert_eq!(
document["paths"]["/plain"]["get"]["operationId"].as_str(),
Some("http.plain")
);
}
#[test]
fn unsecured_operations_are_untouched_by_the_guard() {
let executable = executable();
let response = future::block_on(TestApp::new(&executable).call(Request::get("/plain")));
assert_eq!(response.status(), 200);
}
#[test]
fn layered_dispatch_runs_the_same_guard() {
let executable = executable();
let audited = future::block_on(
TestApp::new(&executable)
.with_middleware(AuditLayer)
.call(Request::get("/secure")),
);
assert_eq!(audited.status(), 500);
assert_eq!(error_code(&audited), "security_verifier_missing");
let verified = future::block_on(
TestApp::new(&executable)
.with_middleware(PassthroughLayer)
.call(Request::get("/secure")),
);
assert_eq!(verified.status(), 200);
}
type EventLog = Rc<RefCell<Vec<String>>>;
type HandleExtractor = fn(&InvocationInput<'_>) -> Result<BackgroundTasks, InputRejection>;
struct RecordingLayer {
name: &'static str,
events: EventLog,
}
impl HttpMiddleware for RecordingLayer {
fn on_request(&self, _context: &mut HttpRequestContext<'_>) -> Option<Response> {
self.events
.borrow_mut()
.push(format!("{}:request", self.name));
None
}
fn on_operation(
&self,
_context: &mut HttpRequestContext<'_>,
operation: &OperationDescriptor,
_security_schemes: &[SecuritySchemeDescriptor],
) -> Option<Response> {
self.events.borrow_mut().push(format!(
"{}:operation:{}",
self.name,
operation.contract.id.as_str()
));
None
}
fn on_response(
&self,
_context: &HttpRequestContext<'_>,
_operation: Option<&OperationDescriptor>,
_response: &mut Response,
) {
self.events
.borrow_mut()
.push(format!("{}:response", self.name));
}
}
struct StatusStamp;
impl HttpMiddleware for StatusStamp {
fn on_response(
&self,
_context: &HttpRequestContext<'_>,
_operation: Option<&OperationDescriptor>,
response: &mut Response,
) {
let status = response.status().to_string();
response.set_header("x-final-status", status);
}
fn verifies_security(&self) -> bool {
false
}
}
struct HouseStyle;
impl HttpErrorHandler for HouseStyle {
fn on_error(&self, error: &HttpError<'_>, response: &mut Response) {
response.set_header("x-error-source", format!("{:?}", error.source()));
match error.source() {
HttpErrorSource::Internal => {
response.set_status(503);
response.replace_body(br#"{"error":{"code":"unavailable"}}"#.to_vec());
}
HttpErrorSource::Domain => response.set_status(500),
HttpErrorSource::Routing
| HttpErrorSource::Request
| HttpErrorSource::Rejection => {}
}
}
}
fn scoped_executable() -> ExecutableApp {
ExecutableApp::new([
connection_operation("/ingest/events", "ingest.events", Vec::new()),
connection_operation("/status", "status.read", Vec::new()),
])
.expect("scoped operation graph should compile")
}
fn ack_descriptor(path: &str, id: &str) -> OperationDescriptor {
OperationDescriptor::new(
HttpMethod::Get,
path,
id,
"Reports an acknowledgement",
None,
vec![ResponseDescriptor::success(
200,
Some(TypeDescriptor::new("Ack")),
)],
)
.expect("test operation id should be valid")
}
fn outcome_operation(
path: &str,
id: &str,
outcome: fn() -> ExecutionOutcome,
) -> ExecutableOperation {
ExecutableOperation::typed(ack_descriptor(path, id), move |_| {
Ok(Box::pin(async move { outcome() }) as OperationFuture)
})
}
fn extension_handle(input: &InvocationInput<'_>) -> Result<BackgroundTasks, InputRejection> {
Extension::<BackgroundTasks>::from_invocation(input, "background", true)
.map(|Extension(tasks)| tasks)
}
fn bare_handle(input: &InvocationInput<'_>) -> Result<BackgroundTasks, InputRejection> {
BackgroundTasks::from_invocation(input, "background", true)
}
fn scheduling_operation(
path: &str,
id: &str,
log: &EventLog,
extract: HandleExtractor,
outcome: fn() -> ExecutionOutcome,
) -> ExecutableOperation {
let log = Rc::clone(log);
ExecutableOperation::typed(ack_descriptor(path, id), move |input| {
let tasks = extract(&input)?;
let log = Rc::clone(&log);
tasks.add_infallible(move || async move {
log.borrow_mut().push("task".to_owned());
});
Ok(Box::pin(async move { outcome() }) as OperationFuture)
})
}
fn accepted() -> ExecutionOutcome {
ExecutionOutcome::Success {
status: 200,
headers: Vec::new(),
body: None,
background: Vec::new(),
}
}
fn conflict() -> ExecutionOutcome {
ExecutionOutcome::DomainError(OperationFailure::new(
409,
"conflict",
"the event was already ingested",
))
}
#[test]
fn a_path_prefix_matches_only_on_segment_boundaries() {
let scope = MiddlewareScope::prefix("/ingest");
assert!(scope.matches_request("/ingest"));
assert!(scope.matches_request("/ingest/events"));
assert!(!scope.matches_request("/ingested"));
assert!(!scope.matches_request("/"));
assert!(MiddlewareScope::prefix("ingest/").matches_request("/ingest/events"));
assert!(MiddlewareScope::all().is_global());
assert!(!scope.is_global());
}
#[test]
fn operation_predicates_select_by_id_after_routing() {
let scope = MiddlewareScope::all().with_operation_prefix("ingest.");
assert!(scope.matches_operation("/anything", "ingest.events"));
assert!(!scope.matches_operation("/anything", "status.read"));
assert!(!scope.matches_request("/anything"));
let scope = MiddlewareScope::all()
.with_operation_filter(|id| id.split('.').next_back() == Some("read"));
assert!(scope.matches_operation("/anything", "status.read"));
assert!(!scope.matches_operation("/anything", "status.write"));
let scope = MiddlewareScope::prefix("/ingest").with_operation("status.read");
assert!(!scope.matches_operation("/status", "status.read"));
}
#[test]
fn a_prefix_scoped_layer_observes_only_its_subtree() {
let executable = scoped_executable();
let events: EventLog = Rc::new(RefCell::new(Vec::new()));
let app = TestApp::new(&executable).with_scoped_middleware(
MiddlewareScope::prefix("/ingest"),
RecordingLayer {
name: "ingest",
events: Rc::clone(&events),
},
);
future::block_on(app.call(Request::get("/ingest/events")));
let recorded = events.borrow().clone();
assert_eq!(
recorded,
[
"ingest:request",
"ingest:operation:ingest.events",
"ingest:response"
]
);
events.borrow_mut().clear();
future::block_on(app.call(Request::get("/status")));
assert!(events.borrow().is_empty());
}
#[test]
fn an_operation_scoped_layer_starts_after_routing() {
let executable = scoped_executable();
let events: EventLog = Rc::new(RefCell::new(Vec::new()));
let app = TestApp::new(&executable).with_scoped_middleware(
MiddlewareScope::operation("ingest.events"),
RecordingLayer {
name: "op",
events: Rc::clone(&events),
},
);
future::block_on(app.call(Request::get("/ingest/events")));
let recorded = events.borrow().clone();
assert_eq!(recorded, ["op:operation:ingest.events", "op:response"]);
events.borrow_mut().clear();
future::block_on(app.call(Request::get("/status")));
assert!(events.borrow().is_empty());
}
#[test]
fn a_scoped_verifier_does_not_cover_operations_outside_its_scope() {
let executable = executable();
let outside = future::block_on(
TestApp::new(&executable)
.with_scoped_middleware(MiddlewareScope::prefix("/other"), PassthroughLayer)
.call(Request::get("/secure")),
);
assert_eq!(outside.status(), 500);
assert_eq!(error_code(&outside), "security_verifier_missing");
let inside = future::block_on(
TestApp::new(&executable)
.with_scoped_middleware(MiddlewareScope::prefix("/secure"), PassthroughLayer)
.call(Request::get("/secure")),
);
assert_eq!(inside.status(), 200);
}
#[test]
fn an_injected_handle_schedules_work_from_the_handler_body() {
let log: EventLog = Rc::new(RefCell::new(Vec::new()));
let executable = ExecutableApp::new([scheduling_operation(
"/ingest",
"tasks.ok",
&log,
extension_handle,
accepted,
)])
.expect("scheduling operation graph should compile");
let mut response =
future::block_on(TestApp::new(&executable).call(Request::get("/ingest")));
assert_eq!(response.status(), 200);
assert!(log.borrow().is_empty());
future::block_on(response.run_background()).expect("scheduled task");
assert_eq!(log.borrow().clone(), ["task"]);
}
#[test]
fn scheduled_work_survives_a_failed_outcome() {
let log: EventLog = Rc::new(RefCell::new(Vec::new()));
let executable = ExecutableApp::new([scheduling_operation(
"/ingest",
"tasks.conflict",
&log,
bare_handle,
conflict,
)])
.expect("scheduling operation graph should compile");
let mut response =
future::block_on(TestApp::new(&executable).call(Request::get("/ingest")));
assert_eq!(response.status(), 409);
assert_eq!(error_code(&response), "conflict");
future::block_on(response.run_background()).expect("scheduled task");
assert_eq!(log.borrow().clone(), ["task"]);
}
struct ReadyWhenScheduled {
scheduled: Rc<Cell<bool>>,
}
impl Future for ReadyWhenScheduled {
type Output = ();
fn poll(self: Pin<&mut Self>, _context: &mut Context<'_>) -> Poll<()> {
if self.scheduled.get() {
Poll::Ready(())
} else {
Poll::Pending
}
}
}
fn aborting_operation(log: &EventLog, scheduled: &Rc<Cell<bool>>) -> ExecutableOperation {
let log = Rc::clone(log);
let scheduled = Rc::clone(scheduled);
ExecutableOperation::typed(ack_descriptor("/ingest", "tasks.aborted"), move |input| {
let tasks = bare_handle(&input)?;
let log = Rc::clone(&log);
tasks.add_infallible(move || async move {
log.borrow_mut().push("task".to_owned());
});
scheduled.set(true);
Ok(Box::pin(async move { accepted() }) as OperationFuture)
})
}
#[test]
fn scheduled_work_survives_an_aborted_invocation() {
let log: EventLog = Rc::new(RefCell::new(Vec::new()));
let scheduled = Rc::new(Cell::new(false));
let executable = ExecutableApp::new([aborting_operation(&log, &scheduled)])
.expect("scheduling operation graph should compile");
let control = InvocationControl::new().with_timeout(ReadyWhenScheduled {
scheduled: Rc::clone(&scheduled),
});
let mut response = future::block_on(
TestApp::new(&executable).call_controlled(Request::get("/ingest"), control),
);
assert_eq!(response.status(), 504);
future::block_on(response.run_background()).expect("scheduled task");
assert_eq!(log.borrow().clone(), ["task"]);
}
#[test]
fn an_unused_handle_leaves_the_response_untouched() {
let executable = scoped_executable();
let mut response =
future::block_on(TestApp::new(&executable).call(Request::get("/status")));
assert_eq!(response.status(), 200);
assert!(response.take_background_tasks().is_empty());
}
#[test]
fn an_error_handler_restyles_every_dispatch_failure() {
let executable = executable();
let app = TestApp::new(&executable).with_error_handler(HouseStyle);
let missing = future::block_on(app.call(Request::get("/nope")));
assert_eq!(missing.status(), 404);
assert_eq!(missing.get_header("x-error-source"), Some("Routing"));
let unverified = future::block_on(app.call(Request::get("/secure")));
assert_eq!(unverified.status(), 503);
assert_eq!(unverified.get_header("x-error-source"), Some("Internal"));
assert_eq!(error_code(&unverified), "unavailable");
let allowed = future::block_on(app.call(Request::post("/plain")));
assert_eq!(allowed.status(), 405);
assert_eq!(allowed.get_header("allow"), Some("GET"));
assert_eq!(allowed.get_header("x-error-source"), Some("Routing"));
}
#[test]
fn a_typed_domain_status_survives_the_error_handler() {
let executable = ExecutableApp::new([
outcome_operation("/conflict", "errors.conflict", conflict),
outcome_operation("/broken", "errors.broken", || {
ExecutionOutcome::InternalError {
code: "boom".to_owned(),
message: "boom".to_owned(),
}
}),
])
.expect("error operation graph should compile");
let app = TestApp::new(&executable).with_error_handler(HouseStyle);
let domain = future::block_on(app.call(Request::get("/conflict")));
assert_eq!(domain.status(), 409);
assert_eq!(error_code(&domain), "conflict");
assert_eq!(domain.get_header("x-error-source"), Some("Domain"));
let internal = future::block_on(app.call(Request::get("/broken")));
assert_eq!(internal.status(), 503);
assert_eq!(error_code(&internal), "unavailable");
}
#[test]
fn error_handlers_run_before_middleware_sees_the_response() {
let executable = executable();
let response = future::block_on(
TestApp::new(&executable)
.with_error_handler(HouseStyle)
.with_middleware(StatusStamp)
.call(Request::get("/secure")),
);
assert_eq!(response.status(), 503);
assert_eq!(response.get_header("x-final-status"), Some("503"));
}
#[test]
fn the_owned_adapter_registers_the_same_seams() {
let app = HttpApp::new(executable())
.with_scoped_middleware(MiddlewareScope::prefix("/other"), PassthroughLayer)
.with_error_handler(HouseStyle);
let response = future::block_on(app.call(Request::get("/secure")));
assert_eq!(response.status(), 503);
assert_eq!(response.get_header("x-error-source"), Some("Internal"));
}
#[test]
fn normalized_connection_reaches_the_operation_context() {
let executable = executable();
let request = || {
Request::get("/plain")
.peer_addr(SocketAddrV4::new(Ipv4Addr::LOCALHOST, 9000).into())
.header("host", "internal:8080")
};
let unlayered = future::block_on(TestApp::new(&executable).call(request()));
let body: Value = unlayered.json().expect("connection body");
assert_eq!(body["scheme"], "http");
assert_eq!(body["host"], "internal:8080");
assert_eq!(body["clientIp"], "127.0.0.1");
let layered = future::block_on(
TestApp::new(&executable)
.with_middleware(NormalizingProxy)
.call(request()),
);
let body: Value = layered.json().expect("connection body");
assert_eq!(body["scheme"], "https");
assert_eq!(body["host"], "api.example");
assert_eq!(body["clientIp"], "198.51.100.7");
}
}