revert SendResponse::Error type

actix-framed/src/service.rs

@@ -1,12 +1,14 @@
 use std::marker::PhantomData;
 
 use actix_codec::{AsyncRead, AsyncWrite, Framed};
+use actix_http::body::{BodySize, MessageBody, ResponseBody};
 use actix_http::error::{Error, ResponseError};
+use actix_http::h1::{Codec, Message};
 use actix_http::ws::{verify_handshake, HandshakeError};
-use actix_http::{h1, Request};
+use actix_http::{Request, Response};
 use actix_service::{NewService, Service};
 use futures::future::{ok, Either, FutureResult};
-use futures::{Async, Future, IntoFuture, Poll};
+use futures::{Async, Future, IntoFuture, Poll, Sink};
 
 /// Service that verifies incoming request if it is valid websocket
 /// upgrade request. In case of error returns `HandshakeError`
@@ -21,9 +23,9 @@ impl<T> Default for VerifyWebSockets<T> {
 }
 
 impl<T> NewService for VerifyWebSockets<T> {
-    type Request = (Request, Framed<T, h1::Codec>);
+    type Request = (Request, Framed<T, Codec>);
-    type Response = (Request, Framed<T, h1::Codec>);
+    type Response = (Request, Framed<T, Codec>);
-    type Error = (HandshakeError, Framed<T, h1::Codec>);
+    type Error = (HandshakeError, Framed<T, Codec>);
     type InitError = ();
     type Service = VerifyWebSockets<T>;
     type Future = FutureResult<Self::Service, Self::InitError>;
@@ -34,16 +36,16 @@ impl<T> NewService for VerifyWebSockets<T> {
 }
 
 impl<T> Service for VerifyWebSockets<T> {
-    type Request = (Request, Framed<T, h1::Codec>);
+    type Request = (Request, Framed<T, Codec>);
-    type Response = (Request, Framed<T, h1::Codec>);
+    type Response = (Request, Framed<T, Codec>);
-    type Error = (HandshakeError, Framed<T, h1::Codec>);
+    type Error = (HandshakeError, Framed<T, Codec>);
     type Future = FutureResult<Self::Response, Self::Error>;
 
     fn poll_ready(&mut self) -> Poll<(), Self::Error> {
         Ok(Async::Ready(()))
     }
 
-    fn call(&mut self, (req, framed): (Request, Framed<T, h1::Codec>)) -> Self::Future {
+    fn call(&mut self, (req, framed): (Request, Framed<T, Codec>)) -> Self::Future {
         match verify_handshake(req.head()) {
             Err(e) => Err((e, framed)).into_future(),
             Ok(_) => Ok((req, framed)).into_future(),
@@ -70,7 +72,7 @@ where
     R: 'static,
     E: ResponseError + 'static,
 {
-    type Request = Result<R, (E, Framed<T, h1::Codec>)>;
+    type Request = Result<R, (E, Framed<T, Codec>)>;
     type Response = R;
     type Error = Error;
     type InitError = ();
@@ -88,7 +90,7 @@ where
     R: 'static,
     E: ResponseError + 'static,
 {
-    type Request = Result<R, (E, Framed<T, h1::Codec>)>;
+    type Request = Result<R, (E, Framed<T, Codec>)>;
     type Response = R;
     type Error = Error;
     type Future = Either<FutureResult<R, Error>, Box<Future<Item = R, Error = Error>>>;
@@ -97,16 +99,123 @@ where
         Ok(Async::Ready(()))
     }
 
-    fn call(&mut self, req: Result<R, (E, Framed<T, h1::Codec>)>) -> Self::Future {
+    fn call(&mut self, req: Result<R, (E, Framed<T, Codec>)>) -> Self::Future {
         match req {
             Ok(r) => Either::A(ok(r)),
             Err((e, framed)) => {
                 let res = e.render_response();
                 let e = Error::from(e);
                 Either::B(Box::new(
-                    h1::SendResponse::new(framed, res).then(move |_| Err(e)),
+                    SendResponse::new(framed, res).then(move |_| Err(e)),
                 ))
             }
         }
     }
 }
+
+/// Send http/1 response
+pub struct SendResponse<T, B> {
+    res: Option<Message<(Response<()>, BodySize)>>,
+    body: Option<ResponseBody<B>>,
+    framed: Option<Framed<T, Codec>>,
+}
+
+impl<T, B> SendResponse<T, B>
+where
+    B: MessageBody,
+{
+    pub fn new(framed: Framed<T, Codec>, response: Response<B>) -> Self {
+        let (res, body) = response.into_parts();
+
+        SendResponse {
+            res: Some((res, body.size()).into()),
+            body: Some(body),
+            framed: Some(framed),
+        }
+    }
+}
+
+impl<T, B> Future for SendResponse<T, B>
+where
+    T: AsyncRead + AsyncWrite,
+    B: MessageBody,
+{
+    type Item = Framed<T, Codec>;
+    type Error = (Error, Framed<T, Codec>);
+
+    fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
+        loop {
+            let mut body_ready = self.body.is_some();
+
+            // send body
+            if self.res.is_none() && self.body.is_some() {
+                while body_ready
+                    && self.body.is_some()
+                    && !self.framed.as_ref().unwrap().is_write_buf_full()
+                {
+                    match self
+                        .body
+                        .as_mut()
+                        .unwrap()
+                        .poll_next()
+                        .map_err(|e| (e, self.framed.take().unwrap()))?
+                    {
+                        Async::Ready(item) => {
+                            // body is done
+                            if item.is_none() {
+                                let _ = self.body.take();
+                            }
+                            self.framed
+                                .as_mut()
+                                .unwrap()
+                                .force_send(Message::Chunk(item))
+                                .map_err(|e| (e.into(), self.framed.take().unwrap()))?;
+                        }
+                        Async::NotReady => body_ready = false,
+                    }
+                }
+            }
+
+            // flush write buffer
+            if !self.framed.as_ref().unwrap().is_write_buf_empty() {
+                match self
+                    .framed
+                    .as_mut()
+                    .unwrap()
+                    .poll_complete()
+                    .map_err(|e| (e.into(), self.framed.take().unwrap()))?
+                {
+                    Async::Ready(_) => {
+                        if body_ready {
+                            continue;
+                        } else {
+                            return Ok(Async::NotReady);
+                        }
+                    }
+                    Async::NotReady => return Ok(Async::NotReady),
+                }
+            }
+
+            // send response
+            if let Some(res) = self.res.take() {
+                self.framed
+                    .as_mut()
+                    .unwrap()
+                    .force_send(res)
+                    .map_err(|e| (e.into(), self.framed.take().unwrap()))?;
+                continue;
+            }
+
+            if self.body.is_some() {
+                if body_ready {
+                    continue;
+                } else {
+                    return Ok(Async::NotReady);
+                }
+            } else {
+                break;
+            }
+        }
+        Ok(Async::Ready(self.framed.take().unwrap()))
+    }
+}

actix-http/src/h1/utils.rs

@@ -34,35 +34,23 @@ where
     B: MessageBody,
 {
     type Item = Framed<T, Codec>;
-    type Error = (Error, Framed<T, Codec>);
+    type Error = Error;
 
     fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
         loop {
             let mut body_ready = self.body.is_some();
+            let framed = self.framed.as_mut().unwrap();
 
             // send body
             if self.res.is_none() && self.body.is_some() {
-                while body_ready
+                while body_ready && self.body.is_some() && !framed.is_write_buf_full() {
-                    && self.body.is_some()
+                    match self.body.as_mut().unwrap().poll_next()? {
-                    && !self.framed.as_ref().unwrap().is_write_buf_full()
-                {
-                    match self
-                        .body
-                        .as_mut()
-                        .unwrap()
-                        .poll_next()
-                        .map_err(|e| (e, self.framed.take().unwrap()))?
-                    {
                         Async::Ready(item) => {
                             // body is done
                             if item.is_none() {
                                 let _ = self.body.take();
                             }
-                            self.framed
+                            framed.force_send(Message::Chunk(item))?;
-                                .as_mut()
-                                .unwrap()
-                                .force_send(Message::Chunk(item))
-                                .map_err(|e| (e.into(), self.framed.take().unwrap()))?;
                         }
                         Async::NotReady => body_ready = false,
                     }
@@ -70,14 +58,8 @@ where
             }
 
             // flush write buffer
-            if !self.framed.as_ref().unwrap().is_write_buf_empty() {
+            if !framed.is_write_buf_empty() {
-                match self
+                match framed.poll_complete()? {
-                    .framed
-                    .as_mut()
-                    .unwrap()
-                    .poll_complete()
-                    .map_err(|e| (e.into(), self.framed.take().unwrap()))?
-                {
                     Async::Ready(_) => {
                         if body_ready {
                             continue;
@@ -91,11 +73,7 @@ where
 
             // send response
             if let Some(res) = self.res.take() {
-                self.framed
+                framed.force_send(res)?;
-                    .as_mut()
-                    .unwrap()
-                    .force_send(res)
-                    .map_err(|e| (e.into(), self.framed.take().unwrap()))?;
                 continue;
             }