Merge pull request #29 from reem/network-stream

Abstract over NetworkStream using dynamic dispatch

Author: seanmonstar

benches/client.rs

@@ -8,8 +8,9 @@ extern crate test;
 use std::fmt::{mod, Show};
 use std::io::net::ip::Ipv4Addr;
 use hyper::server::{Incoming, Server};
+use hyper::net::HttpAcceptor;
 
-fn listen() -> hyper::server::Listening {
+fn listen() -> hyper::server::Listening<HttpAcceptor> {
     let server = Server::http(Ipv4Addr(127, 0, 0, 1), 0);
     server.listen(handle).unwrap()
 }

examples/concurrent-server.rs

@@ -1,4 +1,4 @@
-#![feature(macro_rules)]
+#![feature(macro_rules, default_type_params)]
 
 extern crate hyper;
 extern crate debug;
@@ -10,15 +10,16 @@ use std::sync::Arc;
 use hyper::{Get, Post};
 use hyper::server::{Server, Handler, Incoming, Request, Response, Fresh};
 use hyper::header::common::ContentLength;
+use hyper::net::{HttpStream, HttpAcceptor};
 
 trait ConcurrentHandler: Send + Sync {
     fn handle(&self, req: Request, res: Response<Fresh>);
 }
 
 struct Concurrent<H: ConcurrentHandler> { handler: Arc<H> }
 
-impl<H: ConcurrentHandler> Handler for Concurrent<H> {
-    fn handle(self, mut incoming: Incoming) {
+impl<H: ConcurrentHandler> Handler<HttpAcceptor, HttpStream> for Concurrent<H> {
+    fn handle(self, mut incoming: Incoming<HttpAcceptor>) {
         for (mut req, mut res) in incoming {
             let clone = self.handler.clone();
             spawn(proc() { clone.handle(req, res) })

examples/server.rs

@@ -7,10 +7,8 @@ use std::io::util::copy;
 use std::io::net::ip::Ipv4Addr;
 
 use hyper::{Get, Post};
-use hyper::server::{Server, Handler, Incoming};
 use hyper::header::common::ContentLength;
-
-struct Echo;
+use hyper::server::{Server, Incoming};
 
 macro_rules! try_continue(
     ($e:expr) => {{
@@ -21,41 +19,39 @@ macro_rules! try_continue(
     }}
 )
 
-impl Handler for Echo {
-    fn handle(self, mut incoming: Incoming) {
-        for (mut req, mut res) in incoming {
-            match req.uri {
-                hyper::uri::AbsolutePath(ref path) => match (&req.method, path.as_slice()) {
-                    (&Get, "/") | (&Get, "/echo") => {
-                        let out = b"Try POST /echo";
-
-                        res.headers_mut().set(ContentLength(out.len()));
-                        let mut res = try_continue!(res.start());
-                        try_continue!(res.write(out));
-                        try_continue!(res.end());
-                        continue;
-                    },
-                    (&Post, "/echo") => (), // fall through, fighting mutable borrows
-                    _ => {
-                        *res.status_mut() = hyper::status::NotFound;
-                        try_continue!(res.start().and_then(|res| res.end()));
-                        continue;
-                    }
+fn echo(mut incoming: Incoming) {
+    for (mut req, mut res) in incoming {
+        match req.uri {
+            hyper::uri::AbsolutePath(ref path) => match (&req.method, path.as_slice()) {
+                (&Get, "/") | (&Get, "/echo") => {
+                    let out = b"Try POST /echo";
+
+                    res.headers_mut().set(ContentLength(out.len()));
+                    let mut res = try_continue!(res.start());
+                    try_continue!(res.write(out));
+                    try_continue!(res.end());
+                    continue;
                 },
+                (&Post, "/echo") => (), // fall through, fighting mutable borrows
                 _ => {
+                    *res.status_mut() = hyper::status::NotFound;
                     try_continue!(res.start().and_then(|res| res.end()));
                     continue;
                 }
-            };
-
-            let mut res = try_continue!(res.start());
-            try_continue!(copy(&mut req, &mut res));
-            try_continue!(res.end());
-        }
+            },
+            _ => {
+                try_continue!(res.start().and_then(|res| res.end()));
+                continue;
+            }
+        };
+
+        let mut res = try_continue!(res.start());
+        try_continue!(copy(&mut req, &mut res));
+        try_continue!(res.end());
     }
 }
 
 fn main() {
     let server = Server::http(Ipv4Addr(127, 0, 0, 1), 1337);
-    server.listen(Echo).unwrap();
+    server.listen(echo).unwrap();
 }

src/client/request.rs

@@ -1,12 +1,12 @@
 //! Client Requests
-use std::io::net::tcp::TcpStream;
 use std::io::{BufferedWriter, IoResult};
 
 use url::Url;
 
 use method;
 use header::Headers;
 use header::common::Host;
+use net::{NetworkStream, HttpStream};
 use rfc7230::LINE_ENDING;
 use version;
 use {HttpResult, HttpUriError};
@@ -24,7 +24,7 @@ pub struct Request {
     /// The HTTP version of this request.
     pub version: version::HttpVersion,
     headers_written: bool,
-    body: BufferedWriter<TcpStream>,
+    body: BufferedWriter<Box<NetworkStream + Send>>,
 }
 
 impl Request {
@@ -43,8 +43,8 @@ impl Request {
         };
         debug!("port={}", port);
 
-        let stream = try_io!(TcpStream::connect(host.as_slice(), port));
-        let stream = BufferedWriter::new(stream);
+        let stream: HttpStream = try_io!(NetworkStream::connect(host.as_slice(), port));
+        let stream = BufferedWriter::new(stream.abstract());
         let mut headers = Headers::new();
         headers.set(Host(host));
         Ok(Request {
@@ -84,8 +84,7 @@ impl Request {
     /// Consumes the Request.
     pub fn send(mut self) -> HttpResult<Response> {
         try_io!(self.flush());
-        let mut raw = self.body.unwrap();
-        try_io!(raw.close_write());
+        let raw = self.body.unwrap();
         Response::new(raw)
     }
 }

src/client/response.rs

@@ -1,33 +1,33 @@
 //! Client Responses
 use std::io::{BufferedReader, IoResult};
-use std::io::net::tcp::TcpStream;
 
 use header;
 use header::common::{ContentLength, TransferEncoding};
 use header::common::transfer_encoding::Chunked;
+use net::{NetworkStream, HttpStream};
 use rfc7230::{read_status_line, HttpReader, SizedReader, ChunkedReader, EofReader};
 use status;
 use version;
 use {HttpResult};
 
 /// A response for a client request to a remote server.
-pub struct Response {
+pub struct Response<S = HttpStream> {
     /// The status from the server.
     pub status: status::StatusCode,
     /// The headers from the server.
     pub headers: header::Headers,
     /// The HTTP version of this response from the server.
     pub version: version::HttpVersion,
-    body: HttpReader<BufferedReader<TcpStream>>,
+    body: HttpReader<BufferedReader<Box<NetworkStream + Send>>>,
 }
 
 impl Response {
 
     /// Creates a new response from a server.
-    pub fn new(tcp: TcpStream) -> HttpResult<Response> {
-        let mut tcp = BufferedReader::new(tcp);
-        let (version, status) = try!(read_status_line(&mut tcp));
-        let mut headers = try!(header::Headers::from_raw(&mut tcp));
+    pub fn new(stream: Box<NetworkStream + Send>) -> HttpResult<Response> {
+        let mut stream = BufferedReader::new(stream.abstract());
+        let (version, status) = try!(read_status_line(&mut stream));
+        let mut headers = try!(header::Headers::from_raw(&mut stream));
 
         debug!("{} {}", version, status);
         debug!("{}", headers);
@@ -40,22 +40,22 @@ impl Response {
                     };
 
                     if codings.contains(&Chunked) {
-                        ChunkedReader(tcp, None)
+                        ChunkedReader(stream, None)
                     } else {
-                        debug!("not chucked. read till eof");
-                        EofReader(tcp)
+                        debug!("not chuncked. read till eof");
+                        EofReader(stream)
                     }
                 }
                 None => unreachable!()
             }
         } else if headers.has::<ContentLength>() {
             match headers.get_ref::<ContentLength>() {
-                Some(&ContentLength(len)) => SizedReader(tcp, len),
+                Some(&ContentLength(len)) => SizedReader(stream, len),
                 None => unreachable!()
             }
         } else {
             debug!("neither Transfer-Encoding nor Content-Length");
-            EofReader(tcp)
+            EofReader(stream)
         };
 
         Ok(Response {
@@ -68,6 +68,7 @@ impl Response {
 }
 
 impl Reader for Response {
+    #[inline]
     fn read(&mut self, buf: &mut [u8]) -> IoResult<uint> {
         self.body.read(buf)
     }

src/lib.rs

@@ -1,6 +1,6 @@
 //! # hyper
-#![feature(macro_rules, phase)]
-#![warn(missing_doc)]
+#![feature(macro_rules, phase, default_type_params)]
+#![deny(missing_doc)]
 #![deny(warnings)]
 #![experimental]
 
@@ -53,6 +53,7 @@ macro_rules! trace(
 pub mod client;
 pub mod method;
 pub mod header;
+pub mod net;
 pub mod server;
 pub mod status;
 pub mod uri;