add bits to deal with Upgrade requests

src/client/response.rs

@@ -65,6 +65,11 @@ impl Response {
             body: body,
         })
     }
+
+    /// Unwraps the Request to return the NetworkStream underneath.
+    pub fn unwrap(self) -> Box<NetworkStream + Send> {
+        self.body.unwrap().unwrap()
+    }
 }
 
 impl Reader for Response {
@@ -73,3 +78,33 @@ impl Reader for Response {
         self.body.read(buf)
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use std::boxed::BoxAny;
+    use std::io::BufferedReader;
+
+    use header::Headers;
+    use http::EofReader;
+    use mock::MockStream;
+    use net::NetworkStream;
+    use status;
+    use version;
+
+    use super::Response;
+
+
+    #[test]
+    fn test_unwrap() {
+        let res = Response {
+            status: status::Ok,
+            headers: Headers::new(),
+            version: version::Http11,
+            body: EofReader(BufferedReader::new(box MockStream as Box<NetworkStream + Send>))
+        };
+
+        let b = res.unwrap().downcast::<MockStream>().unwrap();
+        assert_eq!(b, box MockStream);
+
+    }
+}

src/header/common/connection.rs

@@ -1,45 +1,62 @@
 use header::Header;
 use std::fmt::{mod, Show};
-use super::util::from_one_raw_str;
+use super::{from_comma_delimited, fmt_comma_delimited};
 use std::from_str::FromStr;
 
 /// The `Connection` header.
-///
-/// Describes whether the socket connection should be closed or reused after
-/// this request/response is completed.
 #[deriving(Clone, PartialEq, Show)]
-pub enum Connection {
+pub struct Connection(Vec<ConnectionOption>);
+
+/// Values that can be in the `Connection` header.
+#[deriving(Clone, PartialEq)]
+pub enum ConnectionOption {
     /// The `keep-alive` connection value.
     KeepAlive,
     /// The `close` connection value.
-    Close
+    Close,
+    /// Values in the Connection header that are supposed to be names of other Headers.
+    ///
+    /// > When a header field aside from Connection is used to supply control
+    /// > information for or about the current connection, the sender MUST list
+    /// > the corresponding field-name within the Connection header field.
+    // TODO: it would be nice if these "Strings" could be stronger types, since
+    // they are supposed to relate to other Header fields (which we have strong
+    // types for).
+    ConnectionHeader(String),
 }
 
-impl FromStr for Connection {
-    fn from_str(s: &str) -> Option<Connection> {
-        debug!("Connection::from_str =? {}", s);
+impl FromStr for ConnectionOption {
+    fn from_str(s: &str) -> Option<ConnectionOption> {
         match s {
             "keep-alive" => Some(KeepAlive),
             "close" => Some(Close),
-            _ => None
+            s => Some(ConnectionHeader(s.to_string()))
         }
     }
 }
 
+impl fmt::Show for ConnectionOption {
+    fn fmt(&self, fmt: &mut fmt::Formatter) -> fmt::Result {
+        match *self {
+            KeepAlive => "keep-alive",
+            Close => "close",
+            ConnectionHeader(ref s) => s.as_slice()
+        }.fmt(fmt)
+    }
+}
+
 impl Header for Connection {
     fn header_name(_: Option<Connection>) -> &'static str {
         "Connection"
     }
 
     fn parse_header(raw: &[Vec<u8>]) -> Option<Connection> {
-        from_one_raw_str(raw)
+        from_comma_delimited(raw).map(|vec| Connection(vec))
     }
 
     fn fmt_header(&self, fmt: &mut fmt::Formatter) -> fmt::Result {
-        match *self {
-            KeepAlive => "keep-alive",
-            Close => "close",
-        }.fmt(fmt)
+        let Connection(ref parts) = *self;
+        fmt_comma_delimited(fmt, parts[])
     }
 }
 

src/header/common/mod.rs

@@ -6,25 +6,21 @@
 //! strongly-typed theme, the [mime](http://seanmonstar.github.io/mime.rs) crate
 //! is used, such as `ContentType(pub Mime)`.
 
-pub use self::host::Host;
-pub use self::content_length::ContentLength;
-pub use self::content_type::ContentType;
 pub use self::accept::Accept;
 pub use self::connection::Connection;
+pub use self::content_length::ContentLength;
+pub use self::content_type::ContentType;
+pub use self::date::Date;
+pub use self::host::Host;
+pub use self::location::Location;
 pub use self::transfer_encoding::TransferEncoding;
+pub use self::upgrade::Upgrade;
 pub use self::user_agent::UserAgent;
 pub use self::server::Server;
-pub use self::date::Date;
-pub use self::location::Location;
 
-/// Exposes the Host header.
-pub mod host;
-
-/// Exposes the ContentLength header.
-pub mod content_length;
-
-/// Exposes the ContentType header.
-pub mod content_type;
+use std::fmt::{mod, Show};
+use std::from_str::FromStr;
+use std::str::from_utf8;
 
 /// Exposes the Accept header.
 pub mod accept;
@@ -32,17 +28,30 @@ pub mod accept;
 /// Exposes the Connection header.
 pub mod connection;
 
-/// Exposes the TransferEncoding header.
-pub mod transfer_encoding;
+/// Exposes the ContentLength header.
+pub mod content_length;
 
-/// Exposes the UserAgent header.
-pub mod user_agent;
+/// Exposes the ContentType header.
+pub mod content_type;
+
+/// Exposes the Date header.
+pub mod date;
+
+/// Exposes the Host header.
+pub mod host;
 
 /// Exposes the Server header.
 pub mod server;
 
-/// Exposes the Date header.
-pub mod date;
+/// Exposes the TransferEncoding header.
+pub mod transfer_encoding;
+
+/// Exposes the Upgrade header.
+pub mod upgrade;
+
+/// Exposes the UserAgent header.
+pub mod user_agent;
+
 
 /// Exposes the Location header.
 pub mod location;
@@ -50,3 +59,29 @@ pub mod location;
 pub mod util;
 
 
+fn from_comma_delimited<T: FromStr>(raw: &[Vec<u8>]) -> Option<Vec<T>> {
+    if raw.len() != 1 {
+        return None;
+    }
+    // we JUST checked that raw.len() == 1, so raw[0] WILL exist.
+    match from_utf8(unsafe { raw.as_slice().unsafe_get(0).as_slice() }) {
+        Some(s) => {
+            Some(s.as_slice()
+                 .split([',', ' '].as_slice())
+                 .filter_map(from_str)
+                 .collect())
+        }
+        None => None
+    }
+}
+
+fn fmt_comma_delimited<T: Show>(fmt: &mut fmt::Formatter, parts: &[T]) -> fmt::Result {
+    let last = parts.len() - 1;
+    for (i, part) in parts.iter().enumerate() {
+        try!(part.fmt(fmt));
+        if i < last {
+            try!(", ".fmt(fmt));
+        }
+    }
+    Ok(())
+}

src/header/common/transfer_encoding.rs

@@ -1,7 +1,7 @@
 use header::Header;
-use std::fmt::{mod, Show};
+use std::fmt;
 use std::from_str::FromStr;
-use std::str::from_utf8;
+use super::{from_comma_delimited, fmt_comma_delimited};
 
 /// The `Transfer-Encoding` header.
 ///
@@ -28,7 +28,7 @@ pub struct TransferEncoding(pub Vec<Encoding>);
 /// # use hyper::header::Headers;
 /// # let mut headers = Headers::new();
 /// headers.set(TransferEncoding(vec![Gzip, Chunked]));
-#[deriving(Clone, PartialEq, Show)]
+#[deriving(Clone, PartialEq)]
 pub enum Encoding {
     /// The `chunked` encoding.
     Chunked,
@@ -43,6 +43,18 @@ pub enum Encoding {
     EncodingExt(String)
 }
 
+impl fmt::Show for Encoding {
+    fn fmt(&self, fmt: &mut fmt::Formatter) -> fmt::Result {
+        match *self {
+            Chunked => "chunked",
+            Gzip => "gzip",
+            Deflate => "deflate",
+            Compress => "compress",
+            EncodingExt(ref s) => s.as_slice()
+        }.fmt(fmt)
+    }
+}
+
 impl FromStr for Encoding {
     fn from_str(s: &str) -> Option<Encoding> {
         match s {
@@ -61,31 +73,12 @@ impl Header for TransferEncoding {
     }
 
     fn parse_header(raw: &[Vec<u8>]) -> Option<TransferEncoding> {
-        if raw.len() != 1 {
-            return None;
-        }
-        // we JUST checked that raw.len() == 1, so raw[0] WILL exist.
-        match from_utf8(unsafe { raw.as_slice().unsafe_get(0).as_slice() }) {
-            Some(s) => {
-                Some(TransferEncoding(s.as_slice()
-                     .split([',', ' '].as_slice())
-                     .filter_map(from_str)
-                     .collect()))
-            }
-            None => None
-        }
+        from_comma_delimited(raw).map(|vec| TransferEncoding(vec))
     }
 
     fn fmt_header(&self, fmt: &mut fmt::Formatter) -> fmt::Result {
         let TransferEncoding(ref parts) = *self;
-        let last = parts.len() - 1;
-        for (i, part) in parts.iter().enumerate() {
-            try!(part.fmt(fmt));
-            if i < last {
-                try!(", ".fmt(fmt));
-            }
-        }
-        Ok(())
+        fmt_comma_delimited(fmt, parts[])
     }
 }
 

src/header/common/upgrade.rs

@@ -0,0 +1,51 @@
+use header::Header;
+use std::fmt::{mod, Show};
+use super::{from_comma_delimited, fmt_comma_delimited};
+use std::from_str::FromStr;
+
+/// The `Upgrade` header.
+#[deriving(Clone, PartialEq, Show)]
+pub struct Upgrade(Vec<Protocol>);
+
+/// Protocol values that can appear in the Upgrade header.
+#[deriving(Clone, PartialEq)]
+pub enum Protocol {
+    /// The websocket protocol.
+    WebSocket,
+    /// Some other less common protocol.
+    ProtocolExt(String),
+}
+
+impl FromStr for Protocol {
+    fn from_str(s: &str) -> Option<Protocol> {
+        match s {
+            "websocket" => Some(WebSocket),
+            s => Some(ProtocolExt(s.to_string()))
+        }
+    }
+}
+
+impl fmt::Show for Protocol {
+    fn fmt(&self, fmt: &mut fmt::Formatter) -> fmt::Result {
+        match *self {
+            WebSocket => "websocket",
+            ProtocolExt(ref s) => s.as_slice()
+        }.fmt(fmt)
+    }
+}
+
+impl Header for Upgrade {
+    fn header_name(_: Option<Upgrade>) -> &'static str {
+        "Upgrade"
+    }
+
+    fn parse_header(raw: &[Vec<u8>]) -> Option<Upgrade> {
+        from_comma_delimited(raw).map(|vec| Upgrade(vec))
+    }
+
+    fn fmt_header(&self, fmt: &mut fmt::Formatter) -> fmt::Result {
+        let Upgrade(ref parts) = *self;
+        fmt_comma_delimited(fmt, parts[])
+    }
+}
+

src/http.rs

@@ -38,6 +38,18 @@ pub enum HttpReader<R> {
     EofReader(R),
 }
 
+impl<R: Reader> HttpReader<R> {
+
+    /// Unwraps this HttpReader and returns the underlying Reader.
+    pub fn unwrap(self) -> R {
+        match self {
+            SizedReader(r, _) => r,
+            ChunkedReader(r, _) => r,
+            EofReader(r) => r,
+        }
+    }
+}
+
 impl<R: Reader> Reader for HttpReader<R> {
     fn read(&mut self, buf: &mut [u8]) -> IoResult<uint> {
         match *self {

src/lib.rs

@@ -184,6 +184,7 @@ pub mod status;
 pub mod uri;
 pub mod version;
 
+#[cfg(test)] mod mock;
 
 mod mimewrapper {
     /// Re-exporting the mime crate, for convenience.

src/mock.rs

@@ -0,0 +1,29 @@
+use std::io::IoResult;
+use std::io::net::ip::SocketAddr;
+
+use net::NetworkStream;
+
+#[deriving(Clone, PartialEq, Show)]
+pub struct MockStream;
+
+impl Reader for MockStream {
+    fn read(&mut self, _buf: &mut [u8]) -> IoResult<uint> {
+        unimplemented!()
+    }
+}
+
+impl Writer for MockStream {
+    fn write(&mut self, _msg: &[u8]) -> IoResult<()> {
+        unimplemented!()
+    }
+}
+
+impl NetworkStream for MockStream {
+    fn connect(_host: &str, _port: u16, _scheme: &str) -> IoResult<MockStream> {
+        Ok(MockStream)
+    }
+
+    fn peer_name(&mut self) -> IoResult<SocketAddr> {
+        Ok(from_str("127.0.0.1:1337").unwrap())
+    }
+}

src/net.rs

@@ -1,10 +1,18 @@
 //! A collection of traits abstracting over Listeners and Streams.
+use std::any::{Any, AnyRefExt};
+use std::boxed::BoxAny;
+use std::fmt;
+use std::intrinsics::TypeId;
 use std::io::{IoResult, IoError, ConnectionAborted, InvalidInput, OtherIoError,
               Stream, Listener, Acceptor};
 use std::io::net::ip::{SocketAddr, Port};
 use std::io::net::tcp::{TcpStream, TcpListener, TcpAcceptor};
+use std::mem::{mod, transmute, transmute_copy};
+use std::raw::{mod, TraitObject};
 use std::sync::{Arc, Mutex};
 
+use uany::UncheckedBoxAnyDowncast;
+use typeable::Typeable;
 use openssl::ssl::{SslStream, SslContext, Sslv23};
 use openssl::ssl::error::{SslError, StreamError, OpenSslErrors, SslSessionClosed};
 
@@ -15,7 +23,7 @@ pub struct Fresh;
 pub struct Streaming;
 
 /// An abstraction to listen for connections on a certain port.
-pub trait NetworkListener<S: NetworkStream, A: NetworkAcceptor<S>>: Listener<S, A> {
+pub trait NetworkListener<S: NetworkStream, A: NetworkAcceptor<S>>: Listener<S, A> + Typeable {
     /// Bind to a socket.
     ///
     /// Note: This does not start listening for connections. You must call
@@ -33,7 +41,7 @@ pub trait NetworkAcceptor<S: NetworkStream>: Acceptor<S> + Clone + Send {
 }
 
 /// An abstraction over streams that a Server can utilize.
-pub trait NetworkStream: Stream + Clone + Send {
+pub trait NetworkStream: Stream + Any + Clone + Send {
     /// Get the remote address of the underlying connection.
     fn peer_name(&mut self) -> IoResult<SocketAddr>;
 
@@ -52,6 +60,12 @@ pub trait NetworkStream: Stream + Clone + Send {
     fn clone_box(&self) -> Box<NetworkStream + Send> { self.clone().dynamic() }
 }
 
+impl fmt::Show for Box<NetworkStream + Send> {
+    fn fmt(&self, fmt: &mut fmt::Formatter) -> fmt::Result {
+        fmt.pad("Box<NetworkStream>")
+    }
+}
+
 impl Clone for Box<NetworkStream + Send> {
     #[inline]
     fn clone(&self) -> Box<NetworkStream + Send> { self.clone_box() }
@@ -70,6 +84,46 @@ impl Writer for Box<NetworkStream + Send> {
     fn flush(&mut self) -> IoResult<()> { (**self).flush() }
 }
 
+impl UncheckedBoxAnyDowncast for Box<NetworkStream + Send> {
+    unsafe fn downcast_unchecked<T: 'static>(self) -> Box<T>  {
+        let to = *mem::transmute::<&Box<NetworkStream + Send>, &raw::TraitObject>(&self);
+        // Prevent double-free.
+        mem::forget(self);
+        mem::transmute(to.data)
+    }
+}
+
+impl<'a> AnyRefExt<'a> for &'a NetworkStream + 'a {
+    #[inline]
+    fn is<T: 'static>(self) -> bool {
+        self.get_type_id() == TypeId::of::<T>()
+    }
+
+    #[inline]
+    fn downcast_ref<T: 'static>(self) -> Option<&'a T> {
+        if self.is::<T>() {
+            unsafe {
+                // Get the raw representation of the trait object
+                let to: TraitObject = transmute_copy(&self);
+                // Extract the data pointer
+                Some(transmute(to.data))
+            }
+        } else {
+            None
+        }
+    }
+}
+
+impl BoxAny for Box<NetworkStream + Send> {
+    fn downcast<T: 'static>(self) -> Result<Box<T>, Box<NetworkStream + Send>> {
+        if self.is::<T>() {
+            Ok(unsafe { self.downcast_unchecked() })
+        } else {
+            Err(self)
+        }
+    }
+}
+
 /// A `NetworkListener` for `HttpStream`s.
 pub struct HttpListener {
     inner: TcpListener
@@ -212,3 +266,30 @@ fn lift_ssl_error(ssl: SslError) -> IoError {
     }
 }
 
+#[cfg(test)]
+mod tests {
+    use std::boxed::BoxAny;
+    use uany::UncheckedBoxAnyDowncast;
+
+    use mock::MockStream;
+    use super::NetworkStream;
+
+    #[test]
+    fn test_downcast_box_stream() {
+        let stream = MockStream.dynamic();
+
+        let mock = stream.downcast::<MockStream>().unwrap();
+        assert_eq!(mock, box MockStream);
+
+    }
+
+    #[test]
+    fn test_downcast_unchecked_box_stream() {
+        let stream = MockStream.dynamic();
+
+        let mock = unsafe { stream.downcast_unchecked::<MockStream>() };
+        assert_eq!(mock, box MockStream);
+
+    }
+
+}