fix(client): error on unsupport 101 responses, ignore other 1xx codes

Author: seanmonstar

src/error.rs

@@ -16,6 +16,7 @@ use self::Error::{
     Header,
     Status,
     Timeout,
+    Upgrade,
     Io,
     TooLarge,
     Incomplete,
@@ -44,6 +45,8 @@ pub enum Error {
     Status,
     /// A timeout occurred waiting for an IO event.
     Timeout,
+    /// A protocol upgrade was encountered, but not yet supported in hyper.
+    Upgrade,
     /// An `io::Error` that occurred while trying to read or write to a network stream.
     Io(IoError),
     /// Parsing a field as string failed
@@ -76,13 +79,14 @@ impl fmt::Display for Error {
 impl StdError for Error {
     fn description(&self) -> &str {
         match *self {
-            Method => "Invalid Method specified",
-            Version => "Invalid HTTP version specified",
-            Header => "Invalid Header provided",
-            TooLarge => "Message head is too large",
-            Status => "Invalid Status provided",
-            Incomplete => "Message is incomplete",
-            Timeout => "Timeout",
+            Method => "invalid Method specified",
+            Version => "invalid HTTP version specified",
+            Header => "invalid Header provided",
+            TooLarge => "message head is too large",
+            Status => "invalid Status provided",
+            Incomplete => "message is incomplete",
+            Timeout => "timeout",
+            Upgrade => "unsupported protocol upgrade",
             Uri(ref e) => e.description(),
             Io(ref e) => e.description(),
             Utf8(ref e) => e.description(),

src/proto/conn.rs

@@ -171,65 +171,71 @@ where I: AsyncRead + AsyncWrite,
         debug_assert!(self.can_read_head());
         trace!("Conn::read_head");
 
-        let (version, head) = match self.io.parse::<T>() {
-            Ok(Async::Ready(head)) => (head.version, head),
-            Ok(Async::NotReady) => return Ok(Async::NotReady),
-            Err(e) => {
-                // If we are currently waiting on a message, then an empty
-                // message should be reported as an error. If not, it is just
-                // the connection closing gracefully.
-                let must_error = self.should_error_on_eof();
-                self.state.close_read();
-                self.io.consume_leading_lines();
-                let was_mid_parse = !self.io.read_buf().is_empty();
-                return if was_mid_parse || must_error {
-                    debug!("parse error ({}) with {} bytes", e, self.io.read_buf().len());
-                    Err(e)
-                } else {
-                    debug!("read eof");
-                    Ok(Async::Ready(None))
-                };
-            }
-        };
+        loop {
+            let (version, head) = match self.io.parse::<T>() {
+                Ok(Async::Ready(head)) => (head.version, head),
+                Ok(Async::NotReady) => return Ok(Async::NotReady),
+                Err(e) => {
+                    // If we are currently waiting on a message, then an empty
+                    // message should be reported as an error. If not, it is just
+                    // the connection closing gracefully.
+                    let must_error = self.should_error_on_eof();
+                    self.state.close_read();
+                    self.io.consume_leading_lines();
+                    let was_mid_parse = !self.io.read_buf().is_empty();
+                    return if was_mid_parse || must_error {
+                        debug!("parse error ({}) with {} bytes", e, self.io.read_buf().len());
+                        Err(e)
+                    } else {
+                        debug!("read eof");
+                        Ok(Async::Ready(None))
+                    };
+                }
+            };
 
-        self.state.version = match version {
-            HttpVersion::Http10 => Version::Http10,
-            HttpVersion::Http11 => Version::Http11,
-            _ => {
-                error!("unimplemented HTTP Version = {:?}", version);
-                self.state.close_read();
-                return Err(::Error::Version);
-            }
-        };
+            self.state.version = match version {
+                HttpVersion::Http10 => Version::Http10,
+                HttpVersion::Http11 => Version::Http11,
+                _ => {
+                    error!("unimplemented HTTP Version = {:?}", version);
+                    self.state.close_read();
+                    return Err(::Error::Version);
+                }
+            };
 
-        let decoder = match T::decoder(&head, &mut self.state.method) {
-            Ok(d) => d,
-            Err(e) => {
-                debug!("decoder error = {:?}", e);
-                self.state.close_read();
-                return Err(e);
-            }
-        };
+            let decoder = match T::decoder(&head, &mut self.state.method) {
+                Ok(Some(d)) => d,
+                Ok(None) => {
+                    // likely a 1xx message that we can ignore
+                    continue;
+                }
+                Err(e) => {
+                    debug!("decoder error = {:?}", e);
+                    self.state.close_read();
+                    return Err(e);
+                }
+            };
 
-        debug!("incoming body is {}", decoder);
+            debug!("incoming body is {}", decoder);
 
-        self.state.busy();
-        if head.expecting_continue() {
-            let msg = b"HTTP/1.1 100 Continue\r\n\r\n";
-            self.state.writing = Writing::Continue(Cursor::new(msg));
-        }
-        let wants_keep_alive = head.should_keep_alive();
-        self.state.keep_alive &= wants_keep_alive;
-        let (body, reading) = if decoder.is_eof() {
-            (false, Reading::KeepAlive)
-        } else {
-            (true, Reading::Body(decoder))
-        };
-        self.state.reading = reading;
-        if !body {
-            self.try_keep_alive();
+            self.state.busy();
+            if head.expecting_continue() {
+                let msg = b"HTTP/1.1 100 Continue\r\n\r\n";
+                self.state.writing = Writing::Continue(Cursor::new(msg));
+            }
+            let wants_keep_alive = head.should_keep_alive();
+            self.state.keep_alive &= wants_keep_alive;
+            let (body, reading) = if decoder.is_eof() {
+                (false, Reading::KeepAlive)
+            } else {
+                (true, Reading::Body(decoder))
+            };
+            self.state.reading = reading;
+            if !body {
+                self.try_keep_alive();
+            }
+            return Ok(Async::Ready(Some((head, body))));
         }
-        Ok(Async::Ready(Some((head, body))))
     }
 
     pub fn read_body(&mut self) -> Poll<Option<super::Chunk>, io::Error> {