Merge pull request #29 from reem/network-stream

Abstract over NetworkStream using dynamic dispatch

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,6 +10,7 @@ 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>);
@@ -17,8 +18,8 @@ trait ConcurrentHandler: Send + Sync {
 
 struct Concurrent<H: ConcurrentHandler> { handler: Arc<H> }
 
-impl<H: ConcurrentHandler> Handler for Concurrent<H> {
+impl<H: ConcurrentHandler> Handler<HttpAcceptor, HttpStream> for Concurrent<H> {
-    fn handle(self, mut incoming: Incoming) {
+    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;
-
+use hyper::server::{Server, Incoming};
-struct Echo;
 
 macro_rules! try_continue(
     ($e:expr) => {{
@@ -21,8 +19,7 @@ macro_rules! try_continue(
     }}
 )
 
-impl Handler for Echo {
+fn echo(mut incoming: Incoming) {
-    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()) {
@@ -53,9 +50,8 @@ impl Handler for Echo {
         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,5 +1,4 @@
 //! Client Requests
-use std::io::net::tcp::TcpStream;
 use std::io::{BufferedWriter, IoResult};
 
 use url::Url;
@@ -7,6 +6,7 @@ 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: HttpStream = try_io!(NetworkStream::connect(host.as_slice(), port));
-        let stream = BufferedWriter::new(stream);
+        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();
+        let raw = self.body.unwrap();
-        try_io!(raw.close_write());
         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> {
+    pub fn new(stream: Box<NetworkStream + Send>) -> HttpResult<Response> {
-        let mut tcp = BufferedReader::new(tcp);
+        let mut stream = BufferedReader::new(stream.abstract());
-        let (version, status) = try!(read_status_line(&mut tcp));
+        let (version, status) = try!(read_status_line(&mut stream));
-        let mut headers = try!(header::Headers::from_raw(&mut tcp));
+        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");
+                        debug!("not chuncked. read till eof");
-                        EofReader(tcp)
+                        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)]
+#![feature(macro_rules, phase, default_type_params)]
-#![warn(missing_doc)]
+#![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;

src/net.rs

@@ -0,0 +1,149 @@
+//! A collection of traits abstracting over Listeners and Streams.
+use std::io::{IoResult, Stream, Listener, Acceptor};
+use std::io::net::ip::{SocketAddr, Port};
+use std::io::net::tcp::{TcpStream, TcpListener, TcpAcceptor};
+
+/// An abstraction to listen for connections on a certain port.
+pub trait NetworkListener<S: NetworkStream, A: NetworkAcceptor<S>>: Listener<S, A> {
+    /// Bind to a socket.
+    ///
+    /// Note: This does not start listening for connections. You must call
+    /// `listen()` to do that.
+    fn bind(host: &str, port: Port) -> IoResult<Self>;
+
+    /// Get the address this Listener ended up listening on.
+    fn socket_name(&mut self) -> IoResult<SocketAddr>;
+}
+
+/// An abstraction to receive `HttpStream`s.
+pub trait NetworkAcceptor<S: NetworkStream>: Acceptor<S> + Clone + Send {
+    /// Closes the Acceptor, so no more incoming connections will be handled.
+    fn close(&mut self) -> IoResult<()>;
+}
+
+/// An abstraction over streams that a Server can utilize.
+pub trait NetworkStream: Stream + Clone + Send {
+    /// Get the remote address of the underlying connection.
+    fn peer_name(&mut self) -> IoResult<SocketAddr>;
+
+    /// Connect to a remote address.
+    fn connect(host: &str, port: Port) -> IoResult<Self>;
+
+    /// Turn this into an appropriately typed trait object.
+    #[inline]
+    fn abstract(self) -> Box<NetworkStream + Send> {
+        box self as Box<NetworkStream + Send>
+    }
+
+    #[doc(hidden)]
+    #[inline]
+    // Hack to work around lack of Clone impl for Box<Clone>
+    fn clone_box(&self) -> Box<NetworkStream + Send> { self.clone().abstract() }
+}
+
+impl Clone for Box<NetworkStream + Send> {
+    #[inline]
+    fn clone(&self) -> Box<NetworkStream + Send> { self.clone_box() }
+}
+
+impl Reader for Box<NetworkStream + Send> {
+    #[inline]
+    fn read(&mut self, buf: &mut [u8]) -> IoResult<uint> { self.read(buf) }
+}
+
+impl Writer for Box<NetworkStream + Send> {
+    #[inline]
+    fn write(&mut self, msg: &[u8]) -> IoResult<()> { self.write(msg) }
+
+    #[inline]
+    fn flush(&mut self) -> IoResult<()> { self.flush() }
+}
+
+/// A `NetworkListener` for `HttpStream`s.
+pub struct HttpListener {
+    inner: TcpListener
+}
+
+impl Listener<HttpStream, HttpAcceptor> for HttpListener {
+    #[inline]
+    fn listen(self) -> IoResult<HttpAcceptor> {
+        Ok(HttpAcceptor {
+            inner: try!(self.inner.listen())
+        })
+    }
+}
+
+impl NetworkListener<HttpStream, HttpAcceptor> for HttpListener {
+    #[inline]
+    fn bind(host: &str, port: Port) -> IoResult<HttpListener> {
+        Ok(HttpListener {
+            inner: try!(TcpListener::bind(host, port))
+        })
+    }
+
+    #[inline]
+    fn socket_name(&mut self) -> IoResult<SocketAddr> {
+        self.inner.socket_name()
+    }
+}
+
+/// A `NetworkAcceptor` for `HttpStream`s.
+#[deriving(Clone)]
+pub struct HttpAcceptor {
+    inner: TcpAcceptor
+}
+
+impl Acceptor<HttpStream> for HttpAcceptor {
+    #[inline]
+    fn accept(&mut self) -> IoResult<HttpStream> {
+        Ok(HttpStream {
+            inner: try!(self.inner.accept())
+        })
+    }
+}
+
+impl NetworkAcceptor<HttpStream> for HttpAcceptor {
+    #[inline]
+    fn close(&mut self) -> IoResult<()> {
+        self.inner.close_accept()
+    }
+}
+
+/// A wrapper around a TcpStream.
+#[deriving(Clone)]
+pub struct HttpStream {
+    inner: TcpStream
+}
+
+impl Reader for HttpStream {
+    #[inline]
+    fn read(&mut self, buf: &mut [u8]) -> IoResult<uint> {
+        self.inner.read(buf)
+    }
+}
+
+impl Writer for HttpStream {
+    #[inline]
+    fn write(&mut self, msg: &[u8]) -> IoResult<()> {
+        self.inner.write(msg)
+    }
+    #[inline]
+    fn flush(&mut self) -> IoResult<()> {
+        self.inner.flush()
+    }
+}
+
+
+impl NetworkStream for HttpStream {
+    #[inline]
+    fn peer_name(&mut self) -> IoResult<SocketAddr> {
+        self.inner.peer_name()
+    }
+
+    #[inline]
+    fn connect(host: &str, port: Port) -> IoResult<HttpStream> {
+        Ok(HttpStream {
+            inner: try!(TcpStream::connect(host, port))
+        })
+    }
+}

src/server/mod.rs

@@ -1,11 +1,12 @@
 //! HTTP Server
-use std::io::net::tcp::{TcpListener, TcpAcceptor};
 use std::io::{Acceptor, Listener, IoResult, EndOfFile, IncomingConnections};
 use std::io::net::ip::{IpAddr, Port, SocketAddr};
 
 pub use self::request::Request;
 pub use self::response::{Response, Fresh, Streaming};
 
+use net::{NetworkListener, NetworkAcceptor, NetworkStream, HttpAcceptor, HttpListener};
+
 pub mod request;
 pub mod response;
 
@@ -13,32 +14,39 @@ pub mod response;
 ///
 /// Once listening, it will create a `Request`/`Response` pair for each
 /// incoming connection, and hand them to the provided handler.
-pub struct Server {
+pub struct Server<L = HttpListener> {
     ip: IpAddr,
     port: Port
 }
 
-
+impl Server<HttpListener> {
-impl Server {
+    /// Creates a new server that will handle `HttpStream`s.
-
-    /// Creates a server to be used for `http` conenctions.
     pub fn http(ip: IpAddr, port: Port) -> Server {
         Server {
             ip: ip,
             port: port
         }
     }
+}
+
+impl<L: NetworkListener<S, A>, S: NetworkStream, A: NetworkAcceptor<S>> Server<L> {
+    /// Creates a server that can listen for and handle `NetworkStreams`.
+    pub fn new(ip: IpAddr, port: Port) -> Server<L> {
+        Server {
+            ip: ip,
+            port: port
+        }
+    }
 
     /// Binds to a socket, and starts handling connections.
-    pub fn listen<H: Handler + 'static>(self, handler: H) -> IoResult<Listening> {
+    pub fn listen<H: Handler<A, S>>(self, handler: H) -> IoResult<Listening<A>> {
-        let mut listener = try!(TcpListener::bind(self.ip.to_string().as_slice(), self.port));
+        let mut listener: L = try!(NetworkListener::bind(self.ip.to_string().as_slice(), self.port));
         let socket = try!(listener.socket_name());
         let acceptor = try!(listener.listen());
-        let worker = acceptor.clone();
+        let mut worker = acceptor.clone();
 
         spawn(proc() {
-            let mut acceptor = worker;
+            handler.handle(Incoming { from: worker.incoming() });
-            handler.handle(Incoming { from: acceptor.incoming() });
         });
 
         Ok(Listening {
@@ -51,11 +59,11 @@ impl Server {
 
 /// An iterator over incoming connections, represented as pairs of
 /// hyper Requests and Responses.
-pub struct Incoming<'a> {
+pub struct Incoming<'a, A: 'a = HttpAcceptor> {
-    from: IncomingConnections<'a, TcpAcceptor>
+    from: IncomingConnections<'a, A>
 }
 
-impl<'a> Iterator<(Request, Response<Fresh>)> for Incoming<'a> {
+impl<'a, S: NetworkStream, A: NetworkAcceptor<S>> Iterator<(Request, Response<Fresh>)> for Incoming<'a, A> {
     fn next(&mut self) -> Option<(Request, Response<Fresh>)> {
         for conn in self.from {
             match conn {
@@ -85,30 +93,30 @@ impl<'a> Iterator<(Request, Response<Fresh>)> for Incoming<'a> {
 }
 
 /// A listening server, which can later be closed.
-pub struct Listening {
+pub struct Listening<A> {
-    acceptor: TcpAcceptor,
+    acceptor: A,
     /// The socket address that the server is bound to.
     pub socket_addr: SocketAddr,
 }
 
-impl Listening {
+impl<A: NetworkAcceptor<S>, S: NetworkStream> Listening<A> {
-    /// Stop the server from listening to its socket address.
+    /// Stop the server from listening to it's socket address.
     pub fn close(mut self) -> IoResult<()> {
         debug!("closing server");
-        self.acceptor.close_accept()
+        self.acceptor.close()
     }
 }
 
 /// A handler that can handle incoming requests for a server.
-pub trait Handler: Send {
+pub trait Handler<A: NetworkAcceptor<S>, S: NetworkStream>: Send {
     /// Receives a `Request`/`Response` pair, and should perform some action on them.
     ///
     /// This could reading from the request, and writing to the response.
-    fn handle(self, Incoming);
+    fn handle(self, Incoming<A>);
 }
 
-impl Handler for fn(Incoming) {
+impl<A: NetworkAcceptor<S>, S: NetworkStream> Handler<A, S> for fn(Incoming<A>) {
-    fn handle(self, incoming: Incoming) {
+    fn handle(self, incoming: Incoming<A>) {
         (self)(incoming)
     }
 }

src/server/request.rs

@@ -4,7 +4,6 @@
 //! target URI, headers, and message body.
 use std::io::{Reader, BufferedReader, IoResult};
 use std::io::net::ip::SocketAddr;
-use std::io::net::tcp::TcpStream;
 
 use {HttpResult};
 use version::{HttpVersion};
@@ -13,9 +12,10 @@ use header::Headers;
 use header::common::ContentLength;
 use rfc7230::{read_request_line};
 use rfc7230::{HttpReader, SizedReader, ChunkedReader};
+use net::NetworkStream;
 use uri::RequestUri;
 
-/// A request bundles several parts of an incoming TCP stream, given to a `Handler`.
+/// A request bundles several parts of an incoming `NetworkStream`, given to a `Handler`.
 pub struct Request {
     /// The IP address of the remote connection.
     pub remote_addr: SocketAddr,
@@ -27,7 +27,7 @@ pub struct Request {
     pub uri: RequestUri,
     /// The version of HTTP for this request.
     pub version: HttpVersion,
-    body: HttpReader<BufferedReader<TcpStream>>
+    body: HttpReader<BufferedReader<Box<NetworkStream + Send>>>
 }
 
 
@@ -35,11 +35,11 @@ impl Request {
 
     /// Create a new Request, reading the StartLine and Headers so they are
     /// immediately useful.
-    pub fn new(mut tcp: TcpStream) -> HttpResult<Request> {
+    pub fn new<S: NetworkStream>(mut stream: S) -> HttpResult<Request> {
-        let remote_addr = try_io!(tcp.peer_name());
+        let remote_addr = try_io!(stream.peer_name());
-        let mut tcp = BufferedReader::new(tcp);
+        let mut stream = BufferedReader::new(stream.abstract());
-        let (method, uri, version) = try!(read_request_line(&mut tcp));
+        let (method, uri, version) = try!(read_request_line(&mut stream));
-        let mut headers = try!(Headers::from_raw(&mut tcp));
+        let mut headers = try!(Headers::from_raw(&mut stream));
 
         debug!("{} {} {}", method, uri, version);
         debug!("{}", headers);
@@ -47,12 +47,12 @@ impl Request {
 
         let body = if headers.has::<ContentLength>() {
             match headers.get_ref::<ContentLength>() {
-                Some(&ContentLength(len)) => SizedReader(tcp, len),
+                Some(&ContentLength(len)) => SizedReader(stream, len),
                 None => unreachable!()
             }
         } else {
             todo!("check for Transfer-Encoding: chunked");
-            ChunkedReader(tcp, None)
+            ChunkedReader(stream, None)
         };
 
         Ok(Request {
@@ -61,7 +61,7 @@ impl Request {
             uri: uri,
             headers: headers,
             version: version,
-            body: body,
+            body: body
         })
     }
 }

src/server/response.rs

@@ -3,15 +3,15 @@
 //! These are responses sent by a `hyper::Server` to clients, after
 //! receiving a request.
 use std::io::{BufferedWriter, IoResult};
-use std::io::net::tcp::TcpStream;
 
 use time::now_utc;
 
 use header;
 use header::common;
-use status;
-use version;
 use rfc7230::{CR, LF, LINE_ENDING};
+use status;
+use net::NetworkStream;
+use version;
 
 /// Phantom type indicating Headers and StatusCode have not been written.
 pub struct Fresh;
@@ -30,7 +30,7 @@ pub struct Response<W: WriteStatus> {
     /// The HTTP version of this response.
     pub version: version::HttpVersion,
     // Stream the Response is writing to, not accessible through UnwrittenResponse
-    body: BufferedWriter<TcpStream>, // TODO: use a HttpWriter from rfc7230
+    body: BufferedWriter<Box<NetworkStream + Send>>, // TODO: use a HttpWriter from rfc7230
     // The status code for the request.
     status: status::StatusCode,
     // The outgoing headers on this response.
@@ -47,7 +47,7 @@ impl<W: WriteStatus> Response<W> {
 
     /// Construct a Response from its constituent parts.
     pub fn construct(version: version::HttpVersion,
-                     body: BufferedWriter<TcpStream>,
+                     body: BufferedWriter<Box<NetworkStream + Send>>,
                      status: status::StatusCode,
                      headers: header::Headers) -> Response<Fresh> {
         Response {
@@ -61,12 +61,12 @@ impl<W: WriteStatus> Response<W> {
 
 impl Response<Fresh> {
     /// Creates a new Response that can be used to write to a network stream.
-    pub fn new(tcp: TcpStream) -> Response<Fresh> {
+    pub fn new<S: NetworkStream>(stream: S) -> Response<Fresh> {
         Response {
             status: status::Ok,
             version: version::Http11,
             headers: header::Headers::new(),
-            body: BufferedWriter::new(tcp)
+            body: BufferedWriter::new(stream.abstract())
         }
     }
 
@@ -104,7 +104,8 @@ impl Response<Fresh> {
     pub fn headers_mut(&mut self) -> &mut header::Headers { &mut self.headers }
 
     /// Deconstruct this Response into its constituent parts.
-    pub fn deconstruct(self) -> (version::HttpVersion, BufferedWriter<TcpStream>, status::StatusCode, header::Headers) {
+    pub fn deconstruct(self) -> (version::HttpVersion, BufferedWriter<Box<NetworkStream + Send>>,
+                                 status::StatusCode, header::Headers) {
         (self.version, self.body, self.status, self.headers)
     }
 }