use std::any::{Any, TypeId};
use std::fmt;
use std::io::{IoResult, IoError, ConnectionAborted, InvalidInput, OtherIoError,
Stream, Listener, Acceptor};
use std::io::net::ip::{SocketAddr, ToSocketAddr, Port};
use std::io::net::tcp::{TcpStream, TcpListener, TcpAcceptor};
use std::mem;
use std::raw::{self, TraitObject};
use std::sync::Arc;
use uany::UnsafeAnyExt;
use openssl::ssl::{Ssl, SslStream, SslContext};
use openssl::ssl::SslVerifyMode::SslVerifyNone;
use openssl::ssl::SslMethod::Sslv23;
use openssl::ssl::error::{SslError, StreamError, OpenSslErrors, SslSessionClosed};
use openssl::x509::X509FileType;
macro_rules! try_some {
($expr:expr) => (match $expr {
Some(val) => { return Err(val); },
_ => {}
})
}
#[allow(missing_copy_implementations)]
pub struct Fresh;
#[allow(missing_copy_implementations)]
pub struct Streaming;
pub trait NetworkListener {
type Acceptor: NetworkAcceptor;
fn listen<To: ToSocketAddr>(&mut self, addr: To) -> IoResult<Self::Acceptor>;
}
pub trait NetworkAcceptor: Clone + Send {
type Stream: NetworkStream + Send + Clone;
fn accept(&mut self) -> IoResult<Self::Stream>;
fn socket_name(&self) -> IoResult<SocketAddr>;
fn close(&mut self) -> IoResult<()>;
fn incoming(&mut self) -> NetworkConnections<Self> {
NetworkConnections(self)
}
}
pub struct NetworkConnections<'a, N: NetworkAcceptor>(&'a mut N);
impl<'a, N: NetworkAcceptor> Iterator for NetworkConnections<'a, N> {
type Item = IoResult<N::Stream>;
fn next(&mut self) -> Option<IoResult<N::Stream>> {
Some(self.0.accept())
}
}
pub trait NetworkStream: Stream + Any + StreamClone + Send {
fn peer_name(&mut self) -> IoResult<SocketAddr>;
}
#[doc(hidden)]
pub trait StreamClone {
fn clone_box(&self) -> Box<NetworkStream + Send>;
}
impl<T: NetworkStream + Send + Clone> StreamClone for T {
#[inline]
fn clone_box(&self) -> Box<NetworkStream + Send> {
Box::new(self.clone())
}
}
pub trait NetworkConnector {
type Stream: NetworkStream + Send;
fn connect(&mut self, host: &str, port: Port, scheme: &str) -> IoResult<Self::Stream>;
}
impl fmt::Show for Box<NetworkStream + Send> {
fn fmt(&self, fmt: &mut fmt::Formatter) -> fmt::Result {
fmt.pad("Box<NetworkStream>")
}
}
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<usize> { (**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() }
}
impl<'a> Reader for &'a mut NetworkStream {
#[inline]
fn read(&mut self, buf: &mut [u8]) -> IoResult<usize> { (**self).read(buf) }
}
impl<'a> Writer for &'a mut NetworkStream {
#[inline]
fn write(&mut self, msg: &[u8]) -> IoResult<()> { (**self).write(msg) }
#[inline]
fn flush(&mut self) -> IoResult<()> { (**self).flush() }
}
impl UnsafeAnyExt for NetworkStream {
unsafe fn downcast_ref_unchecked<T: 'static>(&self) -> &T {
mem::transmute(mem::transmute::<&NetworkStream,
raw::TraitObject>(self).data)
}
unsafe fn downcast_mut_unchecked<T: 'static>(&mut self) -> &mut T {
mem::transmute(mem::transmute::<&mut NetworkStream,
raw::TraitObject>(self).data)
}
unsafe fn downcast_unchecked<T: 'static>(self: Box<NetworkStream>) -> Box<T> {
mem::transmute(mem::transmute::<Box<NetworkStream>,
raw::TraitObject>(self).data)
}
}
impl NetworkStream {
#[inline]
pub fn is<T: 'static>(&self) -> bool {
self.get_type_id() == TypeId::of::<T>()
}
#[inline]
pub fn downcast_ref<T: 'static>(&self) -> Option<&T> {
if self.is::<T>() {
Some(unsafe { self.downcast_ref_unchecked() })
} else {
None
}
}
#[inline]
pub fn downcast_mut<T: 'static>(&mut self) -> Option<&mut T> {
if self.is::<T>() {
Some(unsafe { self.downcast_mut_unchecked() })
} else {
None
}
}
pub fn downcast<T: 'static>(self: Box<NetworkStream>)
-> Result<Box<T>, Box<NetworkStream>> {
if self.is::<T>() {
Ok(unsafe { self.downcast_unchecked() })
} else {
Err(self)
}
}
}
#[allow(missing_copy_implementations)]
pub enum HttpListener {
Http,
Https(Path, Path),
}
impl NetworkListener for HttpListener {
type Acceptor = HttpAcceptor;
#[inline]
fn listen<To: ToSocketAddr>(&mut self, addr: To) -> IoResult<HttpAcceptor> {
let mut tcp = try!(TcpListener::bind(addr));
let addr = try!(tcp.socket_name());
Ok(match *self {
HttpListener::Http => HttpAcceptor::Http(try!(tcp.listen()), addr),
HttpListener::Https(ref cert, ref key) => {
let mut ssl_context = try!(SslContext::new(Sslv23).map_err(lift_ssl_error));
try_some!(ssl_context.set_cipher_list("DEFAULT").map(lift_ssl_error));
try_some!(ssl_context.set_certificate_file(
cert, X509FileType::PEM).map(lift_ssl_error));
try_some!(ssl_context.set_private_key_file(
key, X509FileType::PEM).map(lift_ssl_error));
ssl_context.set_verify(SslVerifyNone, None);
HttpAcceptor::Https(try!(tcp.listen()), addr, Arc::new(ssl_context))
}
})
}
}
#[derive(Clone)]
pub enum HttpAcceptor {
Http(TcpAcceptor, SocketAddr),
Https(TcpAcceptor, SocketAddr, Arc<SslContext>),
}
impl NetworkAcceptor for HttpAcceptor {
type Stream = HttpStream;
#[inline]
fn accept(&mut self) -> IoResult<HttpStream> {
Ok(match *self {
HttpAcceptor::Http(ref mut tcp, _) => HttpStream::Http(try!(tcp.accept())),
HttpAcceptor::Https(ref mut tcp, _, ref ssl_context) => {
let stream = try!(tcp.accept());
let ssl_stream = try!(SslStream::<TcpStream>::new_server(&**ssl_context, stream).
map_err(lift_ssl_error));
HttpStream::Https(ssl_stream)
}
})
}
#[inline]
fn close(&mut self) -> IoResult<()> {
match *self {
HttpAcceptor::Http(ref mut tcp, _) => tcp.close_accept(),
HttpAcceptor::Https(ref mut tcp, _, _) => tcp.close_accept(),
}
}
#[inline]
fn socket_name(&self) -> IoResult<SocketAddr> {
match *self {
HttpAcceptor::Http(_, addr) => Ok(addr),
HttpAcceptor::Https(_, addr, _) => Ok(addr),
}
}
}
#[derive(Clone)]
pub enum HttpStream {
Http(TcpStream),
Https(SslStream<TcpStream>),
}
impl Reader for HttpStream {
#[inline]
fn read(&mut self, buf: &mut [u8]) -> IoResult<usize> {
match *self {
HttpStream::Http(ref mut inner) => inner.read(buf),
HttpStream::Https(ref mut inner) => inner.read(buf)
}
}
}
impl Writer for HttpStream {
#[inline]
fn write(&mut self, msg: &[u8]) -> IoResult<()> {
match *self {
HttpStream::Http(ref mut inner) => inner.write(msg),
HttpStream::Https(ref mut inner) => inner.write(msg)
}
}
#[inline]
fn flush(&mut self) -> IoResult<()> {
match *self {
HttpStream::Http(ref mut inner) => inner.flush(),
HttpStream::Https(ref mut inner) => inner.flush(),
}
}
}
impl NetworkStream for HttpStream {
fn peer_name(&mut self) -> IoResult<SocketAddr> {
match *self {
HttpStream::Http(ref mut inner) => inner.peer_name(),
HttpStream::Https(ref mut inner) => inner.get_mut().peer_name()
}
}
}
#[allow(missing_copy_implementations)]
pub struct HttpConnector<'v>(pub Option<ContextVerifier<'v>>);
pub type ContextVerifier<'v> = Box<FnMut(&mut SslContext) -> ()+'v>;
impl<'v> NetworkConnector for HttpConnector<'v> {
type Stream = HttpStream;
fn connect(&mut self, host: &str, port: Port, scheme: &str) -> IoResult<HttpStream> {
let addr = (host, port);
match scheme {
"http" => {
debug!("http scheme");
Ok(HttpStream::Http(try!(TcpStream::connect(addr))))
},
"https" => {
debug!("https scheme");
let stream = try!(TcpStream::connect(addr));
let mut context = try!(SslContext::new(Sslv23).map_err(lift_ssl_error));
if let Some(ref mut verifier) = self.0 {
verifier(&mut context);
}
let ssl = try!(Ssl::new(&context).map_err(lift_ssl_error));
try!(ssl.set_hostname(host).map_err(lift_ssl_error));
let stream = try!(SslStream::new(&context, stream).map_err(lift_ssl_error));
Ok(HttpStream::Https(stream))
},
_ => {
Err(IoError {
kind: InvalidInput,
desc: "Invalid scheme for Http",
detail: None
})
}
}
}
}
fn lift_ssl_error(ssl: SslError) -> IoError {
debug!("lift_ssl_error: {:?}", ssl);
match ssl {
StreamError(err) => err,
SslSessionClosed => IoError {
kind: ConnectionAborted,
desc: "SSL Connection Closed",
detail: None
},
OpenSslErrors(errs) => IoError {
kind: OtherIoError,
desc: "Error in OpenSSL",
detail: Some(format!("{:?}", errs))
}
}
}
#[cfg(test)]
mod tests {
use uany::UnsafeAnyExt;
use mock::MockStream;
use super::NetworkStream;
#[test]
fn test_downcast_box_stream() {
let stream = box MockStream::new() as Box<NetworkStream + Send>;
let mock = stream.downcast::<MockStream>().ok().unwrap();
assert_eq!(mock, box MockStream::new());
}
#[test]
fn test_downcast_unchecked_box_stream() {
let stream = box MockStream::new() as Box<NetworkStream + Send>;
let mock = unsafe { stream.downcast_unchecked::<MockStream>() };
assert_eq!(mock, box MockStream::new());
}
}