update version

CHANGES.md

@@ -1,5 +1,17 @@
 # Changes
 
+### Added
+
+* Add helper functions for reading response body `test::read_body()`
+
+* Added support for `remainder match` (i.e "/path/{tail}*")
+
+
+### Changed
+
+* `.to_async()` handler can return `Responder` type #792
+
+
 ## [1.0.0-beta.1] - 2019-04-20
 
 ### Added

Cargo.toml

@@ -68,7 +68,7 @@ rust-tls = ["rustls", "actix-server/rust-tls"]
 actix-codec = "0.1.2"
 actix-service = "0.3.6"
 actix-utils = "0.3.4"
-actix-router = "0.1.2"
+actix-router = "0.1.3"
 actix-rt = "0.2.2"
 actix-web-codegen = "0.1.0-beta.1"
 actix-http = { version = "0.1.1", features=["fail"] }

actix-files/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "actix-files"
-version = "0.1.0-betsa.1"
+version = "0.1.0-beta.1"
 authors = ["Nikolay Kim <fafhrd91@gmail.com>"]
 description = "Static files support for actix web."
 readme = "README.md"

actix-http/CHANGES.md

@@ -1,5 +1,12 @@
 # Changes
 
+## [0.1.2] - 2019-04-23
+
+### Fixed
+
+* Fix BorrowMutError panic in client connector #793
+
+
 ## [0.1.1] - 2019-04-19
 
 ### Changes

actix-http/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "actix-http"
-version = "0.1.1"
+version = "0.1.2"
 authors = ["Nikolay Kim <fafhrd91@gmail.com>"]
 description = "Actix http primitives"
 readme = "README.md"

actix-http/src/client/pool.rs

@@ -113,13 +113,23 @@ where
         match self.1.as_ref().borrow_mut().acquire(&key) {
             Acquire::Acquired(io, created) => {
                 // use existing connection
-                Either::A(ok(IoConnection::new(
+                return Either::A(ok(IoConnection::new(
                     io,
                     created,
                     Some(Acquired(key, Some(self.1.clone()))),
-                )))
+                )));
             }
-            Acquire::NotAvailable => {
+            Acquire::Available => {
+                // open new connection
+                return Either::B(Either::B(OpenConnection::new(
+                    key,
+                    self.1.clone(),
+                    self.0.call(req),
+                )));
+            }
+            _ => (),
+        }
+
         // connection is not available, wait
         let (rx, token) = self.1.as_ref().borrow_mut().wait_for(req);
         Either::B(Either::A(WaitForConnection {
@@ -129,16 +139,6 @@ where
             inner: Some(self.1.clone()),
         }))
     }
-            Acquire::Available => {
-                // open new connection
-                Either::B(Either::B(OpenConnection::new(
-                    key,
-                    self.1.clone(),
-                    self.0.call(req),
-                )))
-            }
-        }
-    }
 }
 
 #[doc(hidden)]

actix-multipart/CHANGES.md

@@ -1,7 +1,9 @@
 # Changes
 
-## [0.1.0-alpha.1] - 2019-04-xx
+## [0.1.0-beta.1] - 2019-04-21
 
 * Do not support nested multipart
 
 * Split multipart support to separate crate
+
+* Optimize multipart handling #634, #769

actix-multipart/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "actix-multipart"
-version = "0.1.0-alpha.1"
+version = "0.1.0-beta.1"
 authors = ["Nikolay Kim <fafhrd91@gmail.com>"]
 description = "Multipart support for actix web framework."
 readme = "README.md"
@@ -18,8 +18,8 @@ name = "actix_multipart"
 path = "src/lib.rs"
 
 [dependencies]
-actix-web = "1.0.0-alpha.6"
-actix-service = "0.3.4"
+actix-web = "1.0.0-beta.1"
+actix-service = "0.3.6"
 bytes = "0.4"
 derive_more = "0.14"
 httparse = "1.3"
@@ -31,4 +31,4 @@ twoway = "0.2"
 
 [dev-dependencies]
 actix-rt = "0.2.2"
-actix-http = "0.1.0"
+actix-http = "0.1.1"

actix-multipart/src/server.rs

@@ -168,7 +168,7 @@ impl InnerMultipart {
         match payload.readline() {
             None => {
                 if payload.eof {
-                    Err(MultipartError::Incomplete)
+                    Ok(Some(true))
                 } else {
                     Ok(None)
                 }
@@ -201,8 +201,7 @@ impl InnerMultipart {
             match payload.readline() {
                 Some(chunk) => {
                     if chunk.is_empty() {
-                        //ValueError("Could not find starting boundary %r"
-                        //% (self._boundary))
+                        return Err(MultipartError::Boundary);
                     }
                     if chunk.len() < boundary.len() {
                         continue;
@@ -505,47 +504,72 @@ impl InnerField {
         payload: &mut PayloadBuffer,
         boundary: &str,
     ) -> Poll<Option<Bytes>, MultipartError> {
-        match payload.read_until(b"\r") {
-            None => {
-                if payload.eof {
-                    Err(MultipartError::Incomplete)
+        let mut pos = 0;
+
+        let len = payload.buf.len();
+        if len == 0 {
+            return Ok(Async::NotReady);
+        }
+
+        // check boundary
+        if len > 4 && payload.buf[0] == b'\r' {
+            let b_len = if &payload.buf[..2] == b"\r\n" && &payload.buf[2..4] == b"--" {
+                Some(4)
+            } else if &payload.buf[1..3] == b"--" {
+                Some(3)
+            } else {
+                None
+            };
+
+            if let Some(b_len) = b_len {
+                let b_size = boundary.len() + b_len;
+                if len < b_size {
+                    return Ok(Async::NotReady);
+                } else {
+                    if &payload.buf[b_len..b_size] == boundary.as_bytes() {
+                        // found boundary
+                        return Ok(Async::Ready(None));
+                    } else {
+                        pos = b_size;
+                    }
+                }
+            }
+        }
+
+        loop {
+            return if let Some(idx) = twoway::find_bytes(&payload.buf[pos..], b"\r") {
+                let cur = pos + idx;
+
+                // check if we have enough data for boundary detection
+                if cur + 4 > len {
+                    if cur > 0 {
+                        Ok(Async::Ready(Some(payload.buf.split_to(cur).freeze())))
                     } else {
                         Ok(Async::NotReady)
                     }
-            }
-            Some(mut chunk) => {
-                if chunk.len() == 1 {
-                    payload.unprocessed(chunk);
-                    match payload.read_exact(boundary.len() + 4) {
-                        None => {
-                            if payload.eof {
-                                Err(MultipartError::Incomplete)
                 } else {
-                                Ok(Async::NotReady)
-                            }
-                        }
-                        Some(mut chunk) => {
-                            if &chunk[..2] == b"\r\n"
-                                && &chunk[2..4] == b"--"
-                                && &chunk[4..] == boundary.as_bytes()
+                    // check boundary
+                    if (&payload.buf[cur..cur + 2] == b"\r\n"
+                        && &payload.buf[cur + 2..cur + 4] == b"--")
+                        || (&payload.buf[cur..cur + 1] == b"\r"
+                            && &payload.buf[cur + 1..cur + 3] == b"--")
                     {
-                                payload.unprocessed(chunk);
-                                Ok(Async::Ready(None))
+                        if cur != 0 {
+                            // return buffer
+                            Ok(Async::Ready(Some(payload.buf.split_to(cur).freeze())))
                         } else {
-                                // \r might be part of data stream
-                                let ch = chunk.split_to(1);
-                                payload.unprocessed(chunk);
-                                Ok(Async::Ready(Some(ch)))
+                            pos = cur + 1;
+                            continue;
                         }
+                    } else {
+                        // not boundary
+                        pos = cur + 1;
+                        continue;
                     }
                 }
             } else {
-                    let to = chunk.len() - 1;
-                    let ch = chunk.split_to(to);
-                    payload.unprocessed(chunk);
-                    Ok(Async::Ready(Some(ch)))
-                }
-            }
+                return Ok(Async::Ready(Some(payload.buf.take().freeze())));
+            };
         }
     }
 
@@ -555,6 +579,7 @@ impl InnerField {
         }
 
         let result = if let Some(payload) = self.payload.as_ref().unwrap().get_mut(s) {
+            if !self.eof {
                 let res = if let Some(ref mut len) = self.length {
                     InnerField::read_len(payload, len)?
                 } else {
@@ -562,10 +587,12 @@ impl InnerField {
                 };
 
                 match res {
-                Async::NotReady => Async::NotReady,
-                Async::Ready(Some(bytes)) => Async::Ready(Some(bytes)),
-                Async::Ready(None) => {
-                    self.eof = true;
+                    Async::NotReady => return Ok(Async::NotReady),
+                    Async::Ready(Some(bytes)) => return Ok(Async::Ready(Some(bytes))),
+                    Async::Ready(None) => self.eof = true,
+                }
+            }
+
             match payload.readline() {
                 None => Async::Ready(None),
                 Some(line) => {
@@ -575,8 +602,6 @@ impl InnerField {
                     Async::Ready(None)
                 }
             }
-                }
-            }
         } else {
             Async::NotReady
         };
@@ -704,7 +729,7 @@ impl PayloadBuffer {
     }
 
     /// Read exact number of bytes
-    #[inline]
+    #[cfg(test)]
     fn read_exact(&mut self, size: usize) -> Option<Bytes> {
         if size <= self.buf.len() {
             Some(self.buf.split_to(size).freeze())
@@ -875,6 +900,78 @@ mod tests {
         });
     }
 
+    #[test]
+    fn test_stream() {
+        run_on(|| {
+            let (sender, payload) = create_stream();
+
+            let bytes = Bytes::from(
+                "testasdadsad\r\n\
+                 --abbc761f78ff4d7cb7573b5a23f96ef0\r\n\
+                 Content-Disposition: form-data; name=\"file\"; filename=\"fn.txt\"\r\n\
+                 Content-Type: text/plain; charset=utf-8\r\n\r\n\
+                 test\r\n\
+                 --abbc761f78ff4d7cb7573b5a23f96ef0\r\n\
+                 Content-Type: text/plain; charset=utf-8\r\n\r\n\
+                 data\r\n\
+                 --abbc761f78ff4d7cb7573b5a23f96ef0--\r\n",
+            );
+            sender.unbounded_send(Ok(bytes)).unwrap();
+
+            let mut headers = HeaderMap::new();
+            headers.insert(
+                header::CONTENT_TYPE,
+                header::HeaderValue::from_static(
+                    "multipart/mixed; boundary=\"abbc761f78ff4d7cb7573b5a23f96ef0\"",
+                ),
+            );
+
+            let mut multipart = Multipart::new(&headers, payload);
+            match multipart.poll().unwrap() {
+                Async::Ready(Some(mut field)) => {
+                    let cd = field.content_disposition().unwrap();
+                    assert_eq!(cd.disposition, DispositionType::FormData);
+                    assert_eq!(cd.parameters[0], DispositionParam::Name("file".into()));
+
+                    assert_eq!(field.content_type().type_(), mime::TEXT);
+                    assert_eq!(field.content_type().subtype(), mime::PLAIN);
+
+                    match field.poll().unwrap() {
+                        Async::Ready(Some(chunk)) => assert_eq!(chunk, "test"),
+                        _ => unreachable!(),
+                    }
+                    match field.poll().unwrap() {
+                        Async::Ready(None) => (),
+                        _ => unreachable!(),
+                    }
+                }
+                _ => unreachable!(),
+            }
+
+            match multipart.poll().unwrap() {
+                Async::Ready(Some(mut field)) => {
+                    assert_eq!(field.content_type().type_(), mime::TEXT);
+                    assert_eq!(field.content_type().subtype(), mime::PLAIN);
+
+                    match field.poll() {
+                        Ok(Async::Ready(Some(chunk))) => assert_eq!(chunk, "data"),
+                        _ => unreachable!(),
+                    }
+                    match field.poll() {
+                        Ok(Async::Ready(None)) => (),
+                        _ => unreachable!(),
+                    }
+                }
+                _ => unreachable!(),
+            }
+
+            match multipart.poll().unwrap() {
+                Async::Ready(None) => (),
+                _ => unreachable!(),
+            }
+        });
+    }
+
     #[test]
     fn test_basic() {
         run_on(|| {

actix-web-actors/Cargo.toml

@@ -19,8 +19,8 @@ path = "src/lib.rs"
 
 [dependencies]
 actix = "0.8.0"
-actix-web = "1.0.0-alpha.5"
-actix-http = "0.1.0"
+actix-web = "1.0.0-beta.1"
+actix-http = "0.1.1"
 actix-codec = "0.1.2"
 bytes = "0.4"
 futures = "0.1.25"

src/handler.rs

@@ -124,7 +124,7 @@ where
 pub trait AsyncFactory<T, R>: Clone + 'static
 where
     R: IntoFuture,
-    R::Item: Into<Response>,
+    R::Item: Responder,
     R::Error: Into<Error>,
 {
     fn call(&self, param: T) -> R;
@@ -134,7 +134,7 @@ impl<F, R> AsyncFactory<(), R> for F
 where
     F: Fn() -> R + Clone + 'static,
     R: IntoFuture,
-    R::Item: Into<Response>,
+    R::Item: Responder,
     R::Error: Into<Error>,
 {
     fn call(&self, _: ()) -> R {
@@ -147,7 +147,7 @@ pub struct AsyncHandler<F, T, R>
 where
     F: AsyncFactory<T, R>,
     R: IntoFuture,
-    R::Item: Into<Response>,
+    R::Item: Responder,
     R::Error: Into<Error>,
 {
     hnd: F,
@@ -158,7 +158,7 @@ impl<F, T, R> AsyncHandler<F, T, R>
 where
     F: AsyncFactory<T, R>,
     R: IntoFuture,
-    R::Item: Into<Response>,
+    R::Item: Responder,
     R::Error: Into<Error>,
 {
     pub fn new(hnd: F) -> Self {
@@ -173,7 +173,7 @@ impl<F, T, R> Clone for AsyncHandler<F, T, R>
 where
     F: AsyncFactory<T, R>,
     R: IntoFuture,
-    R::Item: Into<Response>,
+    R::Item: Responder,
     R::Error: Into<Error>,
 {
     fn clone(&self) -> Self {
@@ -188,7 +188,7 @@ impl<F, T, R> Service for AsyncHandler<F, T, R>
 where
     F: AsyncFactory<T, R>,
     R: IntoFuture,
-    R::Item: Into<Response>,
+    R::Item: Responder,
     R::Error: Into<Error>,
 {
     type Request = (T, HttpRequest);
@@ -203,31 +203,38 @@ where
     fn call(&mut self, (param, req): (T, HttpRequest)) -> Self::Future {
         AsyncHandlerServiceResponse {
             fut: self.hnd.call(param).into_future(),
+            fut2: None,
             req: Some(req),
         }
     }
 }
 
 #[doc(hidden)]
-pub struct AsyncHandlerServiceResponse<T> {
+pub struct AsyncHandlerServiceResponse<T>
+where
+    T: Future,
+    T::Item: Responder,
+{
     fut: T,
+    fut2: Option<<<T::Item as Responder>::Future as IntoFuture>::Future>,
     req: Option<HttpRequest>,
 }
 
 impl<T> Future for AsyncHandlerServiceResponse<T>
 where
     T: Future,
-    T::Item: Into<Response>,
+    T::Item: Responder,
     T::Error: Into<Error>,
 {
     type Item = ServiceResponse;
     type Error = Void;
 
     fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
-        match self.fut.poll() {
+        if let Some(ref mut fut) = self.fut2 {
+            return match fut.poll() {
                 Ok(Async::Ready(res)) => Ok(Async::Ready(ServiceResponse::new(
                     self.req.take().unwrap(),
-                res.into(),
+                    res,
                 ))),
                 Ok(Async::NotReady) => Ok(Async::NotReady),
                 Err(e) => {
@@ -237,6 +244,23 @@ where
                         res,
                     )))
                 }
+            };
+        }
+
+        match self.fut.poll() {
+            Ok(Async::Ready(res)) => {
+                self.fut2 =
+                    Some(res.respond_to(self.req.as_ref().unwrap()).into_future());
+                return self.poll();
+            }
+            Ok(Async::NotReady) => Ok(Async::NotReady),
+            Err(e) => {
+                let res: Response = e.into().into();
+                Ok(Async::Ready(ServiceResponse::new(
+                    self.req.take().unwrap(),
+                    res,
+                )))
+            }
         }
     }
 }
@@ -357,7 +381,7 @@ macro_rules! factory_tuple ({ $(($n:tt, $T:ident)),+} => {
     impl<Func, $($T,)+ Res> AsyncFactory<($($T,)+), Res> for Func
     where Func: Fn($($T,)+) -> Res + Clone + 'static,
           Res: IntoFuture,
-          Res::Item: Into<Response>,
+          Res::Item: Responder,
           Res::Error: Into<Error>,
     {
         fn call(&self, param: ($($T,)+)) -> Res {

src/resource.rs

@@ -217,7 +217,7 @@ where
         F: AsyncFactory<I, R>,
         I: FromRequest + 'static,
         R: IntoFuture + 'static,
-        R::Item: Into<Response>,
+        R::Item: Responder,
         R::Error: Into<Error>,
     {
         self.routes.push(Route::new().to_async(handler));

src/route.rs

@@ -1,7 +1,7 @@
 use std::cell::RefCell;
 use std::rc::Rc;
 
-use actix_http::{http::Method, Error, Extensions, Response};
+use actix_http::{http::Method, Error, Extensions};
 use actix_service::{NewService, Service};
 use futures::future::{ok, Either, FutureResult};
 use futures::{Async, Future, IntoFuture, Poll};
@@ -278,7 +278,7 @@ impl Route {
         F: AsyncFactory<T, R>,
         T: FromRequest + 'static,
         R: IntoFuture + 'static,
-        R::Item: Into<Response>,
+        R::Item: Responder,
         R::Error: Into<Error>,
     {
         self.service = Box::new(RouteNewService::new(Extract::new(
@@ -418,18 +418,25 @@ where
 mod tests {
     use std::time::Duration;
 
+    use bytes::Bytes;
     use futures::Future;
+    use serde_derive::Serialize;
     use tokio_timer::sleep;
 
     use crate::http::{Method, StatusCode};
-    use crate::test::{call_service, init_service, TestRequest};
+    use crate::test::{call_service, init_service, read_body, TestRequest};
     use crate::{error, web, App, HttpResponse};
 
+    #[derive(Serialize, PartialEq, Debug)]
+    struct MyObject {
+        name: String,
+    }
+
     #[test]
     fn test_route() {
-        let mut srv =
-            init_service(
-                App::new().service(
+        let mut srv = init_service(
+            App::new()
+                .service(
                     web::resource("/test")
                         .route(web::get().to(|| HttpResponse::Ok()))
                         .route(web::put().to(|| {
@@ -444,7 +451,14 @@ mod tests {
                                 Err::<HttpResponse, _>(error::ErrorBadRequest("err"))
                             })
                         })),
-                ),
+                )
+                .service(web::resource("/json").route(web::get().to_async(|| {
+                    sleep(Duration::from_millis(25)).then(|_| {
+                        Ok::<_, crate::Error>(web::Json(MyObject {
+                            name: "test".to_string(),
+                        }))
+                    })
+                }))),
         );
 
         let req = TestRequest::with_uri("/test")
@@ -476,5 +490,12 @@ mod tests {
             .to_request();
         let resp = call_service(&mut srv, req);
         assert_eq!(resp.status(), StatusCode::METHOD_NOT_ALLOWED);
+
+        let req = TestRequest::with_uri("/json").to_request();
+        let resp = call_service(&mut srv, req);
+        assert_eq!(resp.status(), StatusCode::OK);
+
+        let body = read_body(resp);
+        assert_eq!(body, Bytes::from_static(b"{\"name\":\"test\"}"));
     }
 }

src/test.rs

@@ -193,6 +193,46 @@ where
     .unwrap_or_else(|_| panic!("read_response failed at block_on unwrap"))
 }
 
+/// Helper function that returns a response body of a ServiceResponse.
+/// This function blocks the current thread until futures complete.
+///
+/// ```rust
+/// use actix_web::{test, web, App, HttpResponse, http::header};
+/// use bytes::Bytes;
+///
+/// #[test]
+/// fn test_index() {
+///     let mut app = test::init_service(
+///         App::new().service(
+///             web::resource("/index.html")
+///                 .route(web::post().to(
+///                     || HttpResponse::Ok().body("welcome!")))));
+///
+///     let req = test::TestRequest::post()
+///         .uri("/index.html")
+///         .header(header::CONTENT_TYPE, "application/json")
+///         .to_request();
+///
+///     let resp = call_service(&mut srv, req);
+///     let result = test::read_body(resp);
+///     assert_eq!(result, Bytes::from_static(b"welcome!"));
+/// }
+/// ```
+pub fn read_body<B>(mut res: ServiceResponse<B>) -> Bytes
+where
+    B: MessageBody,
+{
+    block_on(run_on(move || {
+        res.take_body()
+            .fold(BytesMut::new(), move |mut body, chunk| {
+                body.extend_from_slice(&chunk);
+                Ok::<_, Error>(body)
+            })
+            .map(|body: BytesMut| body.freeze())
+    }))
+    .unwrap_or_else(|_| panic!("read_response failed at block_on unwrap"))
+}
+
 /// Helper function that returns a deserialized response body of a TestRequest
 /// This function blocks the current thread until futures complete.
 ///

src/web.rs

@@ -1,5 +1,5 @@
 //! Essentials helper functions and types for application registration.
-use actix_http::{http::Method, Response};
+use actix_http::http::Method;
 use futures::{Future, IntoFuture};
 
 pub use actix_http::Response as HttpResponse;
@@ -268,7 +268,7 @@ where
     F: AsyncFactory<I, R>,
     I: FromRequest + 'static,
     R: IntoFuture + 'static,
-    R::Item: Into<Response>,
+    R::Item: Responder,
     R::Error: Into<Error>,
 {
     Route::new().to_async(handler)