Implement Reply for Result<impl Reply, impl Reply>

examples/error_handling.rs

@@ -0,0 +1,65 @@
+#![deny(warnings)]
+
+use std::num::NonZeroU16;
+
+use serde_derive::Serialize;
+use warp::http::StatusCode;
+use warp::reply::Response;
+use warp::{Filter, Reply};
+
+#[tokio::main]
+async fn main() {
+    let math = warp::path!("math" / u16)
+        .and(warp::header::<u16>("div-by"))
+        .map_async(div_by);
+
+    let routes = warp::get().and(math);
+
+    warp::serve(routes).run(([127, 0, 0, 1], 3030)).await;
+}
+
+// Reply is implemented for Result<impl Reply, impl Reply>.
+// This makes it easy to handle errors since bad requests are just mapped to Result::Err().
+async fn div_by(num: u16, denom: u16) -> Result<impl Reply, DivideByZero> {
+    let denom = NonZeroU16::new(denom).ok_or(DivideByZero)?;
+
+    Ok(warp::reply::json(&Math {
+        op: format!("{} / {}", num, denom),
+        output: num / denom.get(),
+    }))
+}
+
+// Error
+
+#[derive(Debug)]
+struct DivideByZero;
+
+// We have to implement Reply for our error types.
+impl Reply for DivideByZero {
+    fn into_response(self) -> Response {
+        let code = StatusCode::BAD_REQUEST;
+        let message = "DIVIDE_BY_ZERO";
+        let json = warp::reply::json(&ErrorMessage {
+            code: code.as_u16(),
+            message: message.into(),
+        });
+
+        warp::reply::with_status(json, code).into_response()
+    }
+}
+
+// JSON replies
+
+/// A successful math operation.
+#[derive(Serialize)]
+struct Math {
+    op: String,
+    output: u16,
+}
+
+/// An API error serializable to JSON.
+#[derive(Serialize)]
+struct ErrorMessage {
+    code: u16,
+    message: String,
+}

examples/todos.rs

@@ -53,7 +53,7 @@ mod filters {
             .and(warp::get())
             .and(warp::query::<ListOptions>())
             .and(with_db(db))
-            .and_then(handlers::list_todos)
+            .map_async(handlers::list_todos)
     }
 
     /// POST /todos with JSON body
@@ -64,7 +64,7 @@ mod filters {
             .and(warp::post())
             .and(json_body())
             .and(with_db(db))
-            .and_then(handlers::create_todo)
+            .map_async(handlers::create_todo)
     }
 
     /// PUT /todos/:id with JSON body
@@ -75,7 +75,7 @@ mod filters {
             .and(warp::put())
             .and(json_body())
             .and(with_db(db))
-            .and_then(handlers::update_todo)
+            .map_async(handlers::update_todo)
     }
 
     /// DELETE /todos/:id
