forked/hyper
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(hyper): switch to std::io, std::net, and std::path.

Browse Source 
All instances of `old_io` and `old_path` were switched to use the new
shiny `std::io`, `std::net`, and `std::path` modules. This means that
`Request` and `Response` implement `Read` and `Write` now.

Because of the changes to `TcpListener`, this also takes the opportunity
to correct the method usage of `Server`. As with other
languages/frameworks, the server is first created with a handler, and
then a host/port is passed to a `listen` method. This reverses what
`Server` used to do.

Closes #347

BREAKING CHANGE: Check the docs. Everything was touched.
This commit is contained in:
Sean McArthur
2015-02-17 15:29:52 -08:00
parent 7235d3f74a
commit 0fd6fcd7c7
22 changed files with 641 additions and 639 deletions
Download Patch File Download Diff File Expand all files Collapse all files

33
src/server/acceptor.rs → src/server/listener.rs
View File

@@ -1,16 +1,16 @@
use std::thread::{self, JoinGuard};
use std::sync::mpsc;
use std::collections::VecMap;
use net::NetworkAcceptor;
use net::NetworkListener;
pub struct AcceptorPool<A: NetworkAcceptor> {
pub struct ListenerPool<A: NetworkListener> {
acceptor: A
}
impl<'a, A: NetworkAcceptor + 'a> AcceptorPool<A> {
impl<'a, A: NetworkListener + Send + 'a> ListenerPool<A> {
/// Create a thread pool to manage the acceptor.
pub fn new(acceptor: A) -> AcceptorPool<A> {
AcceptorPool { acceptor: acceptor }
pub fn new(acceptor: A) -> ListenerPool<A> {
ListenerPool { acceptor: acceptor }
}
/// Runs the acceptor pool. Blocks until the acceptors are closed.
@@ -44,23 +44,16 @@ impl<'a, A: NetworkAcceptor + 'a> AcceptorPool<A> {
}
}
fn spawn_with<'a, A, F>(supervisor: mpsc::Sender<usize>, work: &'a F, mut acceptor: A, id: usize) -> JoinGuard<'a, ()>
where A: NetworkAcceptor + 'a,
F: Fn(<A as NetworkAcceptor>::Stream) + Send + Sync + 'a {
use std::old_io::EndOfFile;
fn spawn_with<'a, A, F>(supervisor: mpsc::Sender<usize>, work: &'a F, mut acceptor: A, id: usize) -> thread::JoinGuard<'a, ()>
where A: NetworkListener + Send + 'a,
F: Fn(<A as NetworkListener>::Stream) + Send + Sync + 'a {
thread::scoped(move || {
let sentinel = Sentinel::new(supervisor, id);
let _sentinel = Sentinel::new(supervisor, id);
loop {
match acceptor.accept() {
Ok(stream) => work(stream),
Err(ref e) if e.kind == EndOfFile => {
debug!("Server closed.");
sentinel.cancel();
return;
},
Err(e) => {
error!("Connection failed: {}", e);
}
@@ -72,7 +65,7 @@ where A: NetworkAcceptor + 'a,
struct Sentinel<T: Send> {
value: Option<T>,
supervisor: mpsc::Sender<T>,
active: bool
//active: bool
}
impl<T: Send> Sentinel<T> {
@@ -80,18 +73,18 @@ impl<T: Send> Sentinel<T> {
Sentinel {
value: Some(data),
supervisor: channel,
active: true
//active: true
}
}
fn cancel(mut self) { self.active = false; }
//fn cancel(mut self) { self.active = false; }
}
#[unsafe_destructor]
impl<T: Send + 'static> Drop for Sentinel<T> {
fn drop(&mut self) {
// If we were cancelled, get out of here.
if !self.active { return; }
//if !self.active { return; }
// Respawn ourselves
let _ = self.supervisor.send(self.value.take().unwrap());

105
src/server/mod.rs
View File

@@ -1,7 +1,9 @@
//! HTTP Server
use std::old_io::{Listener, BufferedReader, BufferedWriter};
use std::old_io::net::ip::{IpAddr, Port, SocketAddr};
use std::io::{BufReader, BufWriter};
use std::marker::PhantomData;
use std::net::{IpAddr, SocketAddr};
use std::os;
use std::path::Path;
use std::thread::{self, JoinGuard};
pub use self::request::Request;
@@ -13,25 +15,24 @@ use HttpError::HttpIoError;
use {HttpResult};
use header::Connection;
use header::ConnectionOption::{Close, KeepAlive};
use net::{NetworkListener, NetworkStream, NetworkAcceptor,
HttpAcceptor, HttpListener};
use net::{NetworkListener, NetworkStream, HttpListener};
use version::HttpVersion::{Http10, Http11};
use self::acceptor::AcceptorPool;
use self::listener::ListenerPool;
pub mod request;
pub mod response;
mod acceptor;
mod listener;
/// 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<L = HttpListener> {
ip: IpAddr,
port: Port,
listener: L,
pub struct Server<'a, H: Handler, L = HttpListener> {
handler: H,
ssl: Option<(&'a Path, &'a Path)>,
_marker: PhantomData<L>
}
macro_rules! try_option(
@@ -43,38 +44,59 @@ macro_rules! try_option(
}}
);
impl Server<HttpListener> {
/// Creates a new server that will handle `HttpStream`s.
pub fn http(ip: IpAddr, port: Port) -> Server {
Server::with_listener(ip, port, HttpListener::Http)
}
/// Creates a new server that will handler `HttpStreams`s using a TLS connection.
pub fn https(ip: IpAddr, port: Port, cert: Path, key: Path) -> Server {
Server::with_listener(ip, port, HttpListener::Https(cert, key))
impl<'a, H: Handler, L: NetworkListener> Server<'a, H, L> {
pub fn new(handler: H) -> Server<'a, H, L> {
Server {
handler: handler,
ssl: None,
_marker: PhantomData
}
}
}
impl<
L: NetworkListener<Acceptor=A> + Send,
A: NetworkAcceptor<Stream=S> + Send + 'static,
S: NetworkStream + Clone + Send> Server<L> {
impl<'a, H: Handler + 'static> Server<'a, H, HttpListener> {
/// Creates a new server that will handle `HttpStream`s.
pub fn with_listener(ip: IpAddr, port: Port, listener: L) -> Server<L> {
pub fn http(handler: H) -> Server<'a, H, HttpListener> {
Server::new(handler)
}
/// Creates a new server that will handler `HttpStreams`s using a TLS connection.
pub fn https(handler: H, cert: &'a Path, key: &'a Path) -> Server<'a, H, HttpListener> {
Server {
ip: ip,
port: port,
listener: listener,
handler: handler,
ssl: Some((cert, key)),
_marker: PhantomData
}
}
}
impl<'a, H: Handler + 'static> Server<'a, H, HttpListener> {
/// Binds to a socket, and starts handling connections using a task pool.
pub fn listen_threads<H: Handler + 'static>(mut self, handler: H, threads: usize) -> HttpResult<Listening<L::Acceptor>> {
debug!("binding to {:?}:{:?}", self.ip, self.port);
let acceptor = try!(self.listener.listen((self.ip, self.port)));
let socket = try!(acceptor.socket_name());
pub fn listen_threads(self, ip: IpAddr, port: u16, threads: usize) -> HttpResult<Listening> {
let addr = &(ip, port);
let listener = try!(match self.ssl {
Some((cert, key)) => HttpListener::https(addr, cert, key),
None => HttpListener::http(addr)
});
self.with_listener(listener, threads)
}
/// Binds to a socket and starts handling connections.
pub fn listen(self, ip: IpAddr, port: u16) -> HttpResult<Listening> {
self.listen_threads(ip, port, os::num_cpus() * 5 / 4)
}
}
impl<
'a,
H: Handler + 'static,
L: NetworkListener<Stream=S> + Send + 'static,
S: NetworkStream + Clone + Send> Server<'a, H, L> {
/// Creates a new server that will handle `HttpStream`s.
pub fn with_listener(self, mut listener: L, threads: usize) -> HttpResult<Listening> {
let socket = try!(listener.socket_addr());
let handler = self.handler;
debug!("threads = {:?}", threads);
let pool = AcceptorPool::new(acceptor.clone());
let pool = ListenerPool::new(listener.clone());
let work = move |stream| handle_connection(stream, &handler);
let guard = thread::scoped(move || pool.accept(work, threads));
@@ -82,21 +104,15 @@ S: NetworkStream + Clone + Send> Server<L> {
Ok(Listening {
_guard: guard,
socket: socket,
acceptor: acceptor
})
}
/// Binds to a socket and starts handling connections.
pub fn listen<H: Handler + 'static>(self, handler: H) -> HttpResult<Listening<L::Acceptor>> {
self.listen_threads(handler, os::num_cpus() * 5 / 4)
}
}
fn handle_connection<S, H>(mut stream: S, handler: &H)
where S: NetworkStream + Clone, H: Handler {
debug!("Incoming stream");
let addr = match stream.peer_name() {
let addr = match stream.peer_addr() {
Ok(addr) => addr,
Err(e) => {
error!("Peer Name error: {:?}", e);
@@ -104,8 +120,8 @@ where S: NetworkStream + Clone, H: Handler {
}
};
let mut rdr = BufferedReader::new(stream.clone());
let mut wrt = BufferedWriter::new(stream);
let mut rdr = BufReader::new(stream.clone());
let mut wrt = BufWriter::new(stream);
let mut keep_alive = true;
while keep_alive {
@@ -135,18 +151,17 @@ where S: NetworkStream + Clone, H: Handler {
}
/// A listening server, which can later be closed.
pub struct Listening<A = HttpAcceptor> {
acceptor: A,
pub struct Listening {
_guard: JoinGuard<'static, ()>,
/// The socket addresses that the server is bound to.
pub socket: SocketAddr,
}
impl<A: NetworkAcceptor> Listening<A> {
impl Listening {
/// Stop the server from listening to its socket address.
pub fn close(&mut self) -> HttpResult<()> {
debug!("closing server");
try!(self.acceptor.close());
//try!(self.acceptor.close());
Ok(())
}
}

52
src/server/request.rs
View File

@@ -2,8 +2,8 @@
//!
//! These are requests that a `hyper::Server` receives, and include its method,
//! target URI, headers, and message body.
use std::old_io::IoResult;
use std::old_io::net::ip::SocketAddr;
use std::io::{self, Read};
use std::net::SocketAddr;
use {HttpResult};
use version::{HttpVersion};
@@ -26,14 +26,14 @@ pub struct Request<'a> {
pub uri: RequestUri,
/// The version of HTTP for this request.
pub version: HttpVersion,
body: HttpReader<&'a mut (Reader + 'a)>
body: HttpReader<&'a mut (Read + 'a)>
}
impl<'a> Request<'a> {
/// Create a new Request, reading the StartLine and Headers so they are
/// immediately useful.
pub fn new(mut stream: &'a mut (Reader + 'a), addr: SocketAddr) -> HttpResult<Request<'a>> {
pub fn new(mut stream: &'a mut (Read + 'a), addr: SocketAddr) -> HttpResult<Request<'a>> {
let (method, uri, version) = try!(read_request_line(&mut stream));
debug!("Request Line: {:?} {:?} {:?}", method, uri, version);
let headers = try!(Headers::from_raw(&mut stream));
@@ -66,14 +66,14 @@ impl<'a> Request<'a> {
/// Deconstruct a Request into its constituent parts.
pub fn deconstruct(self) -> (SocketAddr, Method, Headers,
RequestUri, HttpVersion,
HttpReader<&'a mut (Reader + 'a)>,) {
HttpReader<&'a mut (Read + 'a)>,) {
(self.remote_addr, self.method, self.headers,
self.uri, self.version, self.body)
}
}
impl<'a> Reader for Request<'a> {
fn read(&mut self, buf: &mut [u8]) -> IoResult<usize> {
impl<'a> Read for Request<'a> {
fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
self.body.read(buf)
}
}
@@ -84,12 +84,19 @@ mod tests {
use mock::MockStream;
use super::Request;
use std::old_io::net::ip::SocketAddr;
use std::io::{self, Read};
use std::net::SocketAddr;
fn sock(s: &str) -> SocketAddr {
s.parse().unwrap()
}
fn read_to_string(mut req: Request) -> io::Result<String> {
let mut s = String::new();
try!(req.read_to_string(&mut s));
Ok(s)
}
#[test]
fn test_get_empty_body() {
let mut stream = MockStream::with_input(b"\
@@ -99,8 +106,8 @@ mod tests {
I'm a bad request.\r\n\
");
let mut req = Request::new(&mut stream, sock("127.0.0.1:80")).unwrap();
assert_eq!(req.read_to_string(), Ok("".to_string()));
let req = Request::new(&mut stream, sock("127.0.0.1:80")).unwrap();
assert_eq!(read_to_string(req), Ok("".to_string()));
}
#[test]
@@ -112,8 +119,8 @@ mod tests {
I'm a bad request.\r\n\
");
let mut req = Request::new(&mut stream, sock("127.0.0.1:80")).unwrap();
assert_eq!(req.read_to_string(), Ok("".to_string()));
let req = Request::new(&mut stream, sock("127.0.0.1:80")).unwrap();
assert_eq!(read_to_string(req), Ok("".to_string()));
}
#[test]
@@ -125,8 +132,8 @@ mod tests {
I'm a bad request.\r\n\
");
let mut req = Request::new(&mut stream, sock("127.0.0.1:80")).unwrap();
assert_eq!(req.read_to_string(), Ok("".to_string()));
let req = Request::new(&mut stream, sock("127.0.0.1:80")).unwrap();
assert_eq!(read_to_string(req), Ok("".to_string()));
}
#[test]
@@ -146,7 +153,7 @@ mod tests {
\r\n"
);
let mut req = Request::new(&mut stream, sock("127.0.0.1:80")).unwrap();
let req = Request::new(&mut stream, sock("127.0.0.1:80")).unwrap();
// The headers are correct?
match req.headers.get::<Host>() {
@@ -163,8 +170,7 @@ mod tests {
None => panic!("Transfer-Encoding: chunked expected!"),
};
// The content is correctly read?
let body = req.read_to_string().unwrap();
assert_eq!("qwert", body);
assert_eq!(read_to_string(req), Ok("qwert".to_string()));
}
/// Tests that when a chunk size is not a valid radix-16 number, an error
@@ -182,9 +188,9 @@ mod tests {
\r\n"
);
let mut req = Request::new(&mut stream, sock("127.0.0.1:80")).unwrap();
let req = Request::new(&mut stream, sock("127.0.0.1:80")).unwrap();
assert!(req.read_to_string().is_err());
assert!(read_to_string(req).is_err());
}
/// Tests that when a chunk size contains an invalid extension, an error is
@@ -202,9 +208,9 @@ mod tests {
\r\n"
);
let mut req = Request::new(&mut stream, sock("127.0.0.1:80")).unwrap();
let req = Request::new(&mut stream, sock("127.0.0.1:80")).unwrap();
assert!(req.read_to_string().is_err());
assert!(read_to_string(req).is_err());
}
/// Tests that when a valid extension that contains a digit is appended to
@@ -222,9 +228,9 @@ mod tests {
\r\n"
);
let mut req = Request::new(&mut stream, sock("127.0.0.1:80")).unwrap();
let req = Request::new(&mut stream, sock("127.0.0.1:80")).unwrap();
assert_eq!("1", req.read_to_string().unwrap())
assert_eq!(read_to_string(req), Ok("1".to_string()));
}
}

29
src/server/response.rs
View File

@@ -2,8 +2,8 @@
//!
//! These are responses sent by a `hyper::Server` to clients, after
//! receiving a request.
use std::old_io::IoResult;
use std::marker::PhantomData;
use std::io::{self, Write};
use time::now_utc;
@@ -19,7 +19,7 @@ pub struct Response<'a, W = Fresh> {
/// The HTTP version of this response.
pub version: version::HttpVersion,
// Stream the Response is writing to, not accessible through UnwrittenResponse
body: HttpWriter<&'a mut (Writer + 'a)>,
body: HttpWriter<&'a mut (Write + 'a)>,
// The status code for the request.
status: status::StatusCode,
// The outgoing headers on this response.
@@ -38,7 +38,7 @@ impl<'a, W> Response<'a, W> {
/// Construct a Response from its constituent parts.
pub fn construct(version: version::HttpVersion,
body: HttpWriter<&'a mut (Writer + 'a)>,
body: HttpWriter<&'a mut (Write + 'a)>,
status: status::StatusCode,
headers: header::Headers) -> Response<'a, Fresh> {
Response {
@@ -51,7 +51,7 @@ impl<'a, W> Response<'a, W> {
}
/// Deconstruct this Response into its constituent parts.
pub fn deconstruct(self) -> (version::HttpVersion, HttpWriter<&'a mut (Writer + 'a)>,
pub fn deconstruct(self) -> (version::HttpVersion, HttpWriter<&'a mut (Write + 'a)>,
status::StatusCode, header::Headers) {
(self.version, self.body, self.status, self.headers)
}
@@ -59,7 +59,7 @@ impl<'a, W> Response<'a, W> {
impl<'a> Response<'a, Fresh> {
/// Creates a new Response that can be used to write to a network stream.
pub fn new(stream: &'a mut (Writer + 'a)) -> Response<'a, Fresh> {
pub fn new(stream: &'a mut (Write + 'a)) -> Response<'a, Fresh> {
Response {
status: status::StatusCode::Ok,
version: version::HttpVersion::Http11,
@@ -70,7 +70,7 @@ impl<'a> Response<'a, Fresh> {
}
/// Consume this Response<Fresh>, writing the Headers and Status and creating a Response<Streaming>
pub fn start(mut self) -> IoResult<Response<'a, Streaming>> {
pub fn start(mut self) -> io::Result<Response<'a, Streaming>> {
debug!("writing head: {:?} {:?}", self.version, self.status);
try!(write!(&mut self.body, "{} {}{}{}", self.version, self.status, CR as char, LF as char));
@@ -110,13 +110,12 @@ impl<'a> Response<'a, Fresh> {
debug!("headers [\n{:?}]", self.headers);
try!(write!(&mut self.body, "{}", self.headers));
try!(self.body.write_str(LINE_ENDING));
try!(write!(&mut self.body, "{}", LINE_ENDING));
let stream = if chunked {
ChunkedWriter(self.body.unwrap())
ChunkedWriter(self.body.into_inner())
} else {
SizedWriter(self.body.unwrap(), len)
SizedWriter(self.body.into_inner(), len)
};
// "copy" to change the phantom type
@@ -139,20 +138,20 @@ impl<'a> Response<'a, Fresh> {
impl<'a> Response<'a, Streaming> {
/// Flushes all writing of a response to the client.
pub fn end(self) -> IoResult<()> {
pub fn end(self) -> io::Result<()> {
debug!("ending");
try!(self.body.end());
Ok(())
}
}
impl<'a> Writer for Response<'a, Streaming> {
fn write_all(&mut self, msg: &[u8]) -> IoResult<()> {
impl<'a> Write for Response<'a, Streaming> {
fn write(&mut self, msg: &[u8]) -> io::Result<usize> {
debug!("write {:?} bytes", msg.len());
self.body.write_all(msg)
self.body.write(msg)
}
fn flush(&mut self) -> IoResult<()> {
fn flush(&mut self) -> io::Result<()> {
self.body.flush()
}
}