custom future for SendError service

actix-framed/src/app.rs

@@ -88,7 +88,7 @@ pub struct FramedAppFactory<T, S> {
     services: Rc<Vec<(String, BoxedHttpNewService<FramedRequest<T, S>>)>>,
 }
 
-impl<T, S> NewService for FramedAppFactory<T, S>
+impl<T, S, C> NewService<C> for FramedAppFactory<T, S>
 where
     T: AsyncRead + AsyncWrite + 'static,
     S: 'static,
@@ -100,7 +100,7 @@ where
     type Service = CloneableService<FramedAppService<T, S>>;
     type Future = CreateService<T, S>;
 
-    fn new_service(&self, _: &()) -> Self::Future {
+    fn new_service(&self, _: &C) -> Self::Future {
         CreateService {
             fut: self
                 .services

actix-framed/src/service.rs

@@ -1,8 +1,8 @@
 use std::marker::PhantomData;
 
 use actix_codec::{AsyncRead, AsyncWrite, Framed};
-use actix_http::body::{BodySize, MessageBody, ResponseBody};
+use actix_http::body::BodySize;
-use actix_http::error::{Error, ResponseError};
+use actix_http::error::ResponseError;
 use actix_http::h1::{Codec, Message};
 use actix_http::ws::{verify_handshake, HandshakeError};
 use actix_http::{Request, Response};
@@ -22,7 +22,7 @@ impl<T> Default for VerifyWebSockets<T> {
     }
 }
 
-impl<T> NewService for VerifyWebSockets<T> {
+impl<T, C> NewService<C> for VerifyWebSockets<T> {
     type Request = (Request, Framed<T, Codec>);
     type Response = (Request, Framed<T, Codec>);
     type Error = (HandshakeError, Framed<T, Codec>);
@@ -30,7 +30,7 @@ impl<T> NewService for VerifyWebSockets<T> {
     type Service = VerifyWebSockets<T>;
     type Future = FutureResult<Self::Service, Self::InitError>;
 
-    fn new_service(&self, _: &()) -> Self::Future {
+    fn new_service(&self, _: &C) -> Self::Future {
         ok(VerifyWebSockets { _t: PhantomData })
     }
 }
@@ -66,7 +66,7 @@ where
     }
 }
 
-impl<T, R, E> NewService for SendError<T, R, E>
+impl<T, R, E, C> NewService<C> for SendError<T, R, E>
 where
     T: AsyncRead + AsyncWrite + 'static,
     R: 'static,
@@ -74,12 +74,12 @@ where
 {
     type Request = Result<R, (E, Framed<T, Codec>)>;
     type Response = R;
-    type Error = Error;
+    type Error = (E, Framed<T, Codec>);
     type InitError = ();
     type Service = SendError<T, R, E>;
     type Future = FutureResult<Self::Service, Self::InitError>;
 
-    fn new_service(&self, _: &()) -> Self::Future {
+    fn new_service(&self, _: &C) -> Self::Future {
         ok(SendError(PhantomData))
     }
 }
@@ -92,8 +92,8 @@ where
 {
     type Request = Result<R, (E, Framed<T, Codec>)>;
     type Response = R;
-    type Error = Error;
+    type Error = (E, Framed<T, Codec>);
-    type Future = Either<FutureResult<R, Error>, Box<Future<Item = R, Error = Error>>>;
+    type Future = Either<FutureResult<R, (E, Framed<T, Codec>)>, SendErrorFut<T, R, E>>;
 
     fn poll_ready(&mut self) -> Poll<(), Self::Error> {
         Ok(Async::Ready(()))
@@ -103,119 +103,45 @@ where
         match req {
             Ok(r) => Either::A(ok(r)),
             Err((e, framed)) => {
-                let res = e.render_response();
+                let res = e.error_response().drop_body();
-                let e = Error::from(e);
+                Either::B(SendErrorFut {
-                Either::B(Box::new(
-                    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),
+                    res: Some((res, BodySize::Empty).into()),
+                    err: Some(e),
+                    _t: PhantomData,
+                })
+            }
         }
     }
 }
 
-impl<T, B> Future for SendResponse<T, B>
+pub struct SendErrorFut<T, R, E> {
+    res: Option<Message<(Response<()>, BodySize)>>,
+    framed: Option<Framed<T, Codec>>,
+    err: Option<E>,
+    _t: PhantomData<R>,
+}
+
+impl<T, R, E> Future for SendErrorFut<T, R, E>
 where
+    E: ResponseError,
     T: AsyncRead + AsyncWrite,
-    B: MessageBody,
 {
-    type Item = Framed<T, Codec>;
+    type Item = R;
-    type Error = (Error, Framed<T, Codec>);
+    type Error = (E, 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
+            if self.framed.as_mut().unwrap().force_send(res).is_err() {
-                    .as_mut()
+                return Err((self.err.take().unwrap(), self.framed.take().unwrap()));
-                    .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()))
+        match self.framed.as_mut().unwrap().poll_complete() {
+            Ok(Async::Ready(_)) => {
+                Err((self.err.take().unwrap(), self.framed.take().unwrap()))
+            }
+            Ok(Async::NotReady) => Ok(Async::NotReady),
+            Err(_) => Err((self.err.take().unwrap(), self.framed.take().unwrap())),
+        }
     }
 }

actix-http/src/error.rs

@@ -153,6 +153,10 @@ impl ResponseError for TimerError {}
 /// `InternalServerError` for `SslError`
 impl ResponseError for openssl::ssl::Error {}
 
+#[cfg(feature = "ssl")]
+/// `InternalServerError` for `SslError`
+impl ResponseError for openssl::ssl::HandshakeError<tokio_tcp::TcpStream> {}
+
 /// Return `BAD_REQUEST` for `de::value::Error`
 impl ResponseError for DeError {
     fn error_response(&self) -> Response {