From 58b29059108b1b9f9ca485d61c105c08f5588bd9 Mon Sep 17 00:00:00 2001 From: Sean McArthur Date: Sun, 7 Sep 2014 14:18:51 -0700 Subject: [PATCH 1/2] abstract out NetworkStreams --- examples/server.rs | 54 ++++++++++---------- src/lib.rs | 5 +- src/net.rs | 110 +++++++++++++++++++++++++++++++++++++++++ src/server/mod.rs | 57 ++++++++++++--------- src/server/request.rs | 26 +++++----- src/server/response.rs | 18 +++---- 6 files changed, 194 insertions(+), 76 deletions(-) create mode 100644 src/net.rs diff --git a/examples/server.rs b/examples/server.rs index 30e24969d1..a75b42a366 100644 --- a/examples/server.rs +++ b/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(); } diff --git a/src/lib.rs b/src/lib.rs index 5beb9b8cbd..44a052c86d 100644 --- a/src/lib.rs +++ b/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; @@ -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; diff --git a/src/net.rs b/src/net.rs new file mode 100644 index 0000000000..26b633377a --- /dev/null +++ b/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>: Listener { + /// 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; + + /// Get the address this Listener ended up listening on. + fn socket_name(&mut self) -> IoResult; +} + +/// An abstraction to receive `HttpStream`s. +pub trait NetworkAcceptor: Acceptor + 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; +} + +/// A `NetworkListener` for `HttpStream`s. +pub struct HttpListener { + inner: TcpListener +} + +impl Listener for HttpListener { + #[inline] + fn listen(self) -> IoResult { + Ok(HttpAcceptor { + inner: try!(self.inner.listen()) + }) + } +} + +impl NetworkListener for HttpListener { + #[inline] + fn bind(host: &str, port: Port) -> IoResult { + Ok(HttpListener { + inner: try!(TcpListener::bind(host, port)) + }) + } + + #[inline] + fn socket_name(&mut self) -> IoResult { + self.inner.socket_name() + } +} + +/// A `NetworkAcceptor` for `HttpStream`s. +#[deriving(Clone)] +pub struct HttpAcceptor { + inner: TcpAcceptor +} + +impl Acceptor for HttpAcceptor { + #[inline] + fn accept(&mut self) -> IoResult { + Ok(HttpStream { + inner: try!(self.inner.accept()) + }) + } +} + +impl NetworkAcceptor 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 { + 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 { + self.inner.peer_name() + } +} diff --git a/src/server/mod.rs b/src/server/mod.rs index 6f4d0816e3..351891b724 100644 --- a/src/server/mod.rs +++ b/src/server/mod.rs @@ -1,11 +1,13 @@ //! 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; @@ -13,32 +15,41 @@ 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 { ip: IpAddr, port: Port } +impl Server { + /// 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, S: NetworkStream, A: NetworkAcceptor> Server { - /// 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 { Server { ip: ip, port: port } } + /// Binds to a socket, and starts handling connections. - pub fn listen(self, handler: H) -> IoResult { - let mut listener = try!(TcpListener::bind(self.ip.to_string().as_slice(), self.port)); + pub fn listen + 'static>(self, handler: H) -> IoResult> { + 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: NetworkStream> Iterator<(Request, Response)> for Incoming<'a, A> { + fn next(&mut self) -> Option<(Request, Response)> { 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 { + acceptor: A, /// The socket address that the server is bound to. pub socket_addr: SocketAddr, } -impl Listening { - /// Stop the server from listening to its socket address. +impl, S: NetworkStream> Listening { + /// 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, 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); } -impl Handler for fn(Incoming) { - fn handle(self, incoming: Incoming) { +impl, S: NetworkStream> Handler for fn(Incoming) { + fn handle(self, incoming: Incoming) { (self)(incoming) } } diff --git a/src/server/request.rs b/src/server/request.rs index e6e9a30bbc..b5c5d40e67 100644 --- a/src/server/request.rs +++ b/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}; @@ -12,10 +11,11 @@ use method; use header::{Headers, ContentLength}; use rfc7230::{read_request_line}; use rfc7230::{HttpReader, SizedReader, ChunkedReader}; +use net::{NetworkStream, HttpStream}; use uri::RequestUri; -/// A request bundles several parts of an incoming TCP stream, given to a `Handler`. -pub struct Request { +/// 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, /// The `Method`, such as `Get`, `Post`, etc. @@ -26,19 +26,19 @@ pub struct Request { pub uri: RequestUri, /// The version of HTTP for this request. pub version: HttpVersion, - body: HttpReader> + body: HttpReader> } -impl Request { +impl Request { /// Create a new Request, reading the StartLine and Headers so they are /// immediately useful. - pub fn new(mut tcp: TcpStream) -> HttpResult { - let remote_addr = try_io!(tcp.peer_name()); - let mut tcp = BufferedReader::new(tcp); - let (method, uri, version) = try!(read_request_line(&mut tcp)); - let mut headers = try!(Headers::from_raw(&mut tcp)); + pub fn new(mut stream: S) -> HttpResult> { + let remote_addr = try_io!(stream.peer_name()); + let mut stream = BufferedReader::new(stream); + let (method, uri, version) = try!(read_request_line(&mut stream)); + let mut headers = try!(Headers::from_raw(&mut stream)); debug!("{} {} {}", method, uri, version); debug!("{}", headers); @@ -46,12 +46,12 @@ impl Request { let body = if headers.has::() { match headers.get_ref::() { - 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 { @@ -65,7 +65,7 @@ impl Request { } } -impl Reader for Request { +impl Reader for Request { fn read(&mut self, buf: &mut [u8]) -> IoResult { self.body.read(buf) } diff --git a/src/server/response.rs b/src/server/response.rs index 4658dd9c72..000cb0956b 100644 --- a/src/server/response.rs +++ b/src/server/response.rs @@ -3,18 +3,18 @@ //! 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 rfc7230::{CR, LF, LINE_ENDING}; use status; +use net::{NetworkStream, HttpStream}; use version; -use rfc7230::{CR, LF, LINE_ENDING}; -/// The outgoing half for a Tcp connection, created by a `Server` and given to a `Handler`. -pub struct Response { +/// The outgoing half for a `NetworkStream`, created by a `Server` and given to a `Handler`. +pub struct Response { /// The status code for the request. pub status: status::StatusCode, /// The outgoing headers on this response. @@ -23,19 +23,19 @@ pub struct Response { pub version: version::HttpVersion, headers_written: bool, // TODO: can this check be moved to compile time? - body: BufferedWriter, // TODO: use a HttpWriter from rfc7230 + body: BufferedWriter, // TODO: use a HttpWriter from rfc7230 } -impl Response { +impl Response { /// Creates a new Response that can be used to write to a network stream. - pub fn new(tcp: TcpStream) -> Response { + pub fn new(stream: S) -> Response { Response { status: status::Ok, version: version::Http11, headers: header::Headers::new(), headers_written: false, - body: BufferedWriter::new(tcp) + body: BufferedWriter::new(stream) } } @@ -69,7 +69,7 @@ impl Response { } -impl Writer for Response { +impl Writer for Response { fn write(&mut self, msg: &[u8]) -> IoResult<()> { if !self.headers_written { try!(self.write_head()); From 3f09a2291f60e5e379d959ce14c80f9fb43e9817 Mon Sep 17 00:00:00 2001 From: Sean McArthur Date: Sun, 7 Sep 2014 14:37:29 -0700 Subject: [PATCH 2/2] use NetworkStream in client --- src/client/request.rs | 19 +++++++++---------- src/client/response.rs | 29 +++++++++++++++-------------- src/net.rs | 10 ++++++++++ 3 files changed, 34 insertions(+), 24 deletions(-) diff --git a/src/client/request.rs b/src/client/request.rs index 562beaa793..df37dcfa75 100644 --- a/src/client/request.rs +++ b/src/client/request.rs @@ -1,10 +1,10 @@ //! Client Requests -use std::io::net::tcp::TcpStream; use std::io::{BufferedWriter, IoResult}; use url::Url; use method; +use net::{NetworkStream, HttpStream}; use header::{Headers, Host}; use rfc7230::LINE_ENDING; use version; @@ -13,7 +13,7 @@ use super::{Response}; /// A client request to a remote server. -pub struct Request { +pub struct Request { /// The method of this request. pub method: method::Method, /// The headers that will be sent with this request. @@ -23,13 +23,13 @@ pub struct Request { /// The HTTP version of this request. pub version: version::HttpVersion, headers_written: bool, - body: BufferedWriter, + body: BufferedWriter, } -impl Request { +impl Request { /// Create a new client request. - pub fn new(method: method::Method, url: Url) -> HttpResult { + pub fn new(method: method::Method, url: Url) -> HttpResult> { debug!("{} {}", method, url); let host = match url.serialize_host() { Some(host) => host, @@ -42,7 +42,7 @@ impl Request { }; debug!("port={}", port); - let stream = try_io!(TcpStream::connect(host.as_slice(), port)); + let stream = try_io!(NetworkStream::connect(host.as_slice(), port)); let stream = BufferedWriter::new(stream); let mut headers = Headers::new(); headers.set(Host(host)); @@ -81,16 +81,15 @@ impl Request { /// Completes writing the request, and returns a response to read from. /// /// Consumes the Request. - pub fn send(mut self) -> HttpResult { + pub fn send(mut self) -> HttpResult> { try_io!(self.flush()); - let mut raw = self.body.unwrap(); - try_io!(raw.close_write()); + let raw = self.body.unwrap(); Response::new(raw) } } -impl Writer for Request { +impl Writer for Request { fn write(&mut self, msg: &[u8]) -> IoResult<()> { if !self.headers_written { try!(self.write_head()); diff --git a/src/client/response.rs b/src/client/response.rs index 538c1247c6..027877422f 100644 --- a/src/client/response.rs +++ b/src/client/response.rs @@ -1,31 +1,31 @@ //! Client Responses use std::io::{BufferedReader, IoResult}; -use std::io::net::tcp::TcpStream; use header::{mod, ContentLength, TransferEncoding, 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 { /// 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>, + body: HttpReader>, } -impl Response { +impl Response { /// Creates a new response from a server. - pub fn new(tcp: TcpStream) -> HttpResult { - 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: S) -> HttpResult> { + let mut stream = BufferedReader::new(stream); + let (version, status) = try!(read_status_line(&mut stream)); + let mut headers = try!(header::Headers::from_raw(&mut stream)); debug!("{} {}", version, status); debug!("{}", headers); @@ -38,22 +38,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::() { match headers.get_ref::() { - 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 { @@ -65,7 +65,8 @@ impl Response { } } -impl Reader for Response { +impl Reader for Response { + #[inline] fn read(&mut self, buf: &mut [u8]) -> IoResult { self.body.read(buf) } diff --git a/src/net.rs b/src/net.rs index 26b633377a..44d6e6b71f 100644 --- a/src/net.rs +++ b/src/net.rs @@ -25,6 +25,9 @@ pub trait NetworkAcceptor: Acceptor + Clone + Send { pub trait NetworkStream: Stream + Clone { /// Get the remote address of the underlying connection. fn peer_name(&mut self) -> IoResult; + + /// Connect to a remote address. + fn connect(host: &str, port: Port) -> IoResult; } /// A `NetworkListener` for `HttpStream`s. @@ -107,4 +110,11 @@ impl NetworkStream for HttpStream { fn peer_name(&mut self) -> IoResult { self.inner.peer_name() } + + #[inline] + fn connect(host: &str, port: Port) -> IoResult { + Ok(HttpStream { + inner: try!(TcpStream::connect(host, port)) + }) + } }