use std::error::Error as StdError;
use std::mem;
use bytes::{BufMut, BytesMut};
use futures::Future;
use http::{uri, Uri};
use tokio_io::{AsyncRead, AsyncWrite};
#[cfg(feature = "runtime")] pub use self::http::HttpConnector;
pub trait Connect: Send + Sync {
type Transport: AsyncRead + AsyncWrite + Send + 'static;
type Error: Into<Box<StdError + Send + Sync>>;
type Future: Future<Item=(Self::Transport, Connected), Error=Self::Error> + Send;
fn connect(&self, dst: Destination) -> Self::Future;
}
#[derive(Clone, Debug)]
pub struct Destination {
pub(super) uri: Uri,
}
#[derive(Debug)]
pub struct Connected {
pub(super) is_proxied: bool,
}
impl Destination {
#[inline]
pub fn scheme(&self) -> &str {
self.uri
.scheme_part()
.map(|s| s.as_str())
.unwrap_or("")
}
#[inline]
pub fn host(&self) -> &str {
self.uri
.host()
.unwrap_or("")
}
#[inline]
pub fn port(&self) -> Option<u16> {
self.uri.port()
}
pub fn set_scheme(&mut self, scheme: &str) -> ::Result<()> {
let scheme = scheme.parse().map_err(::error::Parse::from)?;
self.update_uri(move |parts| {
parts.scheme = Some(scheme);
})
}
pub fn set_host(&mut self, host: &str) -> ::Result<()> {
if host.contains(&['@',':'][..]) {
return Err(::error::Parse::Uri.into());
}
let auth = if let Some(port) = self.port() {
format!("{}:{}", host, port).parse().map_err(::error::Parse::from)?
} else {
host.parse().map_err(::error::Parse::from)?
};
self.update_uri(move |parts| {
parts.authority = Some(auth);
})
}
pub fn set_port<P>(&mut self, port: P)
where
P: Into<Option<u16>>,
{
self.set_port_opt(port.into());
}
fn set_port_opt(&mut self, port: Option<u16>) {
use std::fmt::Write;
let auth = if let Some(port) = port {
let host = self.host();
let cap = host.len() + 1 + 5;
let mut buf = BytesMut::with_capacity(cap);
buf.put_slice(host.as_bytes());
buf.put_u8(b':');
write!(buf, "{}", port)
.expect("should have space for 5 digits");
uri::Authority::from_shared(buf.freeze())
.expect("valid host + :port should be valid authority")
} else {
self.host().parse()
.expect("valid host without port should be valid authority")
};
self.update_uri(move |parts| {
parts.authority = Some(auth);
})
.expect("valid uri should be valid with port");
}
fn update_uri<F>(&mut self, f: F) -> ::Result<()>
where
F: FnOnce(&mut uri::Parts)
{
let old_uri = mem::replace(&mut self.uri, Uri::default());
let mut parts: uri::Parts = old_uri.clone().into();
f(&mut parts);
match Uri::from_parts(parts) {
Ok(uri) => {
self.uri = uri;
Ok(())
},
Err(err) => {
self.uri = old_uri;
Err(::error::Parse::from(err).into())
},
}
}
}
impl Connected {
pub fn new() -> Connected {
Connected {
is_proxied: false,
}
}
pub fn proxy(mut self, is_proxied: bool) -> Connected {
self.is_proxied = is_proxied;
self
}
}
#[cfg(test)]
mod tests {
use super::Destination;
#[test]
fn test_destination_set_scheme() {
let mut dst = Destination {
uri: "http://hyper.rs".parse().expect("initial parse"),
};
assert_eq!(dst.scheme(), "http");
assert_eq!(dst.host(), "hyper.rs");
dst.set_scheme("https").expect("set https");
assert_eq!(dst.scheme(), "https");
assert_eq!(dst.host(), "hyper.rs");
dst.set_scheme("<im not a scheme//?>").unwrap_err();
assert_eq!(dst.scheme(), "https", "error doesn't modify dst");
assert_eq!(dst.host(), "hyper.rs", "error doesn't modify dst");
}
#[test]
fn test_destination_set_host() {
let mut dst = Destination {
uri: "http://hyper.rs".parse().expect("initial parse"),
};
assert_eq!(dst.scheme(), "http");
assert_eq!(dst.host(), "hyper.rs");
assert_eq!(dst.port(), None);
dst.set_host("seanmonstar.com").expect("set https");
assert_eq!(dst.scheme(), "http");
assert_eq!(dst.host(), "seanmonstar.com");
assert_eq!(dst.port(), None);
dst.set_host("/im-not a host! >:)").unwrap_err();
assert_eq!(dst.scheme(), "http", "error doesn't modify dst");
assert_eq!(dst.host(), "seanmonstar.com", "error doesn't modify dst");
assert_eq!(dst.port(), None, "error doesn't modify dst");
let mut dst = Destination {
uri: "http://hyper.rs:8080".parse().expect("initial parse 2"),
};
assert_eq!(dst.scheme(), "http");
assert_eq!(dst.host(), "hyper.rs");
assert_eq!(dst.port(), Some(8080));
dst.set_host("seanmonstar.com").expect("set host");
assert_eq!(dst.scheme(), "http");
assert_eq!(dst.host(), "seanmonstar.com");
assert_eq!(dst.port(), Some(8080));
dst.set_host("/im-not a host! >:)").unwrap_err();
assert_eq!(dst.scheme(), "http", "error doesn't modify dst");
assert_eq!(dst.host(), "seanmonstar.com", "error doesn't modify dst");
assert_eq!(dst.port(), Some(8080), "error doesn't modify dst");
dst.set_host("seanmonstar.com:3030").expect_err("set_host sneaky port");
assert_eq!(dst.scheme(), "http", "error doesn't modify dst");
assert_eq!(dst.host(), "seanmonstar.com", "error doesn't modify dst");
assert_eq!(dst.port(), Some(8080), "error doesn't modify dst");
dst.set_host("sean@nope").expect_err("set_host sneaky userinfo");
assert_eq!(dst.scheme(), "http", "error doesn't modify dst");
assert_eq!(dst.host(), "seanmonstar.com", "error doesn't modify dst");
assert_eq!(dst.port(), Some(8080), "error doesn't modify dst");
}
#[test]
fn test_destination_set_port() {
let mut dst = Destination {
uri: "http://hyper.rs".parse().expect("initial parse"),
};
assert_eq!(dst.scheme(), "http");
assert_eq!(dst.host(), "hyper.rs");
assert_eq!(dst.port(), None);
dst.set_port(None);
assert_eq!(dst.scheme(), "http");
assert_eq!(dst.host(), "hyper.rs");
assert_eq!(dst.port(), None);
dst.set_port(8080);
assert_eq!(dst.scheme(), "http");
assert_eq!(dst.host(), "hyper.rs");
assert_eq!(dst.port(), Some(8080));
let mut dst = Destination {
uri: "http://hyper.rs:8080".parse().expect("initial parse 2"),
};
assert_eq!(dst.scheme(), "http");
assert_eq!(dst.host(), "hyper.rs");
assert_eq!(dst.port(), Some(8080));
dst.set_port(3030);
assert_eq!(dst.scheme(), "http");
assert_eq!(dst.host(), "hyper.rs");
assert_eq!(dst.port(), Some(3030));
dst.set_port(None);
assert_eq!(dst.scheme(), "http");
assert_eq!(dst.host(), "hyper.rs");
assert_eq!(dst.port(), None);
}
}
#[cfg(feature = "runtime")]
mod http {
use super::*;
use std::borrow::Cow;
use std::fmt;
use std::io;
use std::mem;
use std::net::{IpAddr, SocketAddr};
use std::sync::Arc;
use std::time::Duration;
use futures::{Async, Poll};
use futures::future::{Executor, ExecuteError};
use futures::sync::oneshot;
use futures_cpupool::{Builder as CpuPoolBuilder};
use http::uri::Scheme;
use net2::TcpBuilder;
use tokio_reactor::Handle;
use tokio_tcp::{TcpStream, ConnectFuture};
use super::super::dns;
use self::http_connector::HttpConnectorBlockingTask;
fn connect(addr: &SocketAddr, local_addr: &Option<IpAddr>, handle: &Option<Handle>) -> io::Result<ConnectFuture> {
let builder = match addr {
&SocketAddr::V4(_) => TcpBuilder::new_v4()?,
&SocketAddr::V6(_) => TcpBuilder::new_v6()?,
};
if let Some(ref local_addr) = *local_addr {
builder.bind(SocketAddr::new(local_addr.clone(), 0))?;
}
else if cfg!(windows) {
let any: SocketAddr = match addr {
&SocketAddr::V4(_) => {
([0, 0, 0, 0], 0).into()
},
&SocketAddr::V6(_) => {
([0, 0, 0, 0, 0, 0, 0, 0], 0).into()
}
};
builder.bind(any)?;
}
let handle = match *handle {
Some(ref handle) => Cow::Borrowed(handle),
None => Cow::Owned(Handle::current()),
};
Ok(TcpStream::connect_std(builder.to_tcp_stream()?, addr, &handle))
}
#[derive(Clone)]
pub struct HttpConnector {
executor: HttpConnectExecutor,
enforce_http: bool,
handle: Option<Handle>,
keep_alive_timeout: Option<Duration>,
nodelay: bool,
local_address: Option<IpAddr>,
}
impl HttpConnector {
#[inline]
pub fn new(threads: usize) -> HttpConnector {
HttpConnector::new_with_handle_opt(threads, None)
}
pub fn new_with_handle(threads: usize, handle: Handle) -> HttpConnector {
HttpConnector::new_with_handle_opt(threads, Some(handle))
}
fn new_with_handle_opt(threads: usize, handle: Option<Handle>) -> HttpConnector {
let pool = CpuPoolBuilder::new()
.name_prefix("hyper-dns")
.pool_size(threads)
.create();
HttpConnector::new_with_executor(pool, handle)
}
pub fn new_with_executor<E: 'static>(executor: E, handle: Option<Handle>) -> HttpConnector
where E: Executor<HttpConnectorBlockingTask> + Send + Sync
{
HttpConnector {
executor: HttpConnectExecutor(Arc::new(executor)),
enforce_http: true,
handle,
keep_alive_timeout: None,
nodelay: false,
local_address: None,
}
}
#[inline]
pub fn enforce_http(&mut self, is_enforced: bool) {
self.enforce_http = is_enforced;
}
#[inline]
pub fn set_keepalive(&mut self, dur: Option<Duration>) {
self.keep_alive_timeout = dur;
}
#[inline]
pub fn set_nodelay(&mut self, nodelay: bool) {
self.nodelay = nodelay;
}
#[inline]
pub fn set_local_address(&mut self, addr: Option<IpAddr>) {
self.local_address = addr;
}
}
impl fmt::Debug for HttpConnector {
#[inline]
fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
f.debug_struct("HttpConnector")
.finish()
}
}
impl Connect for HttpConnector {
type Transport = TcpStream;
type Error = io::Error;
type Future = HttpConnecting;
fn connect(&self, dst: Destination) -> Self::Future {
trace!(
"Http::connect; scheme={}, host={}, port={:?}",
dst.scheme(),
dst.host(),
dst.port(),
);
if self.enforce_http {
if dst.uri.scheme_part() != Some(&Scheme::HTTP) {
return invalid_url(InvalidUrl::NotHttp, &self.handle);
}
} else if dst.uri.scheme_part().is_none() {
return invalid_url(InvalidUrl::MissingScheme, &self.handle);
}
let host = match dst.uri.host() {
Some(s) => s,
None => return invalid_url(InvalidUrl::MissingAuthority, &self.handle),
};
let port = match dst.uri.port() {
Some(port) => port,
None => if dst.uri.scheme_part() == Some(&Scheme::HTTPS) { 443 } else { 80 },
};
HttpConnecting {
state: State::Lazy(self.executor.clone(), host.into(), port, self.local_address),
handle: self.handle.clone(),
keep_alive_timeout: self.keep_alive_timeout,
nodelay: self.nodelay,
}
}
}
#[inline]
fn invalid_url(err: InvalidUrl, handle: &Option<Handle>) -> HttpConnecting {
HttpConnecting {
state: State::Error(Some(io::Error::new(io::ErrorKind::InvalidInput, err))),
handle: handle.clone(),
keep_alive_timeout: None,
nodelay: false,
}
}
#[derive(Debug, Clone, Copy)]
enum InvalidUrl {
MissingScheme,
NotHttp,
MissingAuthority,
}
impl fmt::Display for InvalidUrl {
fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
f.write_str(self.description())
}
}
impl StdError for InvalidUrl {
fn description(&self) -> &str {
match *self {
InvalidUrl::MissingScheme => "invalid URL, missing scheme",
InvalidUrl::NotHttp => "invalid URL, scheme must be http",
InvalidUrl::MissingAuthority => "invalid URL, missing domain",
}
}
}
#[must_use = "futures do nothing unless polled"]
pub struct HttpConnecting {
state: State,
handle: Option<Handle>,
keep_alive_timeout: Option<Duration>,
nodelay: bool,
}
enum State {
Lazy(HttpConnectExecutor, String, u16, Option<IpAddr>),
Resolving(oneshot::SpawnHandle<dns::IpAddrs, io::Error>, Option<IpAddr>),
Connecting(ConnectingTcp),
Error(Option<io::Error>),
}
impl Future for HttpConnecting {
type Item = (TcpStream, Connected);
type Error = io::Error;
fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
loop {
let state;
match self.state {
State::Lazy(ref executor, ref mut host, port, local_addr) => {
if let Some(addrs) = dns::IpAddrs::try_parse(host, port) {
state = State::Connecting(ConnectingTcp {
addrs: addrs,
local_addr: local_addr,
current: None
})
} else {
let host = mem::replace(host, String::new());
let work = dns::Work::new(host, port);
state = State::Resolving(oneshot::spawn(work, executor), local_addr);
}
},
State::Resolving(ref mut future, local_addr) => {
match try!(future.poll()) {
Async::NotReady => return Ok(Async::NotReady),
Async::Ready(addrs) => {
state = State::Connecting(ConnectingTcp {
addrs: addrs,
local_addr: local_addr,
current: None,
})
}
};
},
State::Connecting(ref mut c) => {
let sock = try_ready!(c.poll(&self.handle));
if let Some(dur) = self.keep_alive_timeout {
sock.set_keepalive(Some(dur))?;
}
sock.set_nodelay(self.nodelay)?;
return Ok(Async::Ready((sock, Connected::new())));
},
State::Error(ref mut e) => return Err(e.take().expect("polled more than once")),
}
self.state = state;
}
}
}
impl fmt::Debug for HttpConnecting {
fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
f.pad("HttpConnecting")
}
}
struct ConnectingTcp {
addrs: dns::IpAddrs,
local_addr: Option<IpAddr>,
current: Option<ConnectFuture>,
}
impl ConnectingTcp {
fn poll(&mut self, handle: &Option<Handle>) -> Poll<TcpStream, io::Error> {
let mut err = None;
loop {
if let Some(ref mut current) = self.current {
match current.poll() {
Ok(ok) => return Ok(ok),
Err(e) => {
trace!("connect error {:?}", e);
err = Some(e);
if let Some(addr) = self.addrs.next() {
debug!("connecting to {}", addr);
*current = connect(&addr, &self.local_addr, handle)?;
continue;
}
}
}
} else if let Some(addr) = self.addrs.next() {
debug!("connecting to {}", addr);
self.current = Some(connect(&addr, &self.local_addr, handle)?);
continue;
}
return Err(err.take().expect("missing connect error"));
}
}
}
mod http_connector {
use super::*;
pub struct HttpConnectorBlockingTask {
pub(super) work: oneshot::Execute<dns::Work>
}
impl fmt::Debug for HttpConnectorBlockingTask {
fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
f.pad("HttpConnectorBlockingTask")
}
}
impl Future for HttpConnectorBlockingTask {
type Item = ();
type Error = ();
fn poll(&mut self) -> Poll<(), ()> {
self.work.poll()
}
}
}
#[derive(Clone)]
struct HttpConnectExecutor(Arc<Executor<HttpConnectorBlockingTask> + Send + Sync>);
impl Executor<oneshot::Execute<dns::Work>> for HttpConnectExecutor {
fn execute(&self, future: oneshot::Execute<dns::Work>) -> Result<(), ExecuteError<oneshot::Execute<dns::Work>>> {
self.0.execute(HttpConnectorBlockingTask { work: future })
.map_err(|err| ExecuteError::new(err.kind(), err.into_future().work))
}
}
#[cfg(test)]
mod tests {
use std::io;
use futures::Future;
use super::{Connect, Destination, HttpConnector};
#[test]
fn test_errors_missing_authority() {
let uri = "/foo/bar?baz".parse().unwrap();
let dst = Destination {
uri,
};
let connector = HttpConnector::new(1);
assert_eq!(connector.connect(dst).wait().unwrap_err().kind(), io::ErrorKind::InvalidInput);
}
#[test]
fn test_errors_enforce_http() {
let uri = "https://example.domain/foo/bar?baz".parse().unwrap();
let dst = Destination {
uri,
};
let connector = HttpConnector::new(1);
assert_eq!(connector.connect(dst).wait().unwrap_err().kind(), io::ErrorKind::InvalidInput);
}
#[test]
fn test_errors_missing_scheme() {
let uri = "example.domain".parse().unwrap();
let dst = Destination {
uri,
};
let connector = HttpConnector::new(1);
assert_eq!(connector.connect(dst).wait().unwrap_err().kind(), io::ErrorKind::InvalidInput);
}
}
}