refactor h1 dispatcher

actix-http/src/h1/codec.rs

@@ -9,7 +9,7 @@ use http::header::{HeaderValue, CONNECTION, CONTENT_LENGTH, DATE, TRANSFER_ENCOD
 use http::{Method, StatusCode, Version};
 
 use super::decoder::{PayloadDecoder, PayloadItem, PayloadType};
-use super::{decoder, encoder, reserve_readbuf};
+use super::{decoder, encoder};
 use super::{Message, MessageType};
 use crate::body::BodySize;
 use crate::config::ServiceConfig;
@@ -31,7 +31,7 @@ const AVERAGE_HEADER_SIZE: usize = 30;
 
 /// HTTP/1 Codec
 pub struct Codec {
-    config: ServiceConfig,
+    pub(crate) config: ServiceConfig,
     decoder: decoder::MessageDecoder<Request>,
     payload: Option<PayloadDecoder>,
     version: Version,
@@ -78,16 +78,25 @@ impl Codec {
         }
     }
 
+    #[inline]
     /// Check if request is upgrade
     pub fn upgrade(&self) -> bool {
         self.ctype == ConnectionType::Upgrade
     }
 
+    #[inline]
     /// Check if last response is keep-alive
     pub fn keepalive(&self) -> bool {
         self.ctype == ConnectionType::KeepAlive
     }
 
+    #[inline]
+    /// Check if keep-alive enabled on server level
+    pub fn keepalive_enabled(&self) -> bool {
+        self.flags.contains(Flags::KEEPALIVE_ENABLED)
+    }
+
+    #[inline]
     /// Check last request's message type
     pub fn message_type(&self) -> MessageType {
         if self.flags.contains(Flags::STREAM) {
@@ -107,10 +116,7 @@ impl Decoder for Codec {
     fn decode(&mut self, src: &mut BytesMut) -> Result<Option<Self::Item>, Self::Error> {
         if self.payload.is_some() {
             Ok(match self.payload.as_mut().unwrap().decode(src)? {
-                Some(PayloadItem::Chunk(chunk)) => {
+                Some(PayloadItem::Chunk(chunk)) => Some(Message::Chunk(Some(chunk))),
-                    reserve_readbuf(src);
-                    Some(Message::Chunk(Some(chunk)))
-                }
                 Some(PayloadItem::Eof) => {
                     self.payload.take();
                     Some(Message::Chunk(None))
@@ -135,7 +141,6 @@ impl Decoder for Codec {
                     self.flags.insert(Flags::STREAM);
                 }
             }
-            reserve_readbuf(src);
             Ok(Some(Message::Item(req)))
         } else {
             Ok(None)

actix-http/src/h1/decoder.rs

@@ -84,7 +84,9 @@ pub(crate) trait MessageType: Sized {
                         header::CONTENT_LENGTH => {
                             if let Ok(s) = value.to_str() {
                                 if let Ok(len) = s.parse::<u64>() {
-                                    content_length = Some(len);
+                                    if len != 0 {
+                                        content_length = Some(len);
+                                    }
                                 } else {
                                     debug!("illegal Content-Length: {:?}", s);
                                     return Err(ParseError::Header);

actix-http/src/h1/dispatcher.rs

@@ -1,20 +1,20 @@
 use std::collections::VecDeque;
-use std::mem;
 use std::time::Instant;
+use std::{fmt, io};
 
-use actix_codec::{AsyncRead, AsyncWrite, Framed};
+use actix_codec::{AsyncRead, AsyncWrite, Decoder, Encoder};
 use actix_service::Service;
 use actix_utils::cloneable::CloneableService;
 use bitflags::bitflags;
-use futures::{Async, Future, Poll, Sink, Stream};
+use bytes::{BufMut, BytesMut};
-use log::{debug, error, trace};
+use futures::{Async, Future, Poll};
+use log::{error, trace};
 use tokio_timer::Delay;
 
 use crate::body::{Body, BodySize, MessageBody, ResponseBody};
 use crate::config::ServiceConfig;
 use crate::error::{DispatchError, Error};
 use crate::error::{ParseError, PayloadError};
-use crate::http::StatusCode;
 use crate::request::Request;
 use crate::response::Response;
 
@@ -22,17 +22,19 @@ use super::codec::Codec;
 use super::payload::{Payload, PayloadSender, PayloadStatus, PayloadWriter};
 use super::{Message, MessageType};
 
+const LW_BUFFER_SIZE: usize = 4096;
+const HW_BUFFER_SIZE: usize = 32_768;
 const MAX_PIPELINED_MESSAGES: usize = 16;
 
 bitflags! {
     pub struct Flags: u8 {
         const STARTED            = 0b0000_0001;
-        const KEEPALIVE_ENABLED  = 0b0000_0010;
+        const KEEPALIVE          = 0b0000_0010;
-        const KEEPALIVE          = 0b0000_0100;
+        const POLLED             = 0b0000_0100;
-        const POLLED             = 0b0000_1000;
+        const SHUTDOWN           = 0b0000_1000;
-        const SHUTDOWN           = 0b0010_0000;
+        const READ_DISCONNECT    = 0b0001_0000;
-        const DISCONNECTED       = 0b0100_0000;
+        const WRITE_DISCONNECT   = 0b0010_0000;
-        const DROPPING           = 0b1000_0000;
+        const DROPPING           = 0b0100_0000;
     }
 }
 
@@ -59,9 +61,7 @@ where
     service: CloneableService<S>,
     expect: CloneableService<X>,
     flags: Flags,
-    framed: Framed<T, Codec>,
     error: Option<DispatchError>,
-    config: ServiceConfig,
 
     state: State<S, B, X>,
     payload: Option<PayloadSender>,
@@ -69,6 +69,11 @@ where
 
     ka_expire: Instant,
     ka_timer: Option<Delay>,
+
+    io: T,
+    read_buf: BytesMut,
+    write_buf: BytesMut,
+    codec: Codec,
 }
 
 enum DispatcherMessage {
@@ -101,6 +106,30 @@ where
             false
         }
     }
+
+    fn is_call(&self) -> bool {
+        if let State::ServiceCall(_) = self {
+            true
+        } else {
+            false
+        }
+    }
+}
+
+impl<S, B, X> fmt::Debug for State<S, B, X>
+where
+    S: Service<Request = Request>,
+    X: Service<Request = Request, Response = Request>,
+    B: MessageBody,
+{
+    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
+        match self {
+            State::None => write!(f, "State::None"),
+            State::ExpectCall(_) => write!(f, "State::ExceptCall"),
+            State::ServiceCall(_) => write!(f, "State::ServiceCall"),
+            State::SendPayload(_) => write!(f, "State::SendPayload"),
+        }
+    }
 }
 
 impl<T, S, B, X> Dispatcher<T, S, B, X>
@@ -121,8 +150,10 @@ where
         expect: CloneableService<X>,
     ) -> Self {
         Dispatcher::with_timeout(
-            Framed::new(stream, Codec::new(config.clone())),
+            stream,
+            Codec::new(config.clone()),
             config,
+            BytesMut::with_capacity(HW_BUFFER_SIZE),
             None,
             service,
             expect,
@@ -131,15 +162,17 @@ where
 
     /// Create http/1 dispatcher with slow request timeout.
     pub fn with_timeout(
-        framed: Framed<T, Codec>,
+        io: T,
+        codec: Codec,
         config: ServiceConfig,
+        read_buf: BytesMut,
         timeout: Option<Delay>,
         service: CloneableService<S>,
         expect: CloneableService<X>,
     ) -> Self {
         let keepalive = config.keep_alive_enabled();
         let flags = if keepalive {
-            Flags::KEEPALIVE | Flags::KEEPALIVE_ENABLED
+            Flags::KEEPALIVE
         } else {
             Flags::empty()
         };
@@ -155,7 +188,10 @@ where
 
         Dispatcher {
             inner: Some(InnerDispatcher {
-                framed,
+                io,
+                codec,
+                read_buf,
+                write_buf: BytesMut::with_capacity(HW_BUFFER_SIZE),
                 payload: None,
                 state: State::None,
                 error: None,
@@ -163,7 +199,6 @@ where
                 service,
                 expect,
                 flags,
-                config,
                 ka_expire,
                 ka_timer,
             }),
@@ -182,11 +217,9 @@ where
     X::Error: Into<Error>,
 {
     fn can_read(&self) -> bool {
-        if self.flags.contains(Flags::DISCONNECTED) {
+        if self.flags.contains(Flags::READ_DISCONNECT) {
             return false;
-        }
+        } else if let Some(ref info) = self.payload {
-
-        if let Some(ref info) = self.payload {
             info.need_read() == PayloadStatus::Read
         } else {
             true
@@ -195,32 +228,52 @@ where
 
     // if checked is set to true, delay disconnect until all tasks have finished.
     fn client_disconnected(&mut self) {
-        self.flags.insert(Flags::DISCONNECTED);
+        self.flags
+            .insert(Flags::READ_DISCONNECT | Flags::WRITE_DISCONNECT);
         if let Some(mut payload) = self.payload.take() {
             payload.set_error(PayloadError::Incomplete(None));
         }
     }
 
     /// Flush stream
-    fn poll_flush(&mut self) -> Poll<bool, DispatchError> {
+    ///
-        if !self.framed.is_write_buf_empty() {
+    /// true - got whouldblock
-            match self.framed.poll_complete() {
+    /// false - didnt get whouldblock
-                Ok(Async::NotReady) => Ok(Async::NotReady),
+    fn poll_flush(&mut self) -> Result<bool, DispatchError> {
-                Err(err) => {
+        if self.write_buf.is_empty() {
-                    debug!("Error sending data: {}", err);
+            return Ok(false);
-                    Err(err.into())
-                }
-                Ok(Async::Ready(_)) => {
-                    // if payload is not consumed we can not use connection
-                    if self.payload.is_some() && self.state.is_empty() {
-                        return Err(DispatchError::PayloadIsNotConsumed);
-                    }
-                    Ok(Async::Ready(true))
-                }
-            }
-        } else {
-            Ok(Async::Ready(false))
         }
+
+        let len = self.write_buf.len();
+        let mut written = 0;
+        while written < len {
+            match self.io.write(&self.write_buf[written..]) {
+                Ok(0) => {
+                    return Err(DispatchError::Io(io::Error::new(
+                        io::ErrorKind::WriteZero,
+                        "",
+                    )));
+                }
+                Ok(n) => {
+                    written += n;
+                }
+                Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => {
+                    if written > 0 {
+                        let _ = self.write_buf.split_to(written);
+                    }
+                    return Ok(true);
+                }
+                Err(err) => return Err(DispatchError::Io(err)),
+            }
+        }
+        if written > 0 {
+            if written == self.write_buf.len() {
+                unsafe { self.write_buf.set_len(0) }
+            } else {
+                let _ = self.write_buf.split_to(written);
+            }
+        }
+        Ok(false)
     }
 
     fn send_response(
@@ -228,8 +281,8 @@ where
         message: Response<()>,
         body: ResponseBody<B>,
     ) -> Result<State<S, B, X>, DispatchError> {
-        self.framed
+        self.codec
-            .force_send(Message::Item((message, body.length())))
+            .encode(Message::Item((message, body.length())), &mut self.write_buf)
             .map_err(|err| {
                 if let Some(mut payload) = self.payload.take() {
                     payload.set_error(PayloadError::Incomplete(None));
@@ -237,113 +290,109 @@ where
                 DispatchError::Io(err)
             })?;
 
-        self.flags
+        self.flags.set(Flags::KEEPALIVE, self.codec.keepalive());
-            .set(Flags::KEEPALIVE, self.framed.get_codec().keepalive());
         match body.length() {
             BodySize::None | BodySize::Empty => Ok(State::None),
             _ => Ok(State::SendPayload(body)),
         }
     }
 
-    fn send_continue(&mut self) -> Result<(), DispatchError> {
+    fn send_continue(&mut self) {
-        self.framed
+        self.write_buf
-            .force_send(Message::Item((
+            .extend_from_slice(b"HTTP/1.1 100 Continue\r\n\r\n");
-                Response::empty(StatusCode::CONTINUE),
-                BodySize::Empty,
-            )))
-            .map_err(|err| DispatchError::Io(err))
     }
 
-    fn poll_response(&mut self) -> Result<(), DispatchError> {
+    fn poll_response(&mut self) -> Result<bool, DispatchError> {
-        let mut retry = self.can_read();
         loop {
-            let state = match mem::replace(&mut self.state, State::None) {
+            let state = match self.state {
                 State::None => match self.messages.pop_front() {
                     Some(DispatcherMessage::Item(req)) => {
                         Some(self.handle_request(req)?)
                     }
                     Some(DispatcherMessage::Error(res)) => {
-                        self.send_response(res, ResponseBody::Other(Body::Empty))?;
+                        Some(self.send_response(res, ResponseBody::Other(Body::Empty))?)
-                        None
                     }
                     None => None,
                 },
-                State::ExpectCall(mut fut) => match fut.poll() {
+                State::ExpectCall(ref mut fut) => match fut.poll() {
                     Ok(Async::Ready(req)) => {
-                        self.send_continue()?;
+                        self.send_continue();
-                        Some(State::ServiceCall(self.service.call(req)))
+                        self.state = State::ServiceCall(self.service.call(req));
-                    }
+                        continue;
-                    Ok(Async::NotReady) => {
-                        self.state = State::ExpectCall(fut);
-                        None
                     }
+                    Ok(Async::NotReady) => None,
                     Err(e) => {
-                        let e = e.into();
+                        let res: Response = e.into().into();
-                        let res: Response = e.into();
                         let (res, body) = res.replace_body(());
                         Some(self.send_response(res, body.into_body())?)
                     }
                 },
-                State::ServiceCall(mut fut) => match fut.poll() {
+                State::ServiceCall(ref mut fut) => match fut.poll() {
                     Ok(Async::Ready(res)) => {
                         let (res, body) = res.into().replace_body(());
-                        Some(self.send_response(res, body)?)
+                        self.state = self.send_response(res, body)?;
+                        continue;
                     }
-                    Ok(Async::NotReady) => {
+                    Ok(Async::NotReady) => None,
-                        self.state = State::ServiceCall(fut);
+                    Err(e) => {
-                        None
+                        let res: Response = e.into().into();
-                    }
-                    Err(_e) => {
-                        let res: Response = Response::InternalServerError().finish();
                         let (res, body) = res.replace_body(());
                         Some(self.send_response(res, body.into_body())?)
                     }
                 },
-                State::SendPayload(mut stream) => {
+                State::SendPayload(ref mut stream) => {
                     loop {
-                        if !self.framed.is_write_buf_full() {
+                        if self.write_buf.len() < HW_BUFFER_SIZE {
                             match stream
                                 .poll_next()
                                 .map_err(|_| DispatchError::Unknown)?
                             {
                                 Async::Ready(Some(item)) => {
-                                    self.framed
+                                    self.codec.encode(
-                                        .force_send(Message::Chunk(Some(item)))?;
+                                        Message::Chunk(Some(item)),
+                                        &mut self.write_buf,
+                                    )?;
                                     continue;
                                 }
                                 Async::Ready(None) => {
-                                    self.framed.force_send(Message::Chunk(None))?;
+                                    self.codec.encode(
-                                }
+                                        Message::Chunk(None),
-                                Async::NotReady => {
+                                        &mut self.write_buf,
-                                    self.state = State::SendPayload(stream);
+                                    )?;
-                                    return Ok(());
+                                    self.state = State::None;
                                 }
+                                Async::NotReady => return Ok(false),
                             }
                         } else {
-                            self.state = State::SendPayload(stream);
+                            return Ok(true);
-                            return Ok(());
                         }
                         break;
                     }
-                    None
+                    continue;
                 }
             };
 
-            match state {
+            // set new state
-                Some(state) => self.state = state,
+            if let Some(state) = state {
-                None => {
+                self.state = state;
-                    // if read-backpressure is enabled and we consumed some data.
+                if !self.state.is_empty() {
-                    // we may read more data and retry
+                    continue;
-                    if !retry && self.can_read() && self.poll_request()? {
+                }
-                        retry = self.can_read();
+            } else {
+                // if read-backpressure is enabled and we consumed some data.
+                // we may read more data and retry
+                if self.state.is_call() {
+                    if self.poll_request()? {
                         continue;
                     }
-                    break;
+                } else if !self.messages.is_empty() {
+                    continue;
                 }
             }
+            break;
         }
 
-        Ok(())
+        Ok(false)
     }
 
     fn handle_request(&mut self, req: Request) -> Result<State<S, B, X>, DispatchError> {
@@ -352,7 +401,7 @@ where
             let mut task = self.expect.call(req);
             match task.poll() {
                 Ok(Async::Ready(req)) => {
-                    self.send_continue()?;
+                    self.send_continue();
                     req
                 }
                 Ok(Async::NotReady) => return Ok(State::ExpectCall(task)),
@@ -375,8 +424,8 @@ where
                 self.send_response(res, body)
             }
             Ok(Async::NotReady) => Ok(State::ServiceCall(task)),
-            Err(_e) => {
+            Err(e) => {
-                let res: Response = Response::InternalServerError().finish();
+                let res: Response = e.into().into();
                 let (res, body) = res.replace_body(());
                 self.send_response(res, body.into_body())
             }
@@ -386,20 +435,20 @@ where
     /// Process one incoming requests
     pub(self) fn poll_request(&mut self) -> Result<bool, DispatchError> {
         // limit a mount of non processed requests
-        if self.messages.len() >= MAX_PIPELINED_MESSAGES {
+        if self.messages.len() >= MAX_PIPELINED_MESSAGES || !self.can_read() {
             return Ok(false);
         }
 
         let mut updated = false;
         loop {
-            match self.framed.poll() {
+            match self.codec.decode(&mut self.read_buf) {
-                Ok(Async::Ready(Some(msg))) => {
+                Ok(Some(msg)) => {
                     updated = true;
                     self.flags.insert(Flags::STARTED);
 
                     match msg {
                         Message::Item(mut req) => {
-                            match self.framed.get_codec().message_type() {
+                            match self.codec.message_type() {
                                 MessageType::Payload | MessageType::Stream => {
                                     let (ps, pl) = Payload::create(false);
                                     let (req1, _) =
@@ -424,7 +473,7 @@ where
                                 error!(
                                     "Internal server error: unexpected payload chunk"
                                 );
-                                self.flags.insert(Flags::DISCONNECTED);
+                                self.flags.insert(Flags::READ_DISCONNECT);
                                 self.messages.push_back(DispatcherMessage::Error(
                                     Response::InternalServerError().finish().drop_body(),
                                 ));
@@ -437,7 +486,7 @@ where
                                 payload.feed_eof();
                             } else {
                                 error!("Internal server error: unexpected eof");
-                                self.flags.insert(Flags::DISCONNECTED);
+                                self.flags.insert(Flags::READ_DISCONNECT);
                                 self.messages.push_back(DispatcherMessage::Error(
                                     Response::InternalServerError().finish().drop_body(),
                                 ));
@@ -447,11 +496,7 @@ where
                         }
                     }
                 }
-                Ok(Async::Ready(None)) => {
+                Ok(None) => break,
-                    self.client_disconnected();
-                    break;
-                }
-                Ok(Async::NotReady) => break,
                 Err(ParseError::Io(e)) => {
                     self.client_disconnected();
                     self.error = Some(DispatchError::Io(e));
@@ -466,15 +511,15 @@ where
                     self.messages.push_back(DispatcherMessage::Error(
                         Response::BadRequest().finish().drop_body(),
                     ));
-                    self.flags.insert(Flags::DISCONNECTED);
+                    self.flags.insert(Flags::READ_DISCONNECT);
                     self.error = Some(e.into());
                     break;
                 }
             }
         }
 
-        if self.ka_timer.is_some() && updated {
+        if updated && self.ka_timer.is_some() {
-            if let Some(expire) = self.config.keep_alive_expire() {
+            if let Some(expire) = self.codec.config.keep_alive_expire() {
                 self.ka_expire = expire;
             }
         }
@@ -486,10 +531,10 @@ where
         if self.ka_timer.is_none() {
             // shutdown timeout
             if self.flags.contains(Flags::SHUTDOWN) {
-                if let Some(interval) = self.config.client_disconnect_timer() {
+                if let Some(interval) = self.codec.config.client_disconnect_timer() {
                     self.ka_timer = Some(Delay::new(interval));
                 } else {
-                    self.flags.insert(Flags::DISCONNECTED);
+                    self.flags.insert(Flags::READ_DISCONNECT);
                     return Ok(());
                 }
             } else {
@@ -507,13 +552,14 @@ where
                     return Err(DispatchError::DisconnectTimeout);
                 } else if self.ka_timer.as_mut().unwrap().deadline() >= self.ka_expire {
                     // check for any outstanding tasks
-                    if self.state.is_empty() && self.framed.is_write_buf_empty() {
+                    if self.state.is_empty() && self.write_buf.is_empty() {
                         if self.flags.contains(Flags::STARTED) {
                             trace!("Keep-alive timeout, close connection");
                             self.flags.insert(Flags::SHUTDOWN);
 
                             // start shutdown timer
-                            if let Some(deadline) = self.config.client_disconnect_timer()
+                            if let Some(deadline) =
+                                self.codec.config.client_disconnect_timer()
                             {
                                 if let Some(timer) = self.ka_timer.as_mut() {
                                     timer.reset(deadline);
@@ -521,7 +567,7 @@ where
                                 }
                             } else {
                                 // no shutdown timeout, drop socket
-                                self.flags.insert(Flags::DISCONNECTED);
+                                self.flags.insert(Flags::WRITE_DISCONNECT);
                                 return Ok(());
                             }
                         } else {
@@ -538,7 +584,8 @@ where
                             self.flags.insert(Flags::STARTED | Flags::SHUTDOWN);
                             self.state = State::None;
                         }
-                    } else if let Some(deadline) = self.config.keep_alive_expire() {
+                    } else if let Some(deadline) = self.codec.config.keep_alive_expire()
+                    {
                         if let Some(timer) = self.ka_timer.as_mut() {
                             timer.reset(deadline);
                             let _ = timer.poll();
@@ -572,34 +619,60 @@ where
     #[inline]
     fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
         let inner = self.inner.as_mut().unwrap();
+        inner.poll_keepalive()?;
 
         if inner.flags.contains(Flags::SHUTDOWN) {
-            inner.poll_keepalive()?;
+            if inner.flags.contains(Flags::WRITE_DISCONNECT) {
-            if inner.flags.contains(Flags::DISCONNECTED) {
                 Ok(Async::Ready(()))
             } else {
-                // try_ready!(inner.poll_flush());
+                // flush buffer
-                match inner.framed.get_mut().shutdown()? {
+                inner.poll_flush()?;
-                    Async::Ready(_) => Ok(Async::Ready(())),
+                if !inner.write_buf.is_empty() {
-                    Async::NotReady => Ok(Async::NotReady),
+                    Ok(Async::NotReady)
+                } else {
+                    match inner.io.shutdown()? {
+                        Async::Ready(_) => Ok(Async::Ready(())),
+                        Async::NotReady => Ok(Async::NotReady),
+                    }
                 }
             }
         } else {
-            inner.poll_keepalive()?;
+            // read socket into a buf
+            if !inner.flags.contains(Flags::READ_DISCONNECT) {
+                if let Some(true) = read_available(&mut inner.io, &mut inner.read_buf)? {
+                    inner.flags.insert(Flags::READ_DISCONNECT)
+                }
+            }
+
             inner.poll_request()?;
             loop {
-                inner.poll_response()?;
+                if inner.write_buf.remaining_mut() < LW_BUFFER_SIZE {
-                if let Async::Ready(false) = inner.poll_flush()? {
+                    inner.write_buf.reserve(HW_BUFFER_SIZE);
+                }
+                let need_write = inner.poll_response()?;
+
+                // we didnt get WouldBlock from write operation,
+                // so data get written to kernel completely (OSX)
+                // and we have to write again otherwise response can get stuck
+                if inner.poll_flush()? || !need_write {
                     break;
                 }
             }
 
-            if inner.flags.contains(Flags::DISCONNECTED) {
+            // client is gone
+            if inner.flags.contains(Flags::WRITE_DISCONNECT) {
                 return Ok(Async::Ready(()));
             }
 
+            let is_empty = inner.state.is_empty();
+
+            // read half is closed and we do not processing any responses
+            if inner.flags.contains(Flags::READ_DISCONNECT) && is_empty {
+                inner.flags.insert(Flags::SHUTDOWN);
+            }
+
             // keep-alive and stream errors
-            if inner.state.is_empty() && inner.framed.is_write_buf_empty() {
+            if is_empty && inner.write_buf.is_empty() {
                 if let Some(err) = inner.error.take() {
                     Err(err)
                 }
@@ -623,13 +696,52 @@ where
     }
 }
 
+fn read_available<T>(io: &mut T, buf: &mut BytesMut) -> Result<Option<bool>, io::Error>
+where
+    T: io::Read,
+{
+    let mut read_some = false;
+    loop {
+        if buf.remaining_mut() < LW_BUFFER_SIZE {
+            buf.reserve(HW_BUFFER_SIZE);
+        }
+
+        let read = unsafe { io.read(buf.bytes_mut()) };
+        match read {
+            Ok(n) => {
+                if n == 0 {
+                    return Ok(Some(true));
+                } else {
+                    read_some = true;
+                    unsafe {
+                        buf.advance_mut(n);
+                    }
+                }
+            }
+            Err(e) => {
+                return if e.kind() == io::ErrorKind::WouldBlock {
+                    if read_some {
+                        Ok(Some(false))
+                    } else {
+                        Ok(None)
+                    }
+                } else if e.kind() == io::ErrorKind::ConnectionReset && read_some {
+                    Ok(Some(true))
+                } else {
+                    Err(e)
+                };
+            }
+        }
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use std::{cmp, io};
 
     use actix_codec::{AsyncRead, AsyncWrite};
     use actix_service::IntoService;
-    use bytes::{Buf, Bytes};
+    use bytes::{Buf, Bytes, BytesMut};
     use futures::future::{lazy, ok};
 
     use super::*;
@@ -638,6 +750,7 @@ mod tests {
 
     struct Buffer {
         buf: Bytes,
+        write_buf: BytesMut,
         err: Option<io::Error>,
     }
 
@@ -645,6 +758,7 @@ mod tests {
         fn new(data: &'static str) -> Buffer {
             Buffer {
                 buf: Bytes::from(data),
+                write_buf: BytesMut::new(),
                 err: None,
             }
         }
@@ -670,6 +784,7 @@ mod tests {
 
     impl io::Write for Buffer {
         fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
+            self.write_buf.extend(buf);
             Ok(buf.len())
         }
         fn flush(&mut self) -> io::Result<()> {
@@ -699,15 +814,17 @@ mod tests {
                 ),
                 CloneableService::new(ExpectHandler),
             );
-            assert!(h1.poll().is_ok());
+            assert!(h1.poll().is_err());
-            assert!(h1.poll().is_ok());
             assert!(h1
                 .inner
                 .as_ref()
                 .unwrap()
                 .flags
-                .contains(Flags::DISCONNECTED));
+                .contains(Flags::READ_DISCONNECT));
-            // assert_eq!(h1.tasks.len(), 1);
+            assert_eq!(
+                &h1.inner.as_ref().unwrap().io.write_buf[..26],
+                b"HTTP/1.1 400 Bad Request\r\n"
+            );
             ok::<_, ()>(())
         }));
     }

actix-http/src/h1/encoder.rs

@@ -41,8 +41,6 @@ impl<T: MessageType> Default for MessageEncoder<T> {
 pub(crate) trait MessageType: Sized {
     fn status(&self) -> Option<StatusCode>;
 
-    // fn connection_type(&self) -> Option<ConnectionType>;
-
     fn headers(&self) -> &HeaderMap;
 
     fn chunked(&self) -> bool;
@@ -171,10 +169,6 @@ impl MessageType for Response<()> {
         self.head().chunked()
     }
 
-    //fn connection_type(&self) -> Option<ConnectionType> {
-    //    self.head().ctype
-    //}
-
     fn headers(&self) -> &HeaderMap {
         &self.head().headers
     }

actix-http/src/response.rs

@@ -51,18 +51,6 @@ impl Response<Body> {
         }
     }
 
-    #[inline]
-    pub(crate) fn empty(status: StatusCode) -> Response<()> {
-        let mut head: Message<ResponseHead> = Message::new();
-        head.status = status;
-
-        Response {
-            head,
-            body: ResponseBody::Body(()),
-            error: None,
-        }
-    }
-
     /// Constructs an error response
     #[inline]
     pub fn from_error(error: Error) -> Response {

actix-http/src/service.rs

@@ -1,7 +1,7 @@
 use std::marker::PhantomData;
 use std::{fmt, io};
 
-use actix_codec::{AsyncRead, AsyncWrite, Framed, FramedParts};
+use actix_codec::{AsyncRead, AsyncWrite};
 use actix_server_config::{Io as ServerIo, Protocol, ServerConfig as SrvConfig};
 use actix_service::{IntoNewService, NewService, Service};
 use actix_utils::cloneable::CloneableService;
@@ -375,13 +375,14 @@ where
                     self.state =
                         State::Handshake(Some((server::handshake(io), cfg, srv)));
                 } else {
-                    let framed = Framed::from_parts(FramedParts::with_read_buf(
+                    self.state = State::H1(h1::Dispatcher::with_timeout(
                         io,
                         h1::Codec::new(cfg.clone()),
+                        cfg,
                         buf,
-                    ));
+                        None,
-                    self.state = State::H1(h1::Dispatcher::with_timeout(
+                        srv,
-                        framed, cfg, None, srv, expect,
+                        expect,
                     ))
                 }
                 self.poll()

actix-http/tests/test_server.rs

@@ -911,11 +911,11 @@ fn test_h1_service_error() {
     });
 
     let response = srv.block_on(srv.get("/").send()).unwrap();
-    assert_eq!(response.status(), http::StatusCode::INTERNAL_SERVER_ERROR);
+    assert_eq!(response.status(), http::StatusCode::BAD_REQUEST);
 
     // read response
     let bytes = srv.load_body(response).unwrap();
-    assert!(bytes.is_empty());
+    assert_eq!(bytes, Bytes::from_static(b"error"));
 }
 
 #[cfg(feature = "ssl")]

tests/test_server.rs

@@ -770,7 +770,7 @@ fn test_reading_deflate_encoding_large_random_ssl() {
                 awc::Connector::new()
                     .timeout(std::time::Duration::from_millis(500))
                     .ssl(builder.build())
-                    .service(),
+                    .finish(),
             )
             .finish()
     });