@@ -93,7 +93,7 @@ mod filters {
             .and(admin_only)
             .and(warp::delete())
             .and(with_db(db))
-            .and_then(handlers::delete_todo)
+            .map_async(handlers::delete_todo)
     }
 
     fn with_db(db: Db) -> impl Filter<Extract = (Db,), Error = std::convert::Infallible> + Clone {
@@ -113,10 +113,9 @@ mod filters {
 /// No tuples are needed, it's auto flattened for the functions.
 mod handlers {
     use super::models::{Db, ListOptions, Todo};
-    use std::convert::Infallible;
     use warp::http::StatusCode;
 
-    pub async fn list_todos(opts: ListOptions, db: Db) -> Result<impl warp::Reply, Infallible> {
+    pub async fn list_todos(opts: ListOptions, db: Db) -> impl warp::Reply {
         // Just return a JSON array of todos, applying the limit and offset.
         let todos = db.lock().await;
         let todos: Vec<Todo> = todos
@@ -125,10 +124,10 @@ mod handlers {
             .skip(opts.offset.unwrap_or(0))
             .take(opts.limit.unwrap_or(std::usize::MAX))
             .collect();
-        Ok(warp::reply::json(&todos))
+        warp::reply::json(&todos)
     }
 
-    pub async fn create_todo(create: Todo, db: Db) -> Result<impl warp::Reply, Infallible> {
+    pub async fn create_todo(create: Todo, db: Db) -> impl warp::Reply {
         log::debug!("create_todo: {:?}", create);
 
         let mut vec = db.lock().await;
@@ -137,39 +136,35 @@ mod handlers {
             if todo.id == create.id {
                 log::debug!("    -> id already exists: {}", create.id);
                 // Todo with id already exists, return `400 BadRequest`.
-                return Ok(StatusCode::BAD_REQUEST);
+                return StatusCode::BAD_REQUEST;
             }
         }
 
         // No existing Todo with id, so insert and return `201 Created`.
         vec.push(create);
 
-        Ok(StatusCode::CREATED)
+        StatusCode::CREATED
     }
 
-    pub async fn update_todo(
-        id: u64,
-        update: Todo,
-        db: Db,
-    ) -> Result<impl warp::Reply, Infallible> {
+    pub async fn update_todo(id: u64, update: Todo, db: Db) -> impl warp::Reply {
         log::debug!("update_todo: id={}, todo={:?}", id, update);
         let mut vec = db.lock().await;
 
         // Look for the specified Todo...
         for todo in vec.iter_mut() {
             if todo.id == id {
                 *todo = update;
-                return Ok(StatusCode::OK);
+                return StatusCode::OK;
             }
         }
 
         log::debug!("    -> todo id not found!");
 
         // If the for loop didn't return OK, then the ID doesn't exist...
-        Ok(StatusCode::NOT_FOUND)
+        StatusCode::NOT_FOUND
     }
 
-    pub async fn delete_todo(id: u64, db: Db) -> Result<impl warp::Reply, Infallible> {
+    pub async fn delete_todo(id: u64, db: Db) -> impl warp::Reply {
         log::debug!("delete_todo: id={}", id);
 
         let mut vec = db.lock().await;
@@ -187,10 +182,10 @@ mod handlers {
         if deleted {
             // respond with a `204 No Content`, which means successful,
             // yet no body expected...
-            Ok(StatusCode::NO_CONTENT)
+            StatusCode::NO_CONTENT
         } else {
             log::debug!("    -> todo id not found!");
-            Ok(StatusCode::NOT_FOUND)
+            StatusCode::NOT_FOUND
         }
     }
 }

src/filter/map_async.rs

@@ -0,0 +1,88 @@
+use std::future::Future;
+use std::pin::Pin;
+use std::task::{Context, Poll};
+
+use futures::{ready, TryFuture};
+use pin_project::{pin_project, project};
+
+use super::{Filter, FilterBase, Func, Internal};
+
+#[derive(Clone, Copy, Debug)]
+pub struct MapAsync<T, F> {
+    pub(super) filter: T,
+    pub(super) callback: F,
+}
+
+impl<T, F> FilterBase for MapAsync<T, F>
+where
+    T: Filter,
+    F: Func<T::Extract> + Clone + Send,
+    F::Output: Future + Send,
+{
+    type Extract = (<F::Output as Future>::Output,);
+    type Error = T::Error;
+    type Future = MapAsyncFuture<T, F>;
+    #[inline]
+    fn filter(&self, _: Internal) -> Self::Future {
+        MapAsyncFuture {
+            state: State::First(self.filter.filter(Internal), self.callback.clone()),
+        }
+    }
+}
+
+#[allow(missing_debug_implementations)]
+#[pin_project]
+pub struct MapAsyncFuture<T: Filter, F>
+where
+    T: Filter,
+    F: Func<T::Extract>,
+    F::Output: Future + Send,
+{
+    #[pin]
+    state: State<T, F>,
+}
+
+#[pin_project]
+enum State<T, F>
+where
+    T: Filter,
+    F: Func<T::Extract>,
+    F::Output: Future + Send,
+{
+    First(#[pin] T::Future, F),
+    Second(#[pin] F::Output),
+    Done,
+}
+
+impl<T, F> Future for MapAsyncFuture<T, F>
+where
+    T: Filter,
+    F: Func<T::Extract>,
+    F::Output: Future + Send,
+{
+    type Output = Result<(<F::Output as Future>::Output,), T::Error>;
+
+    #[project]
+    fn poll(mut self: Pin<&mut Self>, cx: &mut Context) -> Poll<Self::Output> {
+        loop {
+            let pin = self.as_mut().project();
+            #[project]
+            let (ex1, second) = match pin.state.project() {
+                State::First(first, second) => match ready!(first.try_poll(cx)) {
+                    Ok(first) => (first, second),
+                    Err(err) => return Poll::Ready(Err(From::from(err))),
+                },
+                State::Second(second) => {
+                    let ex3 = ready!(second.poll(cx));
+                    self.set(MapAsyncFuture { state: State::Done });
+                    return Poll::Ready(Ok((ex3,)));
+                }
+                State::Done => panic!("polled after complete"),
+            };
+            let fut2 = second.call(ex1);
+            self.set(MapAsyncFuture {
+                state: State::Second(fut2),
+            });
+        }
+    }
+}

src/filter/mod.rs

@@ -2,6 +2,7 @@ mod and;
 mod and_then;
 mod boxed;
 mod map;
+mod map_async;
 mod map_err;
 mod or;
 mod or_else;
@@ -24,6 +25,7 @@ pub(crate) use self::and::And;
 use self::and_then::AndThen;
 pub use self::boxed::BoxedFilter;
 pub(crate) use self::map::Map;
+pub(crate) use self::map_async::MapAsync;
 pub(crate) use self::map_err::MapErr;
 pub(crate) use self::or::Or;
 use self::or_else::OrElse;
@@ -197,6 +199,19 @@ pub trait Filter: FilterBase {
         }
     }
 
+    /// TODO
+    fn map_async<F>(self, fun: F) -> MapAsync<Self, F>
+    where
+        Self: Sized,
+        F: Func<Self::Extract> + Clone,
+        F::Output: Future + Send,
+    {
+        MapAsync {
+            filter: self,
+            callback: fun,
+        }
+    }
+
     /// Composes this `Filter` with a function receiving the extracted value.
     ///
     /// The function should return some `TryFuture` type.

src/reply.rs

@@ -408,18 +408,24 @@ impl Reply for ::http::StatusCode {
     }
 }
 
-impl<T> Reply for Result<T, ::http::Error>
+impl Reply for ::http::Error {
+    #[inline]
+    fn into_response(self) -> Response {
+        log::error!("reply error: {:?}", self);
+        StatusCode::INTERNAL_SERVER_ERROR.into_response()
+    }
+}
+
+impl<T, E> Reply for Result<T, E>
 where
     T: Reply + Send,
+    E: Reply + Send,
 {
     #[inline]
     fn into_response(self) -> Response {
         match self {
             Ok(t) => t.into_response(),
-            Err(e) => {
-                log::error!("reply error: {:?}", e);
-                StatusCode::INTERNAL_SERVER_ERROR.into_response()
-            }
+            Err(e) => e.into_response(),
         }
     }
 }