added JsonBody future

src/error.rs

@@ -10,6 +10,7 @@ use std::error::Error as StdError;
 use cookie;
 use httparse;
 use failure::Fail;
+use futures::Canceled;
 use http2::Error as Http2Error;
 use http::{header, StatusCode, Error as HttpError};
 use http::uri::InvalidUriBytes;
@@ -110,6 +111,9 @@ impl ResponseError for io::Error {
 /// `InternalServerError` for `InvalidHeaderValue`
 impl ResponseError for header::InvalidHeaderValue {}
 
+/// `InternalServerError` for `futures::Canceled`
+impl ResponseError for Canceled {}
+
 /// Internal error
 #[derive(Fail, Debug)]
 #[fail(display="Unexpected task frame")]
@@ -393,6 +397,43 @@ impl From<PayloadError> for UrlencodedError {
     }
 }
 
+/// A set of errors that can occur during parsing json payloads
+#[derive(Fail, Debug)]
+pub enum JsonPayloadError {
+    /// Payload size is bigger than 256k
+    #[fail(display="Payload size is bigger than 256k")]
+    Overflow,
+    /// Content type error
+    #[fail(display="Content type error")]
+    ContentType,
+    /// Deserialize error
+    #[fail(display="Json deserialize error")]
+    Deserialize(JsonError),
+    /// Payload error
+    #[fail(display="Error that occur during reading payload")]
+    Payload(PayloadError),
+}
+
+/// Return `BadRequest` for `UrlencodedError`
+impl ResponseError for JsonPayloadError {
+
+    fn error_response(&self) -> HttpResponse {
+        HttpResponse::new(StatusCode::BAD_REQUEST, Body::Empty)
+    }
+}
+
+impl From<PayloadError> for JsonPayloadError {
+    fn from(err: PayloadError) -> JsonPayloadError {
+        JsonPayloadError::Payload(err)
+    }
+}
+
+impl From<JsonError> for JsonPayloadError {
+    fn from(err: JsonError) -> JsonPayloadError {
+        JsonPayloadError::Deserialize(err)
+    }
+}
+
 /// Errors which can occur when attempting to interpret a segment string as a
 /// valid path segment.
 #[derive(Fail, Debug, PartialEq)]

src/handler.rs

@@ -36,6 +36,21 @@ pub trait Responder {
     fn respond_to(self, req: HttpRequest) -> Result<Self::Item, Self::Error>;
 }
 
+/// Convinience trait that convert `Future` object into `Boxed` future
+pub trait AsyncResponder<I, E>: Sized {
+    fn responder(self) -> Box<Future<Item=I, Error=E>>;
+}
+
+impl<F, I, E> AsyncResponder<I, E> for F
+    where F: Future<Item=I, Error=E> + 'static,
+          I: Responder + 'static,
+          E: Into<Error> + 'static,
+{
+    fn responder(self) -> Box<Future<Item=I, Error=E>> {
+        Box::new(self)
+    }
+}
+
 /// Handler<S> for Fn()
 impl<F, R, S> Handler<S> for F
     where F: Fn(HttpRequest<S>) -> R + 'static,

src/httprequest.rs

@@ -5,9 +5,10 @@ use std::rc::Rc;
 use std::net::SocketAddr;
 use std::collections::HashMap;
 use bytes::BytesMut;
-use futures::{Async, Future, Stream, Poll};
 use cookie::Cookie;
+use futures::{Async, Future, Stream, Poll};
 use http_range::HttpRange;
+use serde::de::DeserializeOwned;
 use url::{Url, form_urlencoded};
 use http::{header, Uri, Method, Version, HeaderMap, Extensions};
 
@@ -15,6 +16,7 @@ use info::ConnectionInfo;
 use param::Params;
 use router::Router;
 use payload::Payload;
+use json::JsonBody;
 use multipart::Multipart;
 use helpers::SharedHttpMessage;
 use error::{ParseError, UrlGenerationError, CookieParseError, HttpRangeError, UrlencodedError};
@@ -468,6 +470,40 @@ impl<S> HttpRequest<S> {
     pub fn urlencoded(&mut self) -> UrlEncoded {
         UrlEncoded::from_request(self)
     }
+
+    /// Parse `application/json` encoded body.
+    /// Return `JsonBody<T>` future. It resolves to a `T` value.
+    ///
+    /// Returns error:
+    ///
+    /// * content type is not `application/json`
+    /// * content length is greater than 256k
+    ///
+    /// ```rust
+    /// # extern crate actix_web;
+    /// # extern crate futures;
+    /// # #[macro_use] extern crate serde_derive;
+    /// use actix_web::*;
+    /// use futures::future::{Future, ok};
+    ///
+    /// #[derive(Deserialize, Debug)]
+    /// struct MyObj {
+    ///     name: String,
+    /// }
+    ///
+    /// fn index(mut req: HttpRequest) -> Box<Future<Item=HttpResponse, Error=Error>> {
+    ///     req.json()                   // <- get JsonBody future
+    ///        .from_err()
+    ///        .and_then(|val: MyObj| {  // <- deserialized value
+    ///            println!("==== BODY ==== {:?}", val);
+    ///            Ok(httpcodes::HTTPOk.response())
+    ///        }).responder()
+    /// }
+    /// # fn main() {}
+    /// ```
+    pub fn json<T: DeserializeOwned>(&mut self) -> JsonBody<S, T> {
+        JsonBody::from_request(self)
+    }
 }
 
 impl Default for HttpRequest<()> {

src/json.rs

@@ -1,7 +1,12 @@
+use bytes::BytesMut;
+use futures::{Poll, Future, Stream};
+use http::header::CONTENT_LENGTH;
+
 use serde_json;
 use serde::Serialize;
+use serde::de::DeserializeOwned;
 
-use error::Error;
+use error::{Error, JsonPayloadError};
 use handler::Responder;
 use httprequest::HttpRequest;
 use httpresponse::HttpResponse;
@@ -42,22 +47,168 @@ impl<T: Serialize> Responder for Json<T> {
     }
 }
 
+/// Request payload json parser that resolves to a deserialized `T` value.
+///
+/// Returns error:
+///
+/// * content type is not `application/json`
+/// * content length is greater than 256k
+///
+/// ```rust
+/// # extern crate actix_web;
+/// # extern crate futures;
+/// # #[macro_use] extern crate serde_derive;
+/// use actix_web::*;
+/// use futures::future::Future;
+///
+/// #[derive(Deserialize, Debug)]
+/// struct MyObj {
+///     name: String,
+/// }
+///
+/// fn index(mut req: HttpRequest) -> Box<Future<Item=HttpResponse, Error=Error>> {
+///     req.json()                   // <- get JsonBody future
+///        .from_err()
+///        .and_then(|val: MyObj| {  // <- deserialized value
+///            println!("==== BODY ==== {:?}", val);
+///            Ok(httpcodes::HTTPOk.response())
+///        }).responder()
+/// }
+/// # fn main() {}
+/// ```
+pub struct JsonBody<S, T: DeserializeOwned>{
+    limit: usize,
+    ct: &'static str,
+    req: Option<HttpRequest<S>>,
+    fut: Option<Box<Future<Item=T, Error=JsonPayloadError>>>,
+}
+
+impl<S, T: DeserializeOwned> JsonBody<S, T> {
+
+    pub fn from_request(req: &mut HttpRequest<S>) -> Self {
+        JsonBody{
+            limit: 262_144,
+            req: Some(req.clone()),
+            fut: None,
+            ct: "application/json",
+        }
+    }
+
+    /// Change max size of payload. By default max size is 256Kb
+    pub fn limit(mut self, limit: usize) -> Self {
+        self.limit = limit;
+        self
+    }
+
+    /// Set allowed content type.
+    ///
+    /// By default *application/json* content type is used. Set content type
+    /// to empty string if you want to disable content type check.
+    pub fn content_type(mut self, ct: &'static str) -> Self {
+        self.ct = ct;
+        self
+    }
+}
+
+impl<S, T: DeserializeOwned + 'static> Future for JsonBody<S, T> {
+    type Item = T;
+    type Error = JsonPayloadError;
+
+    fn poll(&mut self) -> Poll<T, JsonPayloadError> {
+        if let Some(mut req) = self.req.take() {
+            if let Some(len) = req.headers().get(CONTENT_LENGTH) {
+                if let Ok(s) = len.to_str() {
+                    if let Ok(len) = s.parse::<usize>() {
+                        if len > self.limit {
+                            return Err(JsonPayloadError::Overflow);
+                        }
+                    } else {
+                        return Err(JsonPayloadError::Overflow);
+                    }
+                }
+            }
+            // check content-type
+            if !self.ct.is_empty() && req.content_type() != self.ct {
+                return Err(JsonPayloadError::ContentType)
+            }
+
+            let limit = self.limit;
+            let fut = req.payload_mut().readany()
+                .from_err()
+                .fold(BytesMut::new(), move |mut body, chunk| {
+                    if (body.len() + chunk.len()) > limit {
+                        Err(JsonPayloadError::Overflow)
+                    } else {
+                        body.extend_from_slice(&chunk);
+                        Ok(body)
+                    }
+                })
+                .and_then(|body| Ok(serde_json::from_slice::<T>(&body)?));
+            self.fut = Some(Box::new(fut));
+        }
+
+        self.fut.as_mut().expect("JsonBody could not be used second time").poll()
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
-    use http::{header, Method};
-    use application::Application;
+    use bytes::Bytes;
+    use http::header;
+    use futures::Async;
 
-    #[derive(Serialize)]
-    struct MyObj {
-        name: &'static str,
+    impl PartialEq for JsonPayloadError {
+        fn eq(&self, other: &JsonPayloadError) -> bool {
+            match *self {
+                JsonPayloadError::Overflow => match *other {
+                    JsonPayloadError::Overflow => true,
+                    _ => false,
+                },
+                JsonPayloadError::ContentType => match *other {
+                    JsonPayloadError::ContentType => true,
+                    _ => false,
+                },
+                _ => false,
+            }
+        }
+    }
+
+    #[derive(Serialize, Deserialize, PartialEq, Debug)]
+    struct MyObject {
+        name: String,
     }
 
     #[test]
     fn test_json() {
-        let json = Json(MyObj{name: "test"});
+        let json = Json(MyObject{name: "test".to_owned()});
         let resp = json.respond_to(HttpRequest::default()).unwrap();
         assert_eq!(resp.headers().get(header::CONTENT_TYPE).unwrap(), "application/json");
     }
 
+    #[test]
+    fn test_json_body() {
+        let mut req = HttpRequest::default();
+        let mut json = req.json::<MyObject>();
+        assert_eq!(json.poll().err().unwrap(), JsonPayloadError::ContentType);
+
+        let mut json = req.json::<MyObject>().content_type("text/json");
+        req.headers_mut().insert(header::CONTENT_TYPE,
+                                 header::HeaderValue::from_static("application/json"));
+        assert_eq!(json.poll().err().unwrap(), JsonPayloadError::ContentType);
+
+        let mut json = req.json::<MyObject>().limit(100);
+        req.headers_mut().insert(header::CONTENT_TYPE,
+                                 header::HeaderValue::from_static("application/json"));
+        req.headers_mut().insert(header::CONTENT_LENGTH,
+                                 header::HeaderValue::from_static("10000"));
+        assert_eq!(json.poll().err().unwrap(), JsonPayloadError::Overflow);
+
+        req.headers_mut().insert(header::CONTENT_LENGTH,
+                                 header::HeaderValue::from_static("16"));
+        req.payload_mut().unread_data(Bytes::from_static(b"{\"name\": \"test\"}"));
+        let mut json = req.json::<MyObject>();
+        assert_eq!(json.poll().ok().unwrap(), Async::Ready(MyObject{name: "test".to_owned()}));
+    }
+
 }

src/lib.rs

@@ -124,7 +124,7 @@ pub use json::{Json};
 pub use application::Application;
 pub use httprequest::HttpRequest;
 pub use httpresponse::HttpResponse;
-pub use handler::{Reply, Responder, NormalizePath};
+pub use handler::{Reply, Responder, NormalizePath, AsyncResponder};
 pub use route::Route;
 pub use resource::Resource;
 pub use server::HttpServer;
@@ -166,6 +166,7 @@ pub mod dev {
     pub use body::BodyStream;
     pub use info::ConnectionInfo;
     pub use handler::Handler;
+    pub use json::JsonBody;
     pub use router::{Router, Pattern};
     pub use pipeline::Pipeline;
     pub use channel::{HttpChannel, HttpHandler, IntoHttpHandler};

src/payload.rs

@@ -433,7 +433,7 @@ impl Inner {
 
     fn unread_data(&mut self, data: Bytes) {
         self.len += data.len();
-        self.items.push_front(data)
+        self.items.push_front(data);
     }
 
     fn capacity(&self) -> usize {