abstract out NetworkStreams

Author: seanmonstar

benches/client.rs

@@ -5,19 +5,19 @@ extern crate hyper;
 extern crate test;
 
 use std::fmt::{mod, Show};
-use std::str::{SendStr, Slice};
-use std::io::IoResult;
 use std::io::net::ip::Ipv4Addr;
-use hyper::server::{Request, Response, Server};
+use hyper::server::{Server, Incoming};
 
 fn listen() -> hyper::server::Listening {
     let server = Server::http(Ipv4Addr(127, 0, 0, 1), 0);
     server.listen(handle).unwrap()
 }
 
-fn handle(_req: Request, mut res: Response) -> IoResult<()> {
-    try!(res.write(b"Benchmarking hyper vs others!"));
-    res.end()
+fn handle(mut incoming: Incoming) {
+    for (_, mut res) in incoming {
+        res.write(b"Benchmarking hyper vs others!").unwrap();
+        res.end().unwrap();
+    }
 }
 
 

examples/server.rs

@@ -7,11 +7,9 @@ use std::io::util::copy;
 use std::io::net::ip::Ipv4Addr;
 
 use hyper::{Get, Post};
-use hyper::server::{Server, Handler, Incoming};
+use hyper::server::{Server, Incoming};
 use hyper::header::ContentLength;
 
-struct Echo;
-
 macro_rules! try_continue(
     ($e:expr) => {{
         match $e {
@@ -21,39 +19,37 @@ 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.set(ContentLength(out.len()));
-                        try_continue!(res.write(out));
-                        try_continue!(res.end());
-                        continue;
-                    },
-                    (&Post, "/echo") => (), // fall through, fighting mutable borrows
-                    _ => {
-                        res.status = hyper::status::NotFound;
-                        try_continue!(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.set(ContentLength(out.len()));
+                    try_continue!(res.write(out));
+                    try_continue!(res.end());
+                    continue;
                 },
+                (&Post, "/echo") => (), // fall through, fighting mutable borrows
                 _ => {
+                    res.status = hyper::status::NotFound;
                     try_continue!(res.end());
-                    continue; 
+                    continue;
                 }
-            };
-
-            try_continue!(copy(&mut req, &mut res));
-            try_continue!(res.end());
-        }
+            },
+            _ => {
+                try_continue!(res.end());
+                continue; 
+            }
+        };
+
+        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/lib.rs

@@ -1,7 +1,7 @@
 //! # hyper
-#![feature(macro_rules, phase)]
+#![feature(macro_rules, phase, default_type_params)]
 #![warn(missing_doc)]
-#![deny(warnings)]
+//#![deny(warnings)]
 #![experimental]
 
 extern crate time;
@@ -47,6 +47,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,110 @@
+//! 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 {
+    /// Get the remote address of the underlying connection.
+    fn peer_name(&mut self) -> IoResult<SocketAddr>;
+}
+
+/// 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()
+    }
+}

src/server/mod.rs

@@ -1,44 +1,55 @@
 //! 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;
 
+use net::{NetworkListener, NetworkAcceptor, NetworkStream};
+use net::{HttpListener, HttpAcceptor};
+
 pub mod request;
 pub mod response;
 
 /// A server can listen on a TCP socket.
 ///
 /// 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> {
+    /// Creates a new server that will handle `HttpStream`s.
+    pub fn http(ip: IpAddr, port: Port) -> Server {
+        Server {
+            ip: ip,
+            port: port
+        }
+    }
+}
 
-impl Server {
+impl<L: NetworkListener<S, A>, S: NetworkStream, A: NetworkAcceptor<S>> Server<L> {
 
-    /// Creates a server to be used for `http` conenctions.
-    pub fn http(ip: IpAddr, port: Port) -> Server {
+    /// 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> {
-        let mut listener = try!(TcpListener::bind(self.ip.to_string().as_slice(), self.port));
+    pub fn listen<H: Handler<A, S> + 'static>(self, handler: H) -> IoResult<Listening<A>> {
+        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: acceptor.incoming() });
+            handler.handle(Incoming { from: worker.incoming() });
         });
 
         Ok(Listening {
@@ -51,12 +62,12 @@ impl Server {
 
 /// An iterator over incoming connections, represented as pairs of
 /// hyper Requests and Responses.
-pub struct Incoming<'a> {
-    from: IncomingConnections<'a, TcpAcceptor>
+pub struct Incoming<'a, A: 'a = HttpAcceptor> {
+    from: IncomingConnections<'a, A>
 }
 
-impl<'a> Iterator<(Request, Response)> for Incoming<'a> {
-    fn next(&mut self) -> Option<(Request, Response)> {
+impl<'a, A: NetworkAcceptor<S>, S: NetworkStream> Iterator<(Request<S>, Response<S>)> for Incoming<'a, A> {
+    fn next(&mut self) -> Option<(Request<S>, Response<S>)> {
         for conn in self.from {
             match conn {
                 Ok(stream) => {
@@ -85,30 +96,30 @@ impl<'a> Iterator<(Request, Response)> for Incoming<'a> {
 }
 
 /// A listening server, which can later be closed.
-pub struct Listening {
-    acceptor: TcpAcceptor,
+pub struct Listening<A = HttpAcceptor> {
+    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 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) {
-    fn handle(self, incoming: Incoming) {
+impl<A: NetworkAcceptor<S>, S: NetworkStream> Handler<A, S> for fn(Incoming<A>) {
+    fn handle(self, incoming: Incoming<A>) {
         (self)(incoming)
     }
 }