use std::future::Future;
use std::pin::Pin;
use crate::primitives::canonical::Response;
use crate::primitives::request::Request;
use crate::primitives::request_body_error::RequestBodyError;
use crate::primitives::request_body_policy::RequestBodyPolicy;
#[derive(Debug)]
pub struct ServiceError {
kind: ServiceErrorKind,
message: String,
}
#[derive(Debug)]
enum ServiceErrorKind {
Internal,
Rejected(u16),
#[allow(dead_code)]
Panic,
Timeout,
}
impl ServiceError {
pub fn internal(message: impl Into<String>) -> Self {
Self {
kind: ServiceErrorKind::Internal,
message: message.into(),
}
}
pub fn rejected(status: u16, message: impl Into<String>) -> Self {
Self {
kind: ServiceErrorKind::Rejected(status),
message: message.into(),
}
}
#[allow(dead_code)]
pub(crate) fn panic(message: impl Into<String>) -> Self {
Self {
kind: ServiceErrorKind::Panic,
message: message.into(),
}
}
pub(crate) fn timeout(message: impl Into<String>) -> Self {
Self {
kind: ServiceErrorKind::Timeout,
message: message.into(),
}
}
pub fn message(&self) -> &str {
&self.message
}
pub fn is_panic(&self) -> bool {
matches!(self.kind, ServiceErrorKind::Panic)
}
pub fn is_timeout(&self) -> bool {
matches!(self.kind, ServiceErrorKind::Timeout)
}
pub(crate) fn to_response(&self) -> hyper::Response<crate::response::BoxBodyInner> {
let status = match self.kind {
ServiceErrorKind::Internal | ServiceErrorKind::Panic => {
hyper::StatusCode::INTERNAL_SERVER_ERROR
}
ServiceErrorKind::Rejected(code) => hyper::StatusCode::from_u16(code)
.unwrap_or(hyper::StatusCode::INTERNAL_SERVER_ERROR),
ServiceErrorKind::Timeout => hyper::StatusCode::GATEWAY_TIMEOUT,
};
let body = match self.kind {
ServiceErrorKind::Panic => "500 Internal Server Error\n",
ServiceErrorKind::Timeout => "504 Gateway Timeout\n",
ServiceErrorKind::Rejected(code) => match code {
400 => "400 Bad Request\n",
403 => "403 Forbidden\n",
404 => "404 Not Found\n",
405 => "405 Method Not Allowed\n",
503 => "503 Service Unavailable\n",
_ => "500 Internal Server Error\n",
},
ServiceErrorKind::Internal => "500 Internal Server Error\n",
};
crate::response::canonical_error(status, body, false)
}
}
impl std::fmt::Display for ServiceError {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self.kind {
ServiceErrorKind::Internal => write!(f, "internal error: {}", self.message),
ServiceErrorKind::Rejected(code) => {
write!(f, "rejected ({}): {}", code, self.message)
}
ServiceErrorKind::Panic => write!(f, "handler panicked: {}", self.message),
ServiceErrorKind::Timeout => write!(f, "handler timeout: {}", self.message),
}
}
}
impl std::error::Error for ServiceError {}
impl From<RequestBodyError> for ServiceError {
fn from(err: RequestBodyError) -> Self {
let status = err.to_status_code();
let message = match err {
RequestBodyError::RejectedByPolicy => "request body rejected by policy",
RequestBodyError::DeclaredLengthTooLarge { .. }
| RequestBodyError::LimitExceeded { .. } => "request body exceeds configured limit",
RequestBodyError::ReadTimeout => "request body read timed out",
RequestBodyError::PrematureEof { .. } => {
"request body ended before its declared length"
}
RequestBodyError::LengthMismatch { .. } => "request body length mismatch",
RequestBodyError::InvalidChunkFraming(_) => "invalid request body framing",
RequestBodyError::Cancelled => "request body consumption cancelled",
RequestBodyError::Disconnected => "client disconnected while sending request body",
RequestBodyError::AlreadyConsumed | RequestBodyError::MixedConsumptionMode => {
"request body was already consumed"
}
RequestBodyError::Transport(_) => "request body transport failure",
};
ServiceError::rejected(status, message)
}
}
pub trait Service: Send + Sync + 'static {
fn request_body_policy(
&self,
_head: &crate::primitives::request_head::RequestHead,
) -> RequestBodyPolicy {
RequestBodyPolicy::Reject
}
fn call(
&self,
request: Request,
) -> Pin<Box<dyn Future<Output = Result<Response, ServiceError>> + Send + '_>>;
}
pub fn service_fn<F, Fut>(f: F) -> ServiceFn<F>
where
F: Fn(Request) -> Fut + Send + Sync + 'static,
Fut: Future<Output = Result<Response, ServiceError>> + Send + 'static,
{
ServiceFn {
f,
body_policy: None,
}
}
pub fn service_fn_head<F, Fut>(f: F) -> ServiceFn<impl Fn(Request) -> Fut + Send + Sync + 'static>
where
F: Fn(crate::primitives::request_head::RequestHead) -> Fut + Send + Sync + 'static,
Fut: Future<Output = Result<Response, ServiceError>> + Send + 'static,
{
ServiceFn {
f: move |req: Request| {
let (head, _body) = req.into_head_and_body();
f(head)
},
body_policy: Some(RequestBodyPolicy::Reject),
}
}
pub fn service_fn_with_policy<F, Fut>(f: F, policy: RequestBodyPolicy) -> ServiceFn<F>
where
F: Fn(Request) -> Fut + Send + Sync + 'static,
Fut: Future<Output = Result<Response, ServiceError>> + Send + 'static,
{
ServiceFn {
f,
body_policy: Some(policy),
}
}
pub struct ServiceFn<F> {
f: F,
body_policy: Option<RequestBodyPolicy>,
}
impl<F, Fut> Service for ServiceFn<F>
where
F: Fn(Request) -> Fut + Send + Sync + 'static,
Fut: Future<Output = Result<Response, ServiceError>> + Send + 'static,
{
fn request_body_policy(
&self,
_head: &crate::primitives::request_head::RequestHead,
) -> RequestBodyPolicy {
self.body_policy.unwrap_or(RequestBodyPolicy::Reject)
}
fn call(
&self,
request: Request,
) -> Pin<Box<dyn Future<Output = Result<Response, ServiceError>> + Send + '_>> {
Box::pin((self.f)(request))
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::primitives::canonical::ResponseBody as CanonicalResponseBody;
use crate::primitives::canonical::StatusCode;
use crate::primitives::connection_info::{ConnectionInfo, Scheme};
use crate::primitives::header_block::HeaderBlock;
use crate::primitives::request_body::RequestBody;
use std::net::SocketAddr;
fn make_test_request(path: &str) -> Request {
Request::new(
crate::primitives::request_head::RequestHead::new(
crate::primitives::method::Method::get(),
crate::primitives::request_target::RequestTarget::parse(path).unwrap(),
crate::primitives::version::HttpVersion::Http11,
HeaderBlock::new(),
),
RequestBody::empty(),
ConnectionInfo {
local_addr: "127.0.0.1:8000".parse::<SocketAddr>().unwrap(),
remote_addr: "127.0.0.1:12345".parse::<SocketAddr>().unwrap(),
scheme: Scheme::Http,
tls: None,
},
)
}
#[tokio::test]
async fn service_fn_calls_handler() {
let svc = service_fn(|_req: Request| async {
Ok(Response::builder()
.status(StatusCode::OK)
.body(CanonicalResponseBody::Bytes(b"ok".to_vec()))
.unwrap())
});
let req = make_test_request("/test");
let resp = svc.call(req).await.unwrap();
assert_eq!(resp.status(), StatusCode::OK);
}
#[tokio::test]
async fn custom_service_returns_bytes() {
struct ByteService;
impl Service for ByteService {
fn call(
&self,
_req: Request,
) -> Pin<Box<dyn Future<Output = Result<Response, ServiceError>> + Send + '_>>
{
Box::pin(async {
Ok(Response::builder()
.status(StatusCode::OK)
.body(CanonicalResponseBody::Bytes(b"custom bytes".to_vec()))
.unwrap())
})
}
}
let svc = ByteService;
let req = make_test_request("/test");
let resp = svc.call(req).await.unwrap();
assert_eq!(resp.status(), StatusCode::OK);
}
#[test]
fn service_error_response_codes() {
let err = ServiceError::internal("oops");
assert_eq!(
err.to_response().status(),
hyper::StatusCode::INTERNAL_SERVER_ERROR
);
let err = ServiceError::panic("crashed");
assert_eq!(
err.to_response().status(),
hyper::StatusCode::INTERNAL_SERVER_ERROR
);
let err = ServiceError::timeout("slow");
assert_eq!(
err.to_response().status(),
hyper::StatusCode::GATEWAY_TIMEOUT
);
let err = ServiceError::rejected(400, "bad");
assert_eq!(err.to_response().status(), hyper::StatusCode::BAD_REQUEST);
let err = ServiceError::rejected(403, "no");
assert_eq!(err.to_response().status(), hyper::StatusCode::FORBIDDEN);
let err = ServiceError::rejected(404, "miss");
assert_eq!(err.to_response().status(), hyper::StatusCode::NOT_FOUND);
let err = ServiceError::rejected(405, "nope");
assert_eq!(
err.to_response().status(),
hyper::StatusCode::METHOD_NOT_ALLOWED
);
let err = ServiceError::rejected(503, "busy");
assert_eq!(
err.to_response().status(),
hyper::StatusCode::SERVICE_UNAVAILABLE
);
let err = ServiceError::rejected(999, "weird");
assert_eq!(
err.to_response().status(),
hyper::StatusCode::from_u16(999).unwrap()
);
}
#[tokio::test]
async fn service_fn_with_captured_state() {
let greeting = "hello";
let svc = service_fn(move |_req: Request| {
let greeting = greeting.to_string();
async move {
Ok(Response::builder()
.status(StatusCode::OK)
.body(CanonicalResponseBody::Bytes(greeting.into_bytes()))
.unwrap())
}
});
let req = make_test_request("/test");
let resp = svc.call(req).await.unwrap();
assert_eq!(resp.status(), StatusCode::OK);
}
#[test]
fn service_fn_implements_service() {
let svc = service_fn(|_req: Request| async {
Ok(Response::builder()
.status(StatusCode::OK)
.body(CanonicalResponseBody::Empty)
.unwrap())
});
fn assert_service<S: Service>(_svc: &S) {}
assert_service(&svc);
}
#[test]
fn service_error_display() {
let err = ServiceError::internal("something broke");
assert!(err.to_string().contains("something broke"));
assert!(!err.is_panic());
assert!(!err.is_timeout());
let err = ServiceError::rejected(404, "not found");
assert!(err.to_string().contains("404"));
let err = ServiceError::panic("oops");
assert!(err.is_panic());
let err = ServiceError::timeout("too slow");
assert!(err.is_timeout());
}
#[test]
fn service_error_to_response() {
let err = ServiceError::panic("oops");
let resp = err.to_response();
assert_eq!(resp.status(), hyper::StatusCode::INTERNAL_SERVER_ERROR);
let err = ServiceError::timeout("slow");
let resp = err.to_response();
assert_eq!(resp.status(), hyper::StatusCode::GATEWAY_TIMEOUT);
let err = ServiceError::rejected(404, "nope");
let resp = err.to_response();
assert_eq!(resp.status(), hyper::StatusCode::NOT_FOUND);
}
}