.to_async() handler can return Responder type #792

CHANGES.md

@@ -1,5 +1,14 @@
 # Changes
 
+### Added
+
+* Add helper functions for reading response body `test::read_body()`
+
+### Changed
+
+* `.to_async()` handler can return `Responder` type #792
+
+
 ## [1.0.0-beta.1] - 2019-04-20
 
 ### Added

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,32 +203,56 @@ 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> {
+        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,
+                ))),
+                Ok(Async::NotReady) => Ok(Async::NotReady),
+                Err(e) => {
+                    let res: Response = e.into().into();
+                    Ok(Async::Ready(ServiceResponse::new(
+                        self.req.take().unwrap(),
+                        res,
+                    )))
+                }
+            };
+        }
+
         match self.fut.poll() {
-            Ok(Async::Ready(res)) => Ok(Async::Ready(ServiceResponse::new(
+            Ok(Async::Ready(res)) => {
-                self.req.take().unwrap(),
+                self.fut2 =
-                res.into(),
+                    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();
@@ -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 =
+        let mut srv = init_service(
-            init_service(
+            App::new()
-                App::new().service(
+                .service(
                     web::resource("/test")
                         .route(web::get().to(|| HttpResponse::Ok()))
                         .route(web::put().to(|| {
@@ -444,8 +451,15 @@ 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")
             .method(Method::GET)
@@ -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)