better error handling

src/server/channel.rs

@@ -38,7 +38,9 @@ where
     H: HttpHandler + 'static,
 {
     pub(crate) fn new(
-        settings: Rc<WorkerSettings<H>>, mut io: T, peer: Option<SocketAddr>,
+        settings: Rc<WorkerSettings<H>>,
+        mut io: T,
+        peer: Option<SocketAddr>,
         http2: bool,
     ) -> HttpChannel<T, H> {
         settings.add_channel();
@@ -61,7 +63,7 @@ where
                     settings,
                     peer,
                     io,
-                    BytesMut::with_capacity(4096),
+                    BytesMut::with_capacity(8192),
                 )),
             }
         }
@@ -93,12 +95,12 @@ where
             let el = self as *mut _;
             self.node = Some(Node::new(el));
             let _ = match self.proto {
-                Some(HttpProtocol::H1(ref mut h1)) => self.node
+                Some(HttpProtocol::H1(ref mut h1)) => {
-                    .as_ref()
+                    self.node.as_ref().map(|n| h1.settings().head().insert(n))
-                    .map(|n| h1.settings().head().insert(n)),
+                }
-                Some(HttpProtocol::H2(ref mut h2)) => self.node
+                Some(HttpProtocol::H2(ref mut h2)) => {
-                    .as_ref()
+                    self.node.as_ref().map(|n| h2.settings().head().insert(n))
-                    .map(|n| h2.settings().head().insert(n)),
+                }
                 Some(HttpProtocol::Unknown(ref mut settings, _, _, _)) => {
                     self.node.as_ref().map(|n| settings.head().insert(n))
                 }
@@ -168,9 +170,8 @@ where
         if let Some(HttpProtocol::Unknown(settings, addr, io, buf)) = self.proto.take() {
             match kind {
                 ProtocolKind::Http1 => {
-                    self.proto = Some(HttpProtocol::H1(h1::Http1::new(
+                    self.proto =
-                        settings, io, addr, buf,
+                        Some(HttpProtocol::H1(h1::Http1::new(settings, io, addr, buf)));
-                    )));
                     return self.poll();
                 }
                 ProtocolKind::Http2 => {

src/server/h1.rs

@@ -162,7 +162,6 @@ where
                     entry.pipe.disconnected()
                 }
                 // kill keepalive
-                self.flags.remove(Flags::KEEPALIVE);
                 self.keepalive_timer.take();
 
                 // on parse error, stop reading stream but tasks need to be
@@ -352,7 +351,7 @@ where
                                         Ok(Async::NotReady) => {}
                                         Err(err) => {
                                             error!("Unhandled error: {}", err);
-                                            self.flags.intersects(Flags::ERROR);
+                                            self.flags.insert(Flags::ERROR);
                                             return;
                                         }
                                     }

src/server/h1decoder.rs

@@ -46,7 +46,9 @@ impl H1Decoder {
     }
 
     pub fn decode<H>(
-        &mut self, src: &mut BytesMut, settings: &WorkerSettings<H>,
+        &mut self,
+        src: &mut BytesMut,
+        settings: &WorkerSettings<H>,
     ) -> Result<Option<Message>, DecoderError> {
         // read payload
         if self.decoder.is_some() {
@@ -64,18 +66,11 @@ impl H1Decoder {
             .map_err(DecoderError::Error)?
         {
             Async::Ready((msg, decoder)) => {
-                if let Some(decoder) = decoder {
+                self.decoder = decoder;
-                    self.decoder = Some(decoder);
                 Ok(Some(Message::Message {
                     msg,
-                        payload: true,
+                    payload: self.decoder.is_some(),
                 }))
-                } else {
-                    Ok(Some(Message::Message {
-                        msg,
-                        payload: false,
-                    }))
-                }
             }
             Async::NotReady => {
                 if src.len() >= MAX_BUFFER_SIZE {
@@ -89,7 +84,9 @@ impl H1Decoder {
     }
 
     fn parse_message<H>(
-        &self, buf: &mut BytesMut, settings: &WorkerSettings<H>,
+        &self,
+        buf: &mut BytesMut,
+        settings: &WorkerSettings<H>,
     ) -> Poll<(SharedHttpInnerMessage, Option<EncodingDecoder>), ParseError> {
         // Parse http message
         let mut has_upgrade = false;
@@ -148,7 +145,7 @@ impl H1Decoder {
                             header::CONTENT_LENGTH => {
                                 if let Ok(s) = value.to_str() {
                                     if let Ok(len) = s.parse::<u64>() {
-                                        content_length = Some(len)
+                                        content_length = Some(len);
                                     } else {
                                         debug!("illegal Content-Length: {:?}", len);
                                         return Err(ParseError::Header);
@@ -351,7 +348,10 @@ macro_rules! byte (
 
 impl ChunkedState {
     fn step(
-        &self, body: &mut BytesMut, size: &mut u64, buf: &mut Option<Bytes>,
+        &self,
+        body: &mut BytesMut,
+        size: &mut u64,
+        buf: &mut Option<Bytes>,
     ) -> Poll<ChunkedState, io::Error> {
         use self::ChunkedState::*;
         match *self {
@@ -414,7 +414,8 @@ impl ChunkedState {
         }
     }
     fn read_size_lf(
-        rdr: &mut BytesMut, size: &mut u64,
+        rdr: &mut BytesMut,
+        size: &mut u64,
     ) -> Poll<ChunkedState, io::Error> {
         match byte!(rdr) {
             b'\n' if *size > 0 => Ok(Async::Ready(ChunkedState::Body)),
@@ -427,7 +428,9 @@ impl ChunkedState {
     }
 
     fn read_body(
-        rdr: &mut BytesMut, rem: &mut u64, buf: &mut Option<Bytes>,
+        rdr: &mut BytesMut,
+        rem: &mut u64,
+        buf: &mut Option<Bytes>,
     ) -> Poll<ChunkedState, io::Error> {
         trace!("Chunked read, remaining={:?}", rem);