Merge branch 'master' into master

.travis.yml

@@ -8,6 +8,7 @@ cache:
 
 matrix:
   include:
+    - rust: 1.31.0
     - rust: stable
     - rust: beta
     - rust: nightly-2019-03-02

Cargo.toml

@@ -113,3 +113,7 @@ flate2 = "1.0.2"
 lto = true
 opt-level = 3
 codegen-units = 1
+
+[patch.crates-io]
+actix = { git = "https://github.com/actix/actix.git" }
+actix-http = { path = "actix-http" }

actix-files/src/lib.rs

@@ -566,7 +566,10 @@ mod tests {
     use bytes::BytesMut;
 
     use super::*;
-    use actix_web::http::{header, header::DispositionType, Method, StatusCode};
+    use actix_web::http::header::{
+        self, ContentDisposition, DispositionParam, DispositionType,
+    };
+    use actix_web::http::{Method, StatusCode};
     use actix_web::test::{self, TestRequest};
     use actix_web::App;
 
@@ -683,7 +686,6 @@ mod tests {
 
     #[test]
     fn test_named_file_image_attachment() {
-        use header::{ContentDisposition, DispositionParam, DispositionType};
         let cd = ContentDisposition {
             disposition: DispositionType::Attachment,
             parameters: vec![DispositionParam::Filename(String::from("test.png"))],

actix-http/CHANGES.md

@@ -1,5 +1,12 @@
 # Changes
 
+## [0.1.0-alpha.3] - 2019-04-xx
+
+### Fixed
+
+* Rust 1.31.0 compatibility
+
+
 ## [0.1.0-alpha.2] - 2019-03-29
 
 ### Added

actix-http/src/cookie/mod.rs

@@ -59,7 +59,7 @@ mod parse;
 #[macro_use]
 mod secure;
 #[cfg(feature = "secure-cookies")]
-pub use secure::*;
+pub use self::secure::*;
 
 use std::borrow::Cow;
 use std::fmt;
@@ -68,11 +68,11 @@ use std::str::FromStr;
 use percent_encoding::{percent_encode, USERINFO_ENCODE_SET};
 use time::{Duration, Tm};
 
-pub use builder::CookieBuilder;
+pub use self::builder::CookieBuilder;
-pub use draft::*;
+pub use self::draft::*;
-pub use jar::{CookieJar, Delta, Iter};
+pub use self::jar::{CookieJar, Delta, Iter};
-use parse::parse_cookie;
+use self::parse::parse_cookie;
-pub use parse::ParseError;
+pub use self::parse::ParseError;
 
 #[derive(Debug, Clone)]
 enum CookieStr {

awc/CHANGES.md

@@ -1,5 +1,22 @@
 # Changes
 
+
+## [0.1.0-alpha.3] - 2019-04-xx
+
+### Added
+
+* Added `Deref<Target = RequestHead>` for `ClientRequest`.
+
+* Export `MessageBody` type
+
+* `ClientResponse::json()` - Loads and parse `application/json` encoded body
+
+
+### Changed
+
+* `ClientResponse::body()` does not consume response object.
+
+
 ## [0.1.0-alpha.2] - 2019-03-29
 
 ### Added

awc/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "awc"
-version = "0.1.0-alpha.2"
+version = "0.1.0-alpha.3"
 authors = ["Nikolay Kim <fafhrd91@gmail.com>"]
 description = "Actix http client."
 readme = "README.md"
@@ -44,6 +44,7 @@ bytes = "0.4"
 derive_more = "0.14"
 futures = "0.1.25"
 log =" 0.4"
+mime = "0.3"
 percent-encoding = "1.0"
 rand = "0.6"
 serde = "1.0"
@@ -62,6 +63,5 @@ actix-server = { version = "0.4.1", features=["ssl"] }
 brotli2 = { version="0.3.2" }
 flate2 = { version="1.0.2" }
 env_logger = "0.6"
-mime = "0.3"
 rand = "0.6"
 tokio-tcp = "0.1"

awc/src/error.rs

@@ -4,6 +4,9 @@ pub use actix_http::error::PayloadError;
 pub use actix_http::ws::HandshakeError as WsHandshakeError;
 pub use actix_http::ws::ProtocolError as WsProtocolError;
 
+use actix_http::{Response, ResponseError};
+use serde_json::error::Error as JsonError;
+
 use actix_http::http::{header::HeaderValue, Error as HttpError, StatusCode};
 use derive_more::{Display, From};
 
@@ -47,3 +50,27 @@ impl From<HttpError> for WsClientError {
         WsClientError::SendRequest(err.into())
     }
 }
+
+/// A set of errors that can occur during parsing json payloads
+#[derive(Debug, Display, From)]
+pub enum JsonPayloadError {
+    /// Payload size is bigger than allowed. (default: 32kB)
+    #[display(fmt = "Json payload size is bigger than allowed.")]
+    Overflow,
+    /// Content type error
+    #[display(fmt = "Content type error")]
+    ContentType,
+    /// Deserialize error
+    #[display(fmt = "Json deserialize error: {}", _0)]
+    Deserialize(JsonError),
+    /// Payload error
+    #[display(fmt = "Error that occur during reading payload: {}", _0)]
+    Payload(PayloadError),
+}
+
+/// Return `InternlaServerError` for `JsonPayloadError`
+impl ResponseError for JsonPayloadError {
+    fn error_response(&self) -> Response {
+        Response::new(StatusCode::INTERNAL_SERVER_ERROR)
+    }
+}

awc/src/lib.rs

@@ -39,7 +39,7 @@ pub mod ws;
 
 pub use self::builder::ClientBuilder;
 pub use self::request::ClientRequest;
-pub use self::response::ClientResponse;
+pub use self::response::{ClientResponse, JsonBody, MessageBody};
 
 use self::connect::{Connect, ConnectorWrapper};
 

awc/src/request.rs

@@ -554,6 +554,20 @@ impl ClientRequest {
     }
 }
 
+impl std::ops::Deref for ClientRequest {
+    type Target = RequestHead;
+
+    fn deref(&self) -> &RequestHead {
+        &self.head
+    }
+}
+
+impl std::ops::DerefMut for ClientRequest {
+    fn deref_mut(&mut self) -> &mut RequestHead {
+        &mut self.head
+    }
+}
+
 impl fmt::Debug for ClientRequest {
     fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
         writeln!(

awc/src/response.rs

@@ -4,13 +4,14 @@ use std::fmt;
 use bytes::{Bytes, BytesMut};
 use futures::{Future, Poll, Stream};
 
-use actix_http::error::PayloadError;
+use actix_http::cookie::Cookie;
+use actix_http::error::{CookieParseError, PayloadError};
 use actix_http::http::header::{CONTENT_LENGTH, SET_COOKIE};
 use actix_http::http::{HeaderMap, StatusCode, Version};
 use actix_http::{Extensions, HttpMessage, Payload, PayloadStream, ResponseHead};
+use serde::de::DeserializeOwned;
 
-use actix_http::cookie::Cookie;
+use crate::error::JsonPayloadError;
-use actix_http::error::CookieParseError;
 
 /// Client Response
 pub struct ClientResponse<S = PayloadStream> {
@@ -104,10 +105,21 @@ impl<S> ClientResponse<S>
 where
     S: Stream<Item = Bytes, Error = PayloadError> + 'static,
 {
-    /// Load http response's body.
+    /// Loads http response's body.
-    pub fn body(self) -> MessageBody<S> {
+    pub fn body(&mut self) -> MessageBody<S> {
         MessageBody::new(self)
     }
+
+    /// Loads and 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
+    pub fn json<T: DeserializeOwned>(&mut self) -> JsonBody<S, T> {
+        JsonBody::new(self)
+    }
 }
 
 impl<S> Stream for ClientResponse<S>
@@ -137,7 +149,7 @@ impl<S> fmt::Debug for ClientResponse<S> {
 pub struct MessageBody<S> {
     limit: usize,
     length: Option<usize>,
-    stream: Option<ClientResponse<S>>,
+    stream: Option<Payload<S>>,
     err: Option<PayloadError>,
     fut: Option<Box<Future<Item = Bytes, Error = PayloadError>>>,
 }
@@ -147,7 +159,7 @@ where
     S: Stream<Item = Bytes, Error = PayloadError> + 'static,
 {
     /// Create `MessageBody` for request.
-    pub fn new(res: ClientResponse<S>) -> MessageBody<S> {
+    pub fn new(res: &mut ClientResponse<S>) -> MessageBody<S> {
         let mut len = None;
         if let Some(l) = res.headers().get(CONTENT_LENGTH) {
             if let Ok(s) = l.to_str() {
@@ -164,7 +176,7 @@ where
         MessageBody {
             limit: 262_144,
             length: len,
-            stream: Some(res),
+            stream: Some(res.take_payload()),
             fut: None,
             err: None,
         }
@@ -230,28 +242,132 @@ where
     }
 }
 
+/// Response's payload json parser, it resolves to a deserialized `T` value.
+///
+/// Returns error:
+///
+/// * content type is not `application/json`
+/// * content length is greater than 64k
+pub struct JsonBody<S, U> {
+    limit: usize,
+    length: Option<usize>,
+    stream: Payload<S>,
+    err: Option<JsonPayloadError>,
+    fut: Option<Box<Future<Item = U, Error = JsonPayloadError>>>,
+}
+
+impl<S, U> JsonBody<S, U>
+where
+    S: Stream<Item = Bytes, Error = PayloadError> + 'static,
+    U: DeserializeOwned,
+{
+    /// Create `JsonBody` for request.
+    pub fn new(req: &mut ClientResponse<S>) -> Self {
+        // check content-type
+        let json = if let Ok(Some(mime)) = req.mime_type() {
+            mime.subtype() == mime::JSON || mime.suffix() == Some(mime::JSON)
+        } else {
+            false
+        };
+        if !json {
+            return JsonBody {
+                limit: 65536,
+                length: None,
+                stream: Payload::None,
+                fut: None,
+                err: Some(JsonPayloadError::ContentType),
+            };
+        }
+
+        let mut len = None;
+        if let Some(l) = req.headers().get(CONTENT_LENGTH) {
+            if let Ok(s) = l.to_str() {
+                if let Ok(l) = s.parse::<usize>() {
+                    len = Some(l)
+                }
+            }
+        }
+
+        JsonBody {
+            limit: 65536,
+            length: len,
+            stream: req.take_payload(),
+            fut: None,
+            err: None,
+        }
+    }
+
+    /// Change max size of payload. By default max size is 64Kb
+    pub fn limit(mut self, limit: usize) -> Self {
+        self.limit = limit;
+        self
+    }
+}
+
+impl<T, U> Future for JsonBody<T, U>
+where
+    T: Stream<Item = Bytes, Error = PayloadError> + 'static,
+    U: DeserializeOwned + 'static,
+{
+    type Item = U;
+    type Error = JsonPayloadError;
+
+    fn poll(&mut self) -> Poll<U, JsonPayloadError> {
+        if let Some(ref mut fut) = self.fut {
+            return fut.poll();
+        }
+
+        if let Some(err) = self.err.take() {
+            return Err(err);
+        }
+
+        let limit = self.limit;
+        if let Some(len) = self.length.take() {
+            if len > limit {
+                return Err(JsonPayloadError::Overflow);
+            }
+        }
+
+        let fut = std::mem::replace(&mut self.stream, Payload::None)
+            .from_err()
+            .fold(BytesMut::with_capacity(8192), 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::<U>(&body)?));
+        self.fut = Some(Box::new(fut));
+        self.poll()
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
     use futures::Async;
+    use serde::{Deserialize, Serialize};
 
-    use crate::{http::header, test::TestResponse};
+    use crate::{http::header, test::block_on, test::TestResponse};
 
     #[test]
     fn test_body() {
-        let req = TestResponse::with_header(header::CONTENT_LENGTH, "xxxx").finish();
+        let mut req = TestResponse::with_header(header::CONTENT_LENGTH, "xxxx").finish();
         match req.body().poll().err().unwrap() {
             PayloadError::UnknownLength => (),
             _ => unreachable!("error"),
         }
 
-        let req = TestResponse::with_header(header::CONTENT_LENGTH, "1000000").finish();
+        let mut req =
+            TestResponse::with_header(header::CONTENT_LENGTH, "1000000").finish();
         match req.body().poll().err().unwrap() {
             PayloadError::Overflow => (),
             _ => unreachable!("error"),
         }
 
-        let req = TestResponse::default()
+        let mut req = TestResponse::default()
             .set_payload(Bytes::from_static(b"test"))
             .finish();
         match req.body().poll().ok().unwrap() {
@@ -259,7 +375,7 @@ mod tests {
             _ => unreachable!("error"),
         }
 
-        let req = TestResponse::default()
+        let mut req = TestResponse::default()
             .set_payload(Bytes::from_static(b"11111111111111"))
             .finish();
         match req.body().limit(5).poll().err().unwrap() {
@@ -267,4 +383,73 @@ mod tests {
             _ => unreachable!("error"),
         }
     }
+
+    #[derive(Serialize, Deserialize, PartialEq, Debug)]
+    struct MyObject {
+        name: String,
+    }
+
+    fn json_eq(err: JsonPayloadError, other: JsonPayloadError) -> bool {
+        match err {
+            JsonPayloadError::Overflow => match other {
+                JsonPayloadError::Overflow => true,
+                _ => false,
+            },
+            JsonPayloadError::ContentType => match other {
+                JsonPayloadError::ContentType => true,
+                _ => false,
+            },
+            _ => false,
+        }
+    }
+
+    #[test]
+    fn test_json_body() {
+        let mut req = TestResponse::default().finish();
+        let json = block_on(JsonBody::<_, MyObject>::new(&mut req));
+        assert!(json_eq(json.err().unwrap(), JsonPayloadError::ContentType));
+
+        let mut req = TestResponse::default()
+            .header(
+                header::CONTENT_TYPE,
+                header::HeaderValue::from_static("application/text"),
+            )
+            .finish();
+        let json = block_on(JsonBody::<_, MyObject>::new(&mut req));
+        assert!(json_eq(json.err().unwrap(), JsonPayloadError::ContentType));
+
+        let mut req = TestResponse::default()
+            .header(
+                header::CONTENT_TYPE,
+                header::HeaderValue::from_static("application/json"),
+            )
+            .header(
+                header::CONTENT_LENGTH,
+                header::HeaderValue::from_static("10000"),
+            )
+            .finish();
+
+        let json = block_on(JsonBody::<_, MyObject>::new(&mut req).limit(100));
+        assert!(json_eq(json.err().unwrap(), JsonPayloadError::Overflow));
+
+        let mut req = TestResponse::default()
+            .header(
+                header::CONTENT_TYPE,
+                header::HeaderValue::from_static("application/json"),
+            )
+            .header(
+                header::CONTENT_LENGTH,
+                header::HeaderValue::from_static("16"),
+            )
+            .set_payload(Bytes::from_static(b"{\"name\": \"test\"}"))
+            .finish();
+
+        let json = block_on(JsonBody::<_, MyObject>::new(&mut req));
+        assert_eq!(
+            json.ok().unwrap(),
+            MyObject {
+                name: "test".to_owned()
+            }
+        );
+    }
 }

awc/src/test.rs

@@ -6,6 +6,8 @@ use actix_http::http::header::{self, Header, HeaderValue, IntoHeaderValue};
 use actix_http::http::{HeaderName, HttpTryFrom, Version};
 use actix_http::{h1, Payload, ResponseHead};
 use bytes::Bytes;
+#[cfg(test)]
+use futures::Future;
 use percent_encoding::{percent_encode, USERINFO_ENCODE_SET};
 
 use crate::ClientResponse;
@@ -18,7 +20,7 @@ thread_local! {
 }
 
 #[cfg(test)]
-pub fn run_on<F, R>(f: F) -> R
+pub(crate) fn run_on<F, R>(f: F) -> R
 where
     F: Fn() -> R,
 {
@@ -29,6 +31,14 @@ where
     .unwrap()
 }
 
+#[cfg(test)]
+pub(crate) fn block_on<F>(f: F) -> Result<F::Item, F::Error>
+where
+    F: Future,
+{
+    RT.with(move |rt| rt.borrow_mut().block_on(f))
+}
+
 /// Test `ClientResponse` builder
 pub struct TestResponse {
     head: ResponseHead,

src/error.rs

@@ -79,7 +79,7 @@ pub enum JsonPayloadError {
     Payload(PayloadError),
 }
 
-/// Return `BadRequest` for `UrlencodedError`
+/// Return `BadRequest` for `JsonPayloadError`
 impl ResponseError for JsonPayloadError {
     fn error_response(&self) -> HttpResponse {
         match *self {

src/lib.rs

@@ -62,7 +62,7 @@
 //! * SSL support with OpenSSL or `native-tls`
 //! * Middlewares (`Logger`, `Session`, `CORS`, `CSRF`, `DefaultHeaders`)
 //! * Supports [Actix actor framework](https://github.com/actix/actix)
-//! * Supported Rust version: 1.32 or later
+//! * Supported Rust version: 1.31 or later
 //!
 //! ## Package feature
 //!