From f4268471b6f4f6e2444ab0f5ee36505596760ca9 Mon Sep 17 00:00:00 2001 From: David Pedersen Date: Sun, 30 May 2021 13:24:03 +0200 Subject: [PATCH] Break things into modules --- src/body.rs | 10 +- src/error.rs | 70 ++++ src/extract.rs | 154 +++++++++ src/handler.rs | 198 ++++++++++++ src/lib.rs | 837 ++++-------------------------------------------- src/response.rs | 79 +++++ src/routing.rs | 284 ++++++++++++++++ 7 files changed, 848 insertions(+), 784 deletions(-) create mode 100644 src/error.rs create mode 100644 src/extract.rs create mode 100644 src/handler.rs create mode 100644 src/response.rs create mode 100644 src/routing.rs diff --git a/src/body.rs b/src/body.rs index 7b895383..44152377 100644 --- a/src/body.rs +++ b/src/body.rs @@ -1,21 +1,23 @@ use bytes::Buf; -use http_body::{Body, Empty}; +use http_body::{Body as _, Empty}; use std::{ fmt, pin::Pin, task::{Context, Poll}, }; +pub use hyper::body::Body; + /// A boxed [`Body`] trait object. pub struct BoxBody { - inner: Pin + Send + Sync + 'static>>, + inner: Pin + Send + Sync + 'static>>, } impl BoxBody { /// Create a new `BoxBody`. pub fn new(body: B) -> Self where - B: Body + Send + Sync + 'static, + B: http_body::Body + Send + Sync + 'static, D: Buf, { Self { @@ -40,7 +42,7 @@ impl fmt::Debug for BoxBody { } } -impl Body for BoxBody +impl http_body::Body for BoxBody where D: Buf, { diff --git a/src/error.rs b/src/error.rs new file mode 100644 index 00000000..0b8088b7 --- /dev/null +++ b/src/error.rs @@ -0,0 +1,70 @@ +use std::convert::Infallible; + +use http::{Response, StatusCode}; +use tower::BoxError; + +#[derive(Debug, thiserror::Error)] +#[non_exhaustive] +pub enum Error { + #[error("failed to deserialize the request body")] + DeserializeRequestBody(#[source] serde_json::Error), + + #[error("failed to serialize the response body")] + SerializeResponseBody(#[source] serde_json::Error), + + #[error("failed to consume the body")] + ConsumeRequestBody(#[source] hyper::Error), + + #[error("URI contained no query string")] + QueryStringMissing, + + #[error("failed to deserialize query string")] + DeserializeQueryString(#[source] serde_urlencoded::de::Error), + + #[error("failed generating the response body")] + ResponseBody(#[source] BoxError), + + #[error("handler service returned an error")] + Service(#[source] BoxError), + + #[error("request extension of type `{type_name}` was not set")] + MissingExtension { type_name: &'static str }, +} + +impl From for Error { + fn from(err: Infallible) -> Self { + match err {} + } +} + +pub(crate) fn handle_error(error: Error) -> Result, Error> +where + B: Default, +{ + fn make_response(status: StatusCode) -> Result, Error> + where + B: Default, + { + let mut res = Response::new(B::default()); + *res.status_mut() = status; + Ok(res) + } + + match error { + Error::DeserializeRequestBody(_) + | Error::QueryStringMissing + | Error::DeserializeQueryString(_) => make_response(StatusCode::BAD_REQUEST), + + Error::MissingExtension { .. } | Error::SerializeResponseBody(_) => { + make_response(StatusCode::INTERNAL_SERVER_ERROR) + } + + Error::Service(err) => match err.downcast::() { + Ok(err) => Err(*err), + Err(err) => Err(Error::Service(err)), + }, + + err @ Error::ConsumeRequestBody(_) => Err(err), + err @ Error::ResponseBody(_) => Err(err), + } +} diff --git a/src/extract.rs b/src/extract.rs new file mode 100644 index 00000000..acfccf6f --- /dev/null +++ b/src/extract.rs @@ -0,0 +1,154 @@ +use crate::{body::Body, Error}; +use futures_util::{future, ready}; +use http::Request; +use pin_project::pin_project; +use serde::de::DeserializeOwned; +use std::{ + future::Future, + pin::Pin, + task::{Context, Poll}, +}; + +pub trait FromRequest: Sized { + type Future: Future> + Send; + + fn from_request(req: &mut Request) -> Self::Future; +} + +impl FromRequest for Option +where + T: FromRequest, +{ + type Future = OptionFromRequestFuture; + + fn from_request(req: &mut Request) -> Self::Future { + OptionFromRequestFuture(T::from_request(req)) + } +} + +#[pin_project] +pub struct OptionFromRequestFuture(#[pin] F); + +impl Future for OptionFromRequestFuture +where + F: Future>, +{ + type Output = Result, Error>; + + fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll { + let value = ready!(self.project().0.poll(cx)); + Poll::Ready(Ok(value.ok())) + } +} + +#[derive(Debug, Clone, Copy)] +pub struct Query(T); + +impl Query { + pub fn into_inner(self) -> T { + self.0 + } +} + +impl FromRequest for Query +where + T: DeserializeOwned + Send, +{ + type Future = future::Ready>; + + fn from_request(req: &mut Request) -> Self::Future { + let result = (|| { + let query = req.uri().query().ok_or(Error::QueryStringMissing)?; + let value = serde_urlencoded::from_str(query).map_err(Error::DeserializeQueryString)?; + Ok(Query(value)) + })(); + + future::ready(result) + } +} + +#[derive(Debug, Clone, Copy)] +pub struct Json(T); + +impl Json { + pub fn into_inner(self) -> T { + self.0 + } +} + +impl FromRequest for Json +where + T: DeserializeOwned, +{ + type Future = future::BoxFuture<'static, Result>; + + fn from_request(req: &mut Request) -> Self::Future { + // TODO(david): require the body to have `content-type: application/json` + + let body = std::mem::take(req.body_mut()); + + Box::pin(async move { + let bytes = hyper::body::to_bytes(body) + .await + .map_err(Error::ConsumeRequestBody)?; + let value = serde_json::from_slice(&bytes).map_err(Error::DeserializeRequestBody)?; + Ok(Json(value)) + }) + } +} + +#[derive(Debug, Clone, Copy)] +pub struct Extension(T); + +impl Extension { + pub fn into_inner(self) -> T { + self.0 + } +} + +impl FromRequest for Extension +where + T: Clone + Send + Sync + 'static, +{ + type Future = future::Ready>; + + fn from_request(req: &mut Request) -> Self::Future { + let result = (|| { + let value = req + .extensions() + .get::() + .ok_or_else(|| Error::MissingExtension { + type_name: std::any::type_name::(), + }) + .map(|x| x.clone())?; + Ok(Extension(value)) + })(); + + future::ready(result) + } +} + +// TODO(david): can we add a length limit somehow? Maybe a const generic? +#[derive(Debug, Clone)] +pub struct Bytes(bytes::Bytes); + +impl Bytes { + pub fn into_inner(self) -> bytes::Bytes { + self.0 + } +} + +impl FromRequest for Bytes { + type Future = future::BoxFuture<'static, Result>; + + fn from_request(req: &mut Request) -> Self::Future { + let body = std::mem::take(req.body_mut()); + + Box::pin(async move { + let bytes = hyper::body::to_bytes(body) + .await + .map_err(Error::ConsumeRequestBody)?; + Ok(Bytes(bytes)) + }) + } +} diff --git a/src/handler.rs b/src/handler.rs new file mode 100644 index 00000000..7c651bff --- /dev/null +++ b/src/handler.rs @@ -0,0 +1,198 @@ +use crate::{body::Body, error::Error, extract::FromRequest, response::IntoResponse}; +use async_trait::async_trait; +use futures_util::future; +use http::{Request, Response}; +use std::{ + future::Future, + marker::PhantomData, + task::{Context, Poll}, +}; +use tower::{BoxError, Layer, Service, ServiceExt}; + +mod sealed { + pub trait HiddentTrait {} + pub struct Hidden; + impl HiddentTrait for Hidden {} +} + +#[async_trait] +pub trait Handler: Sized { + type Response: IntoResponse; + + // This seals the trait. We cannot use the regular "sealed super trait" approach + // due to coherence. + #[doc(hidden)] + type Sealed: sealed::HiddentTrait; + + async fn call(self, req: Request) -> Result; + + fn layer(self, layer: L) -> Layered + where + L: Layer>, + { + Layered::new(layer.layer(HandlerSvc::new(self))) + } +} + +#[async_trait] +impl Handler for F +where + F: Fn(Request) -> Fut + Send + Sync, + Fut: Future> + Send, + Res: IntoResponse, +{ + type Response = Res; + + type Sealed = sealed::Hidden; + + async fn call(self, req: Request) -> Result { + self(req).await + } +} + +macro_rules! impl_handler { + ( $head:ident $(,)? ) => { + #[async_trait] + #[allow(non_snake_case)] + impl Handler for F + where + F: Fn(Request, $head) -> Fut + Send + Sync, + Fut: Future> + Send, + Res: IntoResponse, + $head: FromRequest + Send, + { + type Response = Res; + + type Sealed = sealed::Hidden; + + async fn call(self, mut req: Request) -> Result { + let $head = $head::from_request(&mut req).await?; + let res = self(req, $head).await?; + Ok(res) + } + } + }; + + ( $head:ident, $($tail:ident),* $(,)? ) => { + #[async_trait] + #[allow(non_snake_case)] + impl Handler for F + where + F: Fn(Request, $head, $($tail,)*) -> Fut + Send + Sync, + Fut: Future> + Send, + Res: IntoResponse, + $head: FromRequest + Send, + $( $tail: FromRequest + Send, )* + { + type Response = Res; + + type Sealed = sealed::Hidden; + + async fn call(self, mut req: Request) -> Result { + let $head = $head::from_request(&mut req).await?; + $( + let $tail = $tail::from_request(&mut req).await?; + )* + let res = self(req, $head, $($tail,)*).await?; + Ok(res) + } + } + + impl_handler!($($tail,)*); + }; +} + +impl_handler!(T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13, T14, T15, T16); + +pub struct Layered { + svc: S, + _input: PhantomData T>, +} + +impl Clone for Layered +where + S: Clone, +{ + fn clone(&self) -> Self { + Self::new(self.svc.clone()) + } +} + +#[async_trait] +impl Handler for Layered +where + S: Service, Response = Response> + Send, + S::Error: Into, + S::Future: Send, +{ + type Response = S::Response; + + type Sealed = sealed::Hidden; + + async fn call(self, req: Request) -> Result { + self.svc + .oneshot(req) + .await + .map_err(|err| Error::Service(err.into())) + } +} + +impl Layered { + pub(crate) fn new(svc: S) -> Self { + Self { + svc, + _input: PhantomData, + } + } +} + +pub struct HandlerSvc { + handler: H, + _input: PhantomData (B, T)>, +} + +impl HandlerSvc { + pub(crate) fn new(handler: H) -> Self { + Self { + handler, + _input: PhantomData, + } + } +} + +impl Clone for HandlerSvc +where + H: Clone, +{ + fn clone(&self) -> Self { + Self { + handler: self.handler.clone(), + _input: PhantomData, + } + } +} + +impl Service> for HandlerSvc +where + H: Handler + Clone + Send + 'static, + H::Response: 'static, +{ + type Response = Response; + type Error = Error; + type Future = future::BoxFuture<'static, Result>; + + fn poll_ready(&mut self, _cx: &mut Context<'_>) -> Poll> { + // HandlerSvc can only be constructed from async functions which are always ready, or from + // `Layered` which bufferes in `::call` and is therefore also always + // ready. + Poll::Ready(Ok(())) + } + + fn call(&mut self, req: Request) -> Self::Future { + let handler = self.handler.clone(); + Box::pin(async move { + let res = Handler::call(handler, req).await?.into_response()?; + Ok(res) + }) + } +} diff --git a/src/lib.rs b/src/lib.rs index 82378f51..ad3068e0 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -4,14 +4,19 @@ Improvements to make: -Break stuff up into modules - Support extracting headers, perhaps via `headers::Header`? Tests */ +use self::{ + body::{Body, BoxBody}, + extract::FromRequest, + handler::{Handler, HandlerSvc}, + response::IntoResponse, + routing::{EmptyRouter, RouteAt}, +}; use async_trait::async_trait; use bytes::Bytes; use futures_util::{future, ready}; @@ -28,10 +33,15 @@ use std::{ }; use tower::{BoxError, Layer, Service, ServiceExt}; -mod body; -pub use body::BoxBody; +pub mod body; +pub mod extract; +pub mod handler; +pub mod response; +pub mod routing; -pub use hyper::body::Body; +mod error; + +pub use self::error::Error; pub fn app() -> App { App { @@ -57,712 +67,6 @@ impl App { } } -#[derive(Debug, Clone)] -pub struct RouteAt { - app: App, - route_spec: Bytes, -} - -impl RouteAt { - pub fn get(self, handler_fn: F) -> RouteBuilder, R>> - where - F: Handler, - { - self.add_route(handler_fn, Method::GET) - } - - pub fn get_service(self, service: S) -> RouteBuilder> - where - S: Service, Response = Response> + Clone, - S::Error: Into, - { - self.add_route_service(service, Method::GET) - } - - pub fn post(self, handler_fn: F) -> RouteBuilder, R>> - where - F: Handler, - { - self.add_route(handler_fn, Method::POST) - } - - pub fn post_service(self, service: S) -> RouteBuilder> - where - S: Service, Response = Response> + Clone, - S::Error: Into, - { - self.add_route_service(service, Method::POST) - } - - fn add_route( - self, - handler: H, - method: Method, - ) -> RouteBuilder, R>> - where - H: Handler, - { - self.add_route_service(HandlerSvc::new(handler), method) - } - - fn add_route_service(self, service: S, method: Method) -> RouteBuilder> { - let new_app = App { - router: Route { - service, - route_spec: RouteSpec { - method, - spec: self.route_spec.clone(), - }, - fallback: self.app.router, - handler_ready: false, - fallback_ready: false, - }, - }; - - RouteBuilder { - app: new_app, - route_spec: self.route_spec, - } - } -} - -pub struct RouteBuilder { - app: App, - route_spec: Bytes, -} - -impl Clone for RouteBuilder -where - R: Clone, -{ - fn clone(&self) -> Self { - Self { - app: self.app.clone(), - route_spec: self.route_spec.clone(), - } - } -} - -impl RouteBuilder { - pub fn at(self, route_spec: &str) -> RouteAt { - self.app.at(route_spec) - } - - pub fn get(self, handler_fn: F) -> RouteBuilder, R>> - where - F: Handler, - { - self.app.at_bytes(self.route_spec).get(handler_fn) - } - - pub fn get_service(self, service: S) -> RouteBuilder> - where - S: Service, Response = Response> + Clone, - S::Error: Into, - { - self.app.at_bytes(self.route_spec).get_service(service) - } - - pub fn post(self, handler_fn: F) -> RouteBuilder, R>> - where - F: Handler, - { - self.app.at_bytes(self.route_spec).post(handler_fn) - } - - pub fn post_service(self, service: S) -> RouteBuilder> - where - S: Service, Response = Response> + Clone, - S::Error: Into, - { - self.app.at_bytes(self.route_spec).post_service(service) - } - - pub fn into_service(self) -> IntoService { - IntoService { - app: self.app, - poll_ready_error: None, - } - } -} - -#[derive(Debug, thiserror::Error)] -#[non_exhaustive] -pub enum Error { - #[error("failed to deserialize the request body")] - DeserializeRequestBody(#[source] serde_json::Error), - - #[error("failed to serialize the response body")] - SerializeResponseBody(#[source] serde_json::Error), - - #[error("failed to consume the body")] - ConsumeRequestBody(#[source] hyper::Error), - - #[error("URI contained no query string")] - QueryStringMissing, - - #[error("failed to deserialize query string")] - DeserializeQueryString(#[source] serde_urlencoded::de::Error), - - #[error("failed generating the response body")] - ResponseBody(#[source] BoxError), - - #[error("handler service returned an error")] - Service(#[source] BoxError), - - #[error("request extension of type `{type_name}` was not set")] - MissingExtension { type_name: &'static str }, -} - -impl From for Error { - fn from(err: Infallible) -> Self { - match err {} - } -} - -mod sealed { - pub trait HiddentTrait {} - pub struct Hidden; - impl HiddentTrait for Hidden {} -} - -#[async_trait] -pub trait Handler: Sized { - type Response: IntoResponse; - - // This seals the trait. We cannot use the regular "sealed super trait" approach - // due to coherence. - #[doc(hidden)] - type Sealed: sealed::HiddentTrait; - - async fn call(self, req: Request) -> Result; - - fn layer(self, layer: L) -> Layered - where - L: Layer>, - { - Layered::new(layer.layer(HandlerSvc::new(self))) - } -} - -pub trait IntoResponse { - fn into_response(self) -> Result, Error>; -} - -impl IntoResponse for Response { - fn into_response(self) -> Result, Error> { - Ok(self) - } -} - -impl IntoResponse for &'static str { - fn into_response(self) -> Result, Error> { - Ok(Response::new(Body::from(self))) - } -} - -impl IntoResponse for String { - fn into_response(self) -> Result, Error> { - Ok(Response::new(Body::from(self))) - } -} - -impl IntoResponse for Bytes { - fn into_response(self) -> Result, Error> { - Ok(Response::new(Body::from(self))) - } -} - -impl IntoResponse for &'static [u8] { - fn into_response(self) -> Result, Error> { - Ok(Response::new(Body::from(self))) - } -} - -impl IntoResponse for Vec { - fn into_response(self) -> Result, Error> { - Ok(Response::new(Body::from(self))) - } -} - -impl IntoResponse for std::borrow::Cow<'static, str> { - fn into_response(self) -> Result, Error> { - Ok(Response::new(Body::from(self))) - } -} - -impl IntoResponse for std::borrow::Cow<'static, [u8]> { - fn into_response(self) -> Result, Error> { - Ok(Response::new(Body::from(self))) - } -} - -// TODO(david): rename this to Json when its in another module -pub struct JsonBody(T); - -impl IntoResponse for JsonBody -where - T: Serialize, -{ - fn into_response(self) -> Result, Error> { - let bytes = serde_json::to_vec(&self.0).map_err(Error::SerializeResponseBody)?; - let len = bytes.len(); - let mut res = Response::new(Body::from(bytes)); - - res.headers_mut().insert( - header::CONTENT_TYPE, - HeaderValue::from_static("application/json"), - ); - - res.headers_mut() - .insert(header::CONTENT_LENGTH, HeaderValue::from(len)); - - Ok(res) - } -} - -#[async_trait] -impl Handler for F -where - F: Fn(Request) -> Fut + Send + Sync, - Fut: Future> + Send, - Res: IntoResponse, -{ - type Response = Res; - - type Sealed = sealed::Hidden; - - async fn call(self, req: Request) -> Result { - self(req).await - } -} - -macro_rules! impl_handler { - ( $head:ident $(,)? ) => { - #[async_trait] - #[allow(non_snake_case)] - impl Handler for F - where - F: Fn(Request, $head) -> Fut + Send + Sync, - Fut: Future> + Send, - Res: IntoResponse, - $head: FromRequest + Send, - { - type Response = Res; - - type Sealed = sealed::Hidden; - - async fn call(self, mut req: Request) -> Result { - let $head = $head::from_request(&mut req).await?; - let res = self(req, $head).await?; - Ok(res) - } - } - }; - - ( $head:ident, $($tail:ident),* $(,)? ) => { - #[async_trait] - #[allow(non_snake_case)] - impl Handler for F - where - F: Fn(Request, $head, $($tail,)*) -> Fut + Send + Sync, - Fut: Future> + Send, - Res: IntoResponse, - $head: FromRequest + Send, - $( $tail: FromRequest + Send, )* - { - type Response = Res; - - type Sealed = sealed::Hidden; - - async fn call(self, mut req: Request) -> Result { - let $head = $head::from_request(&mut req).await?; - $( - let $tail = $tail::from_request(&mut req).await?; - )* - let res = self(req, $head, $($tail,)*).await?; - Ok(res) - } - } - - impl_handler!($($tail,)*); - }; -} - -impl_handler!(T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13, T14, T15, T16); - -pub struct Layered { - svc: S, - _input: PhantomData T>, -} - -impl Clone for Layered -where - S: Clone, -{ - fn clone(&self) -> Self { - Self::new(self.svc.clone()) - } -} - -#[async_trait] -impl Handler for Layered -where - S: Service, Response = Response> + Send, - S::Error: Into, - S::Future: Send, -{ - type Response = S::Response; - - type Sealed = sealed::Hidden; - - async fn call(self, req: Request) -> Result { - self.svc - .oneshot(req) - .await - .map_err(|err| Error::Service(err.into())) - } -} - -impl Layered { - fn new(svc: S) -> Self { - Self { - svc, - _input: PhantomData, - } - } -} - -pub struct HandlerSvc { - handler: H, - _input: PhantomData (B, T)>, -} - -impl HandlerSvc { - fn new(handler: H) -> Self { - Self { - handler, - _input: PhantomData, - } - } -} - -impl Clone for HandlerSvc -where - H: Clone, -{ - fn clone(&self) -> Self { - Self { - handler: self.handler.clone(), - _input: PhantomData, - } - } -} - -impl Service> for HandlerSvc -where - H: Handler + Clone + Send + 'static, - H::Response: 'static, -{ - type Response = Response; - type Error = Error; - type Future = future::BoxFuture<'static, Result>; - - fn poll_ready(&mut self, _cx: &mut Context<'_>) -> Poll> { - // HandlerSvc can only be constructed from async functions which are always ready, or from - // `Layered` which bufferes in `::call` and is therefore also always - // ready. - Poll::Ready(Ok(())) - } - - fn call(&mut self, req: Request) -> Self::Future { - let handler = self.handler.clone(); - Box::pin(async move { - let res = Handler::call(handler, req).await?.into_response()?; - Ok(res) - }) - } -} - -pub trait FromRequest: Sized { - type Future: Future> + Send; - - fn from_request(req: &mut Request) -> Self::Future; -} - -impl FromRequest for Option -where - T: FromRequest, -{ - type Future = OptionFromRequestFuture; - - fn from_request(req: &mut Request) -> Self::Future { - OptionFromRequestFuture(T::from_request(req)) - } -} - -#[pin_project] -pub struct OptionFromRequestFuture(#[pin] F); - -impl Future for OptionFromRequestFuture -where - F: Future>, -{ - type Output = Result, Error>; - - fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll { - let value = ready!(self.project().0.poll(cx)); - Poll::Ready(Ok(value.ok())) - } -} - -#[derive(Debug, Clone, Copy)] -pub struct Query(T); - -impl Query { - pub fn into_inner(self) -> T { - self.0 - } -} - -impl FromRequest for Query -where - T: DeserializeOwned + Send, -{ - type Future = future::Ready>; - - fn from_request(req: &mut Request) -> Self::Future { - let result = (|| { - let query = req.uri().query().ok_or(Error::QueryStringMissing)?; - let value = serde_urlencoded::from_str(query).map_err(Error::DeserializeQueryString)?; - Ok(Query(value)) - })(); - - future::ready(result) - } -} - -#[derive(Debug, Clone, Copy)] -pub struct Json(T); - -impl Json { - pub fn into_inner(self) -> T { - self.0 - } -} - -impl FromRequest for Json -where - T: DeserializeOwned, -{ - type Future = future::BoxFuture<'static, Result>; - - fn from_request(req: &mut Request) -> Self::Future { - // TODO(david): require the body to have `content-type: application/json` - - let body = std::mem::take(req.body_mut()); - - Box::pin(async move { - let bytes = hyper::body::to_bytes(body) - .await - .map_err(Error::ConsumeRequestBody)?; - let value = serde_json::from_slice(&bytes).map_err(Error::DeserializeRequestBody)?; - Ok(Json(value)) - }) - } -} - -#[derive(Debug, Clone, Copy)] -pub struct Extension(T); - -impl Extension { - pub fn into_inner(self) -> T { - self.0 - } -} - -impl FromRequest for Extension -where - T: Clone + Send + Sync + 'static, -{ - type Future = future::Ready>; - - fn from_request(req: &mut Request) -> Self::Future { - let result = (|| { - let value = req - .extensions() - .get::() - .ok_or_else(|| Error::MissingExtension { - type_name: std::any::type_name::(), - }) - .map(|x| x.clone())?; - Ok(Extension(value)) - })(); - - future::ready(result) - } -} - -// TODO(david): rename this to Bytes when its in another module -// TODO(david): can we add a length limit somehow? Maybe a const generic? -#[derive(Debug, Clone)] -pub struct BytesBody(Bytes); - -impl BytesBody { - pub fn into_inner(self) -> Bytes { - self.0 - } -} - -impl FromRequest for BytesBody { - type Future = future::BoxFuture<'static, Result>; - - fn from_request(req: &mut Request) -> Self::Future { - let body = std::mem::take(req.body_mut()); - - Box::pin(async move { - let bytes = hyper::body::to_bytes(body) - .await - .map_err(Error::ConsumeRequestBody)?; - Ok(BytesBody(bytes)) - }) - } -} - -#[derive(Clone, Copy)] -pub struct EmptyRouter(()); - -impl Service for EmptyRouter { - type Response = Response; - type Error = Infallible; - type Future = future::Ready>; - - fn poll_ready(&mut self, _cx: &mut Context<'_>) -> Poll> { - Poll::Ready(Ok(())) - } - - fn call(&mut self, _req: R) -> Self::Future { - let mut res = Response::new(Body::empty()); - *res.status_mut() = StatusCode::NOT_FOUND; - future::ok(res) - } -} - -pub struct Route { - service: H, - route_spec: RouteSpec, - fallback: F, - handler_ready: bool, - fallback_ready: bool, -} - -impl Clone for Route -where - H: Clone, - F: Clone, -{ - fn clone(&self) -> Self { - Self { - service: self.service.clone(), - fallback: self.fallback.clone(), - route_spec: self.route_spec.clone(), - // important to reset readiness when cloning - handler_ready: false, - fallback_ready: false, - } - } -} - -#[derive(Clone)] -struct RouteSpec { - method: Method, - spec: Bytes, -} - -impl RouteSpec { - fn matches(&self, req: &Request) -> bool { - // TODO(david): support dynamic placeholders like `/users/:id` - req.method() == self.method && req.uri().path().as_bytes() == self.spec - } -} - -impl Service> for Route -where - H: Service, Response = Response>, - H::Error: Into, - HB: http_body::Body + Send + Sync + 'static, - HB::Error: Into, - - F: Service, Response = Response>, - F::Error: Into, - FB: http_body::Body + Send + Sync + 'static, - FB::Error: Into, -{ - type Response = Response>; - type Error = Error; - type Future = future::Either, BoxResponseBody>; - - fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll> { - loop { - if !self.handler_ready { - ready!(self.service.poll_ready(cx)).map_err(Into::into)?; - self.handler_ready = true; - } - - if !self.fallback_ready { - ready!(self.fallback.poll_ready(cx)).map_err(Into::into)?; - self.fallback_ready = true; - } - - if self.handler_ready && self.fallback_ready { - return Poll::Ready(Ok(())); - } - } - } - - fn call(&mut self, req: Request) -> Self::Future { - if self.route_spec.matches(&req) { - assert!( - self.handler_ready, - "handler not ready. Did you forget to call `poll_ready`?" - ); - self.handler_ready = false; - future::Either::Left(BoxResponseBody(self.service.call(req))) - } else { - assert!( - self.fallback_ready, - "fallback not ready. Did you forget to call `poll_ready`?" - ); - self.fallback_ready = false; - // TODO(david): this leads to each route creating one box body, probably not great - future::Either::Right(BoxResponseBody(self.fallback.call(req))) - } - } -} - -#[pin_project] -pub struct BoxResponseBody(#[pin] F); - -impl Future for BoxResponseBody -where - F: Future, E>>, - E: Into, - B: http_body::Body + Send + Sync + 'static, - B::Error: Into, -{ - type Output = Result>, Error>; - - fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll { - let response: Response = ready!(self.project().0.poll(cx)).map_err(Into::into)?; - let response = response.map(|body| { - // TODO(david): attempt to downcast this into `Error` - let body = body.map_err(|err| Error::ResponseBody(err.into())); - BoxBody::new(body) - }); - Poll::Ready(Ok(response)) - } -} - pub struct IntoService { app: App, poll_ready_error: Option, @@ -792,12 +96,16 @@ where #[inline] fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll> { - self.app.router.poll_ready(cx).map_err(Into::into) + if let Err(err) = ready!(self.app.router.poll_ready(cx)).map_err(Into::into) { + self.poll_ready_error = Some(err); + } + + Poll::Ready(Ok(())) } fn call(&mut self, req: T) -> Self::Future { if let Some(poll_ready_error) = self.poll_ready_error.take() { - match handle_error::(poll_ready_error) { + match error::handle_error::(poll_ready_error) { Ok(res) => { return HandleErrorFuture(Kind::Response(Some(res))); } @@ -834,44 +142,12 @@ where KindProj::Error(err) => Poll::Ready(Err(err.take().unwrap())), KindProj::Future(fut) => match ready!(fut.poll(cx)) { Ok(res) => Poll::Ready(Ok(res)), - Err(err) => Poll::Ready(handle_error(err.into())), + Err(err) => Poll::Ready(error::handle_error(err.into())), }, } } } -fn handle_error(error: Error) -> Result, Error> -where - B: Default, -{ - fn make_response(status: StatusCode) -> Result, Error> - where - B: Default, - { - let mut res = Response::new(B::default()); - *res.status_mut() = status; - Ok(res) - } - - match error { - Error::DeserializeRequestBody(_) - | Error::QueryStringMissing - | Error::DeserializeQueryString(_) => make_response(StatusCode::BAD_REQUEST), - - Error::MissingExtension { .. } | Error::SerializeResponseBody(_) => { - make_response(StatusCode::INTERNAL_SERVER_ERROR) - } - - Error::Service(err) => match err.downcast::() { - Ok(err) => Err(*err), - Err(err) => Err(Error::Service(err)), - }, - - err @ Error::ConsumeRequestBody(_) => Err(err), - err @ Error::ResponseBody(_) => Err(err), - } -} - #[cfg(test)] mod tests { #![allow(warnings)] @@ -909,44 +185,45 @@ mod tests { Ok(Response::new(Body::empty())) } - let app = - app() - // routes with functions - .at("/") - .get(root) - // routes with closures - .at("/users") - .get(|_: Request, pagination: Query| async { + let app = app() + // routes with functions + .at("/") + .get(root) + // routes with closures + .at("/users") + .get( + |_: Request, pagination: extract::Query| async { let pagination = pagination.into_inner(); assert_eq!(pagination.page, 1); assert_eq!(pagination.per_page, 30); Ok::<_, Error>("users#index".to_string()) - }) - .post( - |_: Request, - payload: Json, - _state: Extension>| async { - let payload = payload.into_inner(); - assert_eq!(payload.username, "bob"); - Ok::<_, Error>(JsonBody( - serde_json::json!({ "username": payload.username }), - )) - }, - ) - // routes with a service - .at("/service") - .get_service(service_fn(root)) - // routes with layers applied - .at("/large-static-file") - .get( - large_static_file.layer( - ServiceBuilder::new() - .layer(TimeoutLayer::new(Duration::from_secs(30))) - .layer(CompressionLayer::new()) - .into_inner(), - ), - ) - .into_service(); + }, + ) + .post( + |_: Request, + payload: extract::Json, + _state: extract::Extension>| async { + let payload = payload.into_inner(); + assert_eq!(payload.username, "bob"); + Ok::<_, Error>(response::Json( + serde_json::json!({ "username": payload.username }), + )) + }, + ) + // routes with a service + .at("/service") + .get_service(service_fn(root)) + // routes with layers applied + .at("/large-static-file") + .get( + large_static_file.layer( + ServiceBuilder::new() + .layer(TimeoutLayer::new(Duration::from_secs(30))) + .layer(CompressionLayer::new()) + .into_inner(), + ), + ) + .into_service(); // state shared by all routes, could hold db connection etc struct State {} diff --git a/src/response.rs b/src/response.rs new file mode 100644 index 00000000..f41b1cb9 --- /dev/null +++ b/src/response.rs @@ -0,0 +1,79 @@ +use crate::{Body, Error}; +use bytes::Bytes; +use http::{header, HeaderValue, Response}; +use serde::Serialize; + +pub trait IntoResponse { + fn into_response(self) -> Result, Error>; +} + +impl IntoResponse for Response { + fn into_response(self) -> Result, Error> { + Ok(self) + } +} + +impl IntoResponse for &'static str { + fn into_response(self) -> Result, Error> { + Ok(Response::new(Body::from(self))) + } +} + +impl IntoResponse for String { + fn into_response(self) -> Result, Error> { + Ok(Response::new(Body::from(self))) + } +} + +impl IntoResponse for Bytes { + fn into_response(self) -> Result, Error> { + Ok(Response::new(Body::from(self))) + } +} + +impl IntoResponse for &'static [u8] { + fn into_response(self) -> Result, Error> { + Ok(Response::new(Body::from(self))) + } +} + +impl IntoResponse for Vec { + fn into_response(self) -> Result, Error> { + Ok(Response::new(Body::from(self))) + } +} + +impl IntoResponse for std::borrow::Cow<'static, str> { + fn into_response(self) -> Result, Error> { + Ok(Response::new(Body::from(self))) + } +} + +impl IntoResponse for std::borrow::Cow<'static, [u8]> { + fn into_response(self) -> Result, Error> { + Ok(Response::new(Body::from(self))) + } +} + +pub struct Json(pub T); + +impl IntoResponse for Json +where + T: Serialize, +{ + fn into_response(self) -> Result, Error> { + let bytes = serde_json::to_vec(&self.0).map_err(Error::SerializeResponseBody)?; + let len = bytes.len(); + let mut res = Response::new(Body::from(bytes)); + + res.headers_mut().insert( + header::CONTENT_TYPE, + HeaderValue::from_static("application/json"), + ); + + res.headers_mut() + .insert(header::CONTENT_LENGTH, HeaderValue::from(len)); + + Ok(res) + } +} diff --git a/src/routing.rs b/src/routing.rs new file mode 100644 index 00000000..5fbaf3ff --- /dev/null +++ b/src/routing.rs @@ -0,0 +1,284 @@ +use crate::{ + body::{Body, BoxBody}, + error::Error, + handler::{Handler, HandlerSvc}, + App, IntoService, +}; +use bytes::Bytes; +use futures_util::{future, ready}; +use http::{Method, Request, Response, StatusCode}; +use pin_project::pin_project; +use std::{ + convert::Infallible, + future::Future, + pin::Pin, + task::{Context, Poll}, +}; +use tower::{BoxError, Layer, Service}; + +#[derive(Clone, Copy)] +pub struct EmptyRouter(pub(crate) ()); + +impl Service for EmptyRouter { + type Response = Response; + type Error = Infallible; + type Future = future::Ready>; + + fn poll_ready(&mut self, _cx: &mut Context<'_>) -> Poll> { + Poll::Ready(Ok(())) + } + + fn call(&mut self, _req: R) -> Self::Future { + let mut res = Response::new(Body::empty()); + *res.status_mut() = StatusCode::NOT_FOUND; + future::ok(res) + } +} + +#[derive(Debug, Clone)] +pub struct RouteAt { + pub(crate) app: App, + pub(crate) route_spec: Bytes, +} + +impl RouteAt { + pub fn get(self, handler_fn: F) -> RouteBuilder, R>> + where + F: Handler, + { + self.add_route(handler_fn, Method::GET) + } + + pub fn get_service(self, service: S) -> RouteBuilder> + where + S: Service, Response = Response> + Clone, + S::Error: Into, + { + self.add_route_service(service, Method::GET) + } + + pub fn post(self, handler_fn: F) -> RouteBuilder, R>> + where + F: Handler, + { + self.add_route(handler_fn, Method::POST) + } + + pub fn post_service(self, service: S) -> RouteBuilder> + where + S: Service, Response = Response> + Clone, + S::Error: Into, + { + self.add_route_service(service, Method::POST) + } + + fn add_route( + self, + handler: H, + method: Method, + ) -> RouteBuilder, R>> + where + H: Handler, + { + self.add_route_service(HandlerSvc::new(handler), method) + } + + fn add_route_service(self, service: S, method: Method) -> RouteBuilder> { + let new_app = App { + router: Route { + service, + route_spec: RouteSpec { + method, + spec: self.route_spec.clone(), + }, + fallback: self.app.router, + handler_ready: false, + fallback_ready: false, + }, + }; + + RouteBuilder { + app: new_app, + route_spec: self.route_spec, + } + } +} + +pub struct RouteBuilder { + app: App, + route_spec: Bytes, +} + +impl Clone for RouteBuilder +where + R: Clone, +{ + fn clone(&self) -> Self { + Self { + app: self.app.clone(), + route_spec: self.route_spec.clone(), + } + } +} + +impl RouteBuilder { + pub fn at(self, route_spec: &str) -> RouteAt { + self.app.at(route_spec) + } + + pub fn get(self, handler_fn: F) -> RouteBuilder, R>> + where + F: Handler, + { + self.app.at_bytes(self.route_spec).get(handler_fn) + } + + pub fn get_service(self, service: S) -> RouteBuilder> + where + S: Service, Response = Response> + Clone, + S::Error: Into, + { + self.app.at_bytes(self.route_spec).get_service(service) + } + + pub fn post(self, handler_fn: F) -> RouteBuilder, R>> + where + F: Handler, + { + self.app.at_bytes(self.route_spec).post(handler_fn) + } + + pub fn post_service(self, service: S) -> RouteBuilder> + where + S: Service, Response = Response> + Clone, + S::Error: Into, + { + self.app.at_bytes(self.route_spec).post_service(service) + } + + pub fn into_service(self) -> IntoService { + IntoService { + app: self.app, + poll_ready_error: None, + } + } +} + +pub struct Route { + service: H, + route_spec: RouteSpec, + fallback: F, + handler_ready: bool, + fallback_ready: bool, +} + +impl Clone for Route +where + H: Clone, + F: Clone, +{ + fn clone(&self) -> Self { + Self { + service: self.service.clone(), + fallback: self.fallback.clone(), + route_spec: self.route_spec.clone(), + // important to reset readiness when cloning + handler_ready: false, + fallback_ready: false, + } + } +} + +#[derive(Clone)] +struct RouteSpec { + method: Method, + spec: Bytes, +} + +impl RouteSpec { + fn matches(&self, req: &Request) -> bool { + // TODO(david): support dynamic placeholders like `/users/:id` + req.method() == self.method && req.uri().path().as_bytes() == self.spec + } +} + +impl Service> for Route +where + H: Service, Response = Response>, + H::Error: Into, + HB: http_body::Body + Send + Sync + 'static, + HB::Error: Into, + + F: Service, Response = Response>, + F::Error: Into, + FB: http_body::Body + Send + Sync + 'static, + FB::Error: Into, +{ + type Response = Response>; + type Error = Error; + type Future = future::Either, BoxResponseBody>; + + fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll> { + loop { + if !self.handler_ready { + ready!(self.service.poll_ready(cx)).map_err(Into::into)?; + self.handler_ready = true; + } + + if !self.fallback_ready { + ready!(self.fallback.poll_ready(cx)).map_err(Into::into)?; + self.fallback_ready = true; + } + + if self.handler_ready && self.fallback_ready { + return Poll::Ready(Ok(())); + } + } + } + + fn call(&mut self, req: Request) -> Self::Future { + if self.route_spec.matches(&req) { + assert!( + self.handler_ready, + "handler not ready. Did you forget to call `poll_ready`?" + ); + + self.handler_ready = false; + + future::Either::Left(BoxResponseBody(self.service.call(req))) + } else { + assert!( + self.fallback_ready, + "fallback not ready. Did you forget to call `poll_ready`?" + ); + + self.fallback_ready = false; + + // TODO(david): this leads to each route creating one box body, probably not great + future::Either::Right(BoxResponseBody(self.fallback.call(req))) + } + } +} + +#[pin_project] +pub struct BoxResponseBody(#[pin] F); + +impl Future for BoxResponseBody +where + F: Future, E>>, + E: Into, + B: http_body::Body + Send + Sync + 'static, + B::Error: Into, +{ + type Output = Result>, Error>; + + fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll { + let response: Response = ready!(self.project().0.poll(cx)).map_err(Into::into)?; + let response = response.map(|body| { + // TODO(david): attempt to downcast this into `Error` + let body = body.map_err(|err| Error::ResponseBody(err.into())); + BoxBody::new(body) + }); + Poll::Ready(Ok(response)) + } +}