From 94d7a7f8733b690612a4844d1c06881b87b45762 Mon Sep 17 00:00:00 2001 From: Nikolay Kim Date: Thu, 11 Apr 2019 15:12:23 -0700 Subject: [PATCH] custom future for SendError service --- actix-framed/src/app.rs | 4 +- actix-framed/src/service.rs | 140 +++++++++--------------------------- actix-http/src/error.rs | 4 ++ 3 files changed, 39 insertions(+), 109 deletions(-) diff --git a/actix-framed/src/app.rs b/actix-framed/src/app.rs index cce618bb9..20bc2f771 100644 --- a/actix-framed/src/app.rs +++ b/actix-framed/src/app.rs @@ -88,7 +88,7 @@ pub struct FramedAppFactory { services: Rc>)>>, } -impl NewService for FramedAppFactory +impl NewService for FramedAppFactory where T: AsyncRead + AsyncWrite + 'static, S: 'static, @@ -100,7 +100,7 @@ where type Service = CloneableService>; type Future = CreateService; - fn new_service(&self, _: &()) -> Self::Future { + fn new_service(&self, _: &C) -> Self::Future { CreateService { fut: self .services diff --git a/actix-framed/src/service.rs b/actix-framed/src/service.rs index 5fb74fa1f..6e5c7a543 100644 --- a/actix-framed/src/service.rs +++ b/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::error::{Error, ResponseError}; +use actix_http::body::BodySize; +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 Default for VerifyWebSockets { } } -impl NewService for VerifyWebSockets { +impl NewService for VerifyWebSockets { type Request = (Request, Framed); type Response = (Request, Framed); type Error = (HandshakeError, Framed); @@ -30,7 +30,7 @@ impl NewService for VerifyWebSockets { type Service = VerifyWebSockets; type Future = FutureResult; - fn new_service(&self, _: &()) -> Self::Future { + fn new_service(&self, _: &C) -> Self::Future { ok(VerifyWebSockets { _t: PhantomData }) } } @@ -66,7 +66,7 @@ where } } -impl NewService for SendError +impl NewService for SendError where T: AsyncRead + AsyncWrite + 'static, R: 'static, @@ -74,12 +74,12 @@ where { type Request = Result)>; type Response = R; - type Error = Error; + type Error = (E, Framed); type InitError = (); type Service = SendError; type Future = FutureResult; - fn new_service(&self, _: &()) -> Self::Future { + fn new_service(&self, _: &C) -> Self::Future { ok(SendError(PhantomData)) } } @@ -92,8 +92,8 @@ where { type Request = Result)>; type Response = R; - type Error = Error; - type Future = Either, Box>>; + type Error = (E, Framed); + type Future = Either)>, SendErrorFut>; 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 e = Error::from(e); - Either::B(Box::new( - SendResponse::new(framed, res).then(move |_| Err(e)), - )) + let res = e.error_response().drop_body(); + Either::B(SendErrorFut { + framed: Some(framed), + res: Some((res, BodySize::Empty).into()), + err: Some(e), + _t: PhantomData, + }) } } } } -/// Send http/1 response -pub struct SendResponse { +pub struct SendErrorFut { res: Option, BodySize)>>, - body: Option>, framed: Option>, + err: Option, + _t: PhantomData, } -impl SendResponse -where - B: MessageBody, -{ - pub fn new(framed: Framed, response: Response) -> Self { - let (res, body) = response.into_parts(); - - SendResponse { - res: Some((res, body.size()).into()), - body: Some(body), - framed: Some(framed), - } - } -} - -impl Future for SendResponse +impl Future for SendErrorFut where + E: ResponseError, T: AsyncRead + AsyncWrite, - B: MessageBody, { - type Item = Framed; - type Error = (Error, Framed); + type Item = R; + type Error = (E, Framed); fn poll(&mut self) -> Poll { - 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; + if let Some(res) = self.res.take() { + if self.framed.as_mut().unwrap().force_send(res).is_err() { + return Err((self.err.take().unwrap(), self.framed.take().unwrap())); } } - 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())), + } } } diff --git a/actix-http/src/error.rs b/actix-http/src/error.rs index 92a046846..1768c9543 100644 --- a/actix-http/src/error.rs +++ b/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 {} + /// Return `BAD_REQUEST` for `de::value::Error` impl ResponseError for DeError { fn error_response(&self) -> Response {