feat(server): Add hooks for HttpListener and HttpsListener to be started from existing listeners.
This allows Servers to be started on existing TcpListeners.
This commit is contained in:
37
src/net.rs
37
src/net.rs
@@ -51,12 +51,15 @@ impl<'a, N: NetworkListener + 'a> Iterator for NetworkConnections<'a, N> {
|
|||||||
pub trait NetworkStream: Read + Write + Any + Send + Typeable {
|
pub trait NetworkStream: Read + Write + Any + Send + Typeable {
|
||||||
/// Get the remote address of the underlying connection.
|
/// Get the remote address of the underlying connection.
|
||||||
fn peer_addr(&mut self) -> io::Result<SocketAddr>;
|
fn peer_addr(&mut self) -> io::Result<SocketAddr>;
|
||||||
|
|
||||||
/// Set the maximum time to wait for a read to complete.
|
/// Set the maximum time to wait for a read to complete.
|
||||||
#[cfg(feature = "timeouts")]
|
#[cfg(feature = "timeouts")]
|
||||||
fn set_read_timeout(&self, dur: Option<Duration>) -> io::Result<()>;
|
fn set_read_timeout(&self, dur: Option<Duration>) -> io::Result<()>;
|
||||||
|
|
||||||
/// Set the maximum time to wait for a write to complete.
|
/// Set the maximum time to wait for a write to complete.
|
||||||
#[cfg(feature = "timeouts")]
|
#[cfg(feature = "timeouts")]
|
||||||
fn set_write_timeout(&self, dur: Option<Duration>) -> io::Result<()>;
|
fn set_write_timeout(&self, dur: Option<Duration>) -> io::Result<()>;
|
||||||
|
|
||||||
/// This will be called when Stream should no longer be kept alive.
|
/// This will be called when Stream should no longer be kept alive.
|
||||||
#[inline]
|
#[inline]
|
||||||
fn close(&mut self, _how: Shutdown) -> io::Result<()> {
|
fn close(&mut self, _how: Shutdown) -> io::Result<()> {
|
||||||
@@ -66,9 +69,8 @@ pub trait NetworkStream: Read + Write + Any + Send + Typeable {
|
|||||||
// Unsure about name and implementation...
|
// Unsure about name and implementation...
|
||||||
|
|
||||||
#[doc(hidden)]
|
#[doc(hidden)]
|
||||||
fn set_previous_response_expected_no_content(&mut self, _expected: bool) {
|
fn set_previous_response_expected_no_content(&mut self, _expected: bool) { }
|
||||||
|
|
||||||
}
|
|
||||||
#[doc(hidden)]
|
#[doc(hidden)]
|
||||||
fn previous_response_expected_no_content(&self) -> bool {
|
fn previous_response_expected_no_content(&self) -> bool {
|
||||||
false
|
false
|
||||||
@@ -79,6 +81,7 @@ pub trait NetworkStream: Read + Write + Any + Send + Typeable {
|
|||||||
pub trait NetworkConnector {
|
pub trait NetworkConnector {
|
||||||
/// Type of Stream to create
|
/// Type of Stream to create
|
||||||
type Stream: Into<Box<NetworkStream + Send>>;
|
type Stream: Into<Box<NetworkStream + Send>>;
|
||||||
|
|
||||||
/// Connect to a remote address.
|
/// Connect to a remote address.
|
||||||
fn connect(&self, host: &str, port: u16, scheme: &str) -> ::Result<Self::Stream>;
|
fn connect(&self, host: &str, port: u16, scheme: &str) -> ::Result<Self::Stream>;
|
||||||
}
|
}
|
||||||
@@ -215,13 +218,17 @@ impl Clone for HttpListener {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
impl HttpListener {
|
impl From<TcpListener> for HttpListener {
|
||||||
|
fn from(listener: TcpListener) -> HttpListener {
|
||||||
|
HttpListener(listener)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
impl HttpListener {
|
||||||
/// Start listening to an address over HTTP.
|
/// Start listening to an address over HTTP.
|
||||||
pub fn new<To: ToSocketAddrs>(addr: To) -> ::Result<HttpListener> {
|
pub fn new<To: ToSocketAddrs>(addr: To) -> ::Result<HttpListener> {
|
||||||
Ok(HttpListener(try!(TcpListener::bind(addr))))
|
Ok(HttpListener(try!(TcpListener::bind(addr))))
|
||||||
}
|
}
|
||||||
|
|
||||||
}
|
}
|
||||||
|
|
||||||
impl NetworkListener for HttpListener {
|
impl NetworkListener for HttpListener {
|
||||||
@@ -382,17 +389,17 @@ impl NetworkConnector for HttpConnector {
|
|||||||
/// A closure as a connector used to generate TcpStreams per request
|
/// A closure as a connector used to generate TcpStreams per request
|
||||||
///
|
///
|
||||||
/// # Example
|
/// # Example
|
||||||
///
|
///
|
||||||
/// Basic example:
|
/// Basic example:
|
||||||
///
|
///
|
||||||
/// ```norun
|
/// ```norun
|
||||||
/// Client::with_connector(|addr: &str, port: u16, scheme: &str| {
|
/// Client::with_connector(|addr: &str, port: u16, scheme: &str| {
|
||||||
/// TcpStream::connect(&(addr, port))
|
/// TcpStream::connect(&(addr, port))
|
||||||
/// });
|
/// });
|
||||||
/// ```
|
/// ```
|
||||||
///
|
///
|
||||||
/// Example using TcpBuilder from the net2 crate if you want to configure your source socket:
|
/// Example using TcpBuilder from the net2 crate if you want to configure your source socket:
|
||||||
///
|
///
|
||||||
/// ```norun
|
/// ```norun
|
||||||
/// Client::with_connector(|addr: &str, port: u16, scheme: &str| {
|
/// Client::with_connector(|addr: &str, port: u16, scheme: &str| {
|
||||||
/// let b = try!(TcpBuilder::new_v4());
|
/// let b = try!(TcpBuilder::new_v4());
|
||||||
@@ -499,7 +506,6 @@ pub struct HttpsListener<S: Ssl> {
|
|||||||
}
|
}
|
||||||
|
|
||||||
impl<S: Ssl> HttpsListener<S> {
|
impl<S: Ssl> HttpsListener<S> {
|
||||||
|
|
||||||
/// Start listening to an address over HTTPS.
|
/// Start listening to an address over HTTPS.
|
||||||
pub fn new<To: ToSocketAddrs>(addr: To, ssl: S) -> ::Result<HttpsListener<S>> {
|
pub fn new<To: ToSocketAddrs>(addr: To, ssl: S) -> ::Result<HttpsListener<S>> {
|
||||||
HttpListener::new(addr).map(|l| HttpsListener {
|
HttpListener::new(addr).map(|l| HttpsListener {
|
||||||
@@ -508,6 +514,13 @@ impl<S: Ssl> HttpsListener<S> {
|
|||||||
})
|
})
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// Construct an HttpsListener from a bound `TcpListener`.
|
||||||
|
pub fn with_listener(listener: HttpListener, ssl: S) -> HttpsListener<S> {
|
||||||
|
HttpsListener {
|
||||||
|
listener: listener,
|
||||||
|
ssl: ssl
|
||||||
|
}
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
impl<S: Ssl + Clone> NetworkListener for HttpsListener<S> {
|
impl<S: Ssl + Clone> NetworkListener for HttpsListener<S> {
|
||||||
@@ -576,7 +589,6 @@ mod openssl {
|
|||||||
use openssl::x509::X509FileType;
|
use openssl::x509::X509FileType;
|
||||||
use super::{NetworkStream, HttpStream};
|
use super::{NetworkStream, HttpStream};
|
||||||
|
|
||||||
|
|
||||||
/// An implementation of `Ssl` for OpenSSL.
|
/// An implementation of `Ssl` for OpenSSL.
|
||||||
///
|
///
|
||||||
/// # Example
|
/// # Example
|
||||||
@@ -678,7 +690,6 @@ mod tests {
|
|||||||
|
|
||||||
let mock = stream.downcast::<MockStream>().ok().unwrap();
|
let mock = stream.downcast::<MockStream>().ok().unwrap();
|
||||||
assert_eq!(mock, Box::new(MockStream::new()));
|
assert_eq!(mock, Box::new(MockStream::new()));
|
||||||
|
|
||||||
}
|
}
|
||||||
|
|
||||||
#[test]
|
#[test]
|
||||||
@@ -688,6 +699,6 @@ mod tests {
|
|||||||
|
|
||||||
let mock = unsafe { stream.downcast_unchecked::<MockStream>() };
|
let mock = unsafe { stream.downcast_unchecked::<MockStream>() };
|
||||||
assert_eq!(mock, Box::new(MockStream::new()));
|
assert_eq!(mock, Box::new(MockStream::new()));
|
||||||
|
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -42,7 +42,6 @@ impl<A: NetworkListener + Send + 'static> ListenerPool<A> {
|
|||||||
fn spawn_with<A, F>(supervisor: mpsc::Sender<()>, work: Arc<F>, mut acceptor: A)
|
fn spawn_with<A, F>(supervisor: mpsc::Sender<()>, work: Arc<F>, mut acceptor: A)
|
||||||
where A: NetworkListener + Send + 'static,
|
where A: NetworkListener + Send + 'static,
|
||||||
F: Fn(<A as NetworkListener>::Stream) + Send + Sync + 'static {
|
F: Fn(<A as NetworkListener>::Stream) + Send + Sync + 'static {
|
||||||
|
|
||||||
thread::spawn(move || {
|
thread::spawn(move || {
|
||||||
let _sentinel = Sentinel::new(supervisor, ());
|
let _sentinel = Sentinel::new(supervisor, ());
|
||||||
|
|
||||||
@@ -77,3 +76,4 @@ impl<T: Send + 'static> Drop for Sentinel<T> {
|
|||||||
let _ = self.supervisor.send(self.value.take().unwrap());
|
let _ = self.supervisor.send(self.value.take().unwrap());
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -194,8 +194,6 @@ impl<L: NetworkListener> Server<L> {
|
|||||||
pub fn set_write_timeout(&mut self, dur: Option<Duration>) {
|
pub fn set_write_timeout(&mut self, dur: Option<Duration>) {
|
||||||
self.timeouts.write = dur;
|
self.timeouts.write = dur;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
||||||
}
|
}
|
||||||
|
|
||||||
impl Server<HttpListener> {
|
impl Server<HttpListener> {
|
||||||
@@ -219,6 +217,7 @@ impl<L: NetworkListener + Send + 'static> Server<L> {
|
|||||||
pub fn handle<H: Handler + 'static>(self, handler: H) -> ::Result<Listening> {
|
pub fn handle<H: Handler + 'static>(self, handler: H) -> ::Result<Listening> {
|
||||||
self.handle_threads(handler, num_cpus::get() * 5 / 4)
|
self.handle_threads(handler, num_cpus::get() * 5 / 4)
|
||||||
}
|
}
|
||||||
|
|
||||||
/// Binds to a socket and starts handling connections with the provided
|
/// Binds to a socket and starts handling connections with the provided
|
||||||
/// number of threads.
|
/// number of threads.
|
||||||
pub fn handle_threads<H: Handler + 'static>(self, handler: H,
|
pub fn handle_threads<H: Handler + 'static>(self, handler: H,
|
||||||
@@ -228,8 +227,7 @@ impl<L: NetworkListener + Send + 'static> Server<L> {
|
|||||||
}
|
}
|
||||||
|
|
||||||
fn handle<H, L>(mut server: Server<L>, handler: H, threads: usize) -> ::Result<Listening>
|
fn handle<H, L>(mut server: Server<L>, handler: H, threads: usize) -> ::Result<Listening>
|
||||||
where H: Handler + 'static,
|
where H: Handler + 'static, L: NetworkListener + Send + 'static {
|
||||||
L: NetworkListener + Send + 'static {
|
|
||||||
let socket = try!(server.listener.local_addr());
|
let socket = try!(server.listener.local_addr());
|
||||||
|
|
||||||
debug!("threads = {:?}", threads);
|
debug!("threads = {:?}", threads);
|
||||||
@@ -251,7 +249,6 @@ struct Worker<H: Handler + 'static> {
|
|||||||
}
|
}
|
||||||
|
|
||||||
impl<H: Handler + 'static> Worker<H> {
|
impl<H: Handler + 'static> Worker<H> {
|
||||||
|
|
||||||
fn new(handler: H, timeouts: Timeouts) -> Worker<H> {
|
fn new(handler: H, timeouts: Timeouts) -> Worker<H> {
|
||||||
Worker {
|
Worker {
|
||||||
handler: handler,
|
handler: handler,
|
||||||
@@ -299,7 +296,6 @@ impl<H: Handler + 'static> Worker<H> {
|
|||||||
self.set_write_timeout(s, self.timeouts.write)
|
self.set_write_timeout(s, self.timeouts.write)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
||||||
#[cfg(not(feature = "timeouts"))]
|
#[cfg(not(feature = "timeouts"))]
|
||||||
fn set_write_timeout(&self, _s: &NetworkStream, _timeout: Option<Duration>) -> io::Result<()> {
|
fn set_write_timeout(&self, _s: &NetworkStream, _timeout: Option<Duration>) -> io::Result<()> {
|
||||||
Ok(())
|
Ok(())
|
||||||
@@ -339,7 +335,6 @@ impl<H: Handler + 'static> Worker<H> {
|
|||||||
}
|
}
|
||||||
};
|
};
|
||||||
|
|
||||||
|
|
||||||
if !self.handle_expect(&req, wrt) {
|
if !self.handle_expect(&req, wrt) {
|
||||||
return false;
|
return false;
|
||||||
}
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user