fix(http): send errors as frame errors

Additionally, only close read side, so it's potentially possible to
write a response for an error.

src/http/conn.rs

@@ -83,25 +83,20 @@ impl<I: Io, T: Http1Transaction, K: KeepAlive> Conn<I, T, K> {
             Ok(Some(head)) => (head.version, head),
             Ok(None) => return Ok(Async::NotReady),
             Err(e) => {
-                let must_respond_with_error = !self.state.was_idle();
+                self.state.close_read();
-                self.state.close();
                 self.io.consume_leading_lines();
-                let ret = if !self.io.read_buf().is_empty() {
+                let was_mid_parse = !self.io.read_buf().is_empty();
-                    error!("parse error ({}) with bytes: {:?}", e, self.io.read_buf());
+                let must_respond_with_error = !self.state.was_idle();
+                return if was_mid_parse {
+                    debug!("parse error ({}) with bytes: {:?}", e, self.io.read_buf());
+                    Ok(Async::Ready(Some(Frame::Error { error: e })))
+                } else if must_respond_with_error {
+                    trace!("parse error with 0 input, err = {:?}", e);
                     Ok(Async::Ready(Some(Frame::Error { error: e })))
                 } else {
-                    trace!("parse error with 0 input, err = {:?}", e);
+                    debug!("socket complete");
-                    if must_respond_with_error {
+                    Ok(Async::Ready(None))
-                        match e {
-                            ::Error::Io(io) => Err(io),
-                            other => Err(io::Error::new(io::ErrorKind::UnexpectedEof, other)),
-                        }
-                    } else {
-                        debug!("socket complete");
-                        Ok(Async::Ready(None))
-                    }
                 };
-                return ret;
             }
         };
 
@@ -111,7 +106,7 @@ impl<I: Io, T: Http1Transaction, K: KeepAlive> Conn<I, T, K> {
                     Ok(d) => d,
                     Err(e) => {
                         error!("decoder error = {:?}", e);
-                        self.state.close();
+                        self.state.close_read();
                         return Ok(Async::Ready(Some(Frame::Error { error: e })));
                     }
                 };
@@ -128,7 +123,7 @@ impl<I: Io, T: Http1Transaction, K: KeepAlive> Conn<I, T, K> {
             },
             _ => {
                 error!("unimplemented HTTP Version = {:?}", version);
-                self.state.close();
+                self.state.close_read();
                 return Ok(Async::Ready(Some(Frame::Error { error: ::Error::Version })));
             }
         }
@@ -310,6 +305,10 @@ where I: Io,
                 .map(|async| async.map(|chunk| Some(Frame::Body {
                     chunk: chunk
                 })))
+                .or_else(|err| {
+                    self.state.close_read();
+                    Ok(Async::Ready(Some(Frame::Error { error: err.into() })))
+                })
         } else {
             trace!("poll when on keep-alive");
             Ok(Async::NotReady)
@@ -476,6 +475,12 @@ impl<K: KeepAlive> State<K> {
         self.keep_alive.disable();
     }
 
+    fn close_read(&mut self) {
+        trace!("State::close_read()");
+        self.reading = Reading::Closed;
+        self.keep_alive.disable();
+    }
+
     fn try_keep_alive(&mut self) {
         match (&self.reading, &self.writing) {
             (&Reading::KeepAlive, &Writing::KeepAlive) => {