feat(client): remove Destination for http::Uri in connectors
BREAKING CHANGE: All usage of `hyper::client::connect::Destination` should be replaced with `http::Uri`.
This commit is contained in:
@@ -5,7 +5,7 @@ extern crate test;
|
|||||||
|
|
||||||
use std::net::SocketAddr;
|
use std::net::SocketAddr;
|
||||||
use tokio::net::TcpListener;
|
use tokio::net::TcpListener;
|
||||||
use hyper::client::connect::{Destination, HttpConnector};
|
use hyper::client::connect::{HttpConnector};
|
||||||
use hyper::service::Service;
|
use hyper::service::Service;
|
||||||
use http::Uri;
|
use http::Uri;
|
||||||
|
|
||||||
@@ -19,8 +19,7 @@ fn http_connector(b: &mut test::Bencher) {
|
|||||||
.expect("rt build");
|
.expect("rt build");
|
||||||
let mut listener = rt.block_on(TcpListener::bind(&SocketAddr::from(([127, 0, 0, 1], 0)))).expect("bind");
|
let mut listener = rt.block_on(TcpListener::bind(&SocketAddr::from(([127, 0, 0, 1], 0)))).expect("bind");
|
||||||
let addr = listener.local_addr().expect("local_addr");
|
let addr = listener.local_addr().expect("local_addr");
|
||||||
let uri: Uri = format!("http://{}/", addr).parse().expect("uri parse");
|
let dst: Uri = format!("http://{}/", addr).parse().expect("uri parse");
|
||||||
let dst = Destination::try_from_uri(uri).expect("destination");
|
|
||||||
let mut connector = HttpConnector::new();
|
let mut connector = HttpConnector::new();
|
||||||
|
|
||||||
rt.spawn(async move {
|
rt.spawn(async move {
|
||||||
|
|||||||
@@ -391,11 +391,10 @@ mod tests {
|
|||||||
|
|
||||||
#[test]
|
#[test]
|
||||||
fn ip_addrs_try_parse_v6() {
|
fn ip_addrs_try_parse_v6() {
|
||||||
let uri = ::http::Uri::from_static("http://[::1]:8080/");
|
let dst = ::http::Uri::from_static("http://[::1]:8080/");
|
||||||
let dst = super::super::Destination { uri };
|
|
||||||
|
|
||||||
let mut addrs =
|
let mut addrs =
|
||||||
IpAddrs::try_parse(dst.host(), dst.port().expect("port")).expect("try_parse");
|
IpAddrs::try_parse(dst.host().expect("host"), dst.port_u16().expect("port")).expect("try_parse");
|
||||||
|
|
||||||
let expected = "[::1]:8080".parse::<SocketAddr>().expect("expected");
|
let expected = "[::1]:8080".parse::<SocketAddr>().expect("expected");
|
||||||
|
|
||||||
|
|||||||
@@ -17,7 +17,7 @@ use tokio::net::TcpStream;
|
|||||||
use tokio::time::Delay;
|
use tokio::time::Delay;
|
||||||
|
|
||||||
use super::dns::{self, resolve, GaiResolver, Resolve};
|
use super::dns::{self, resolve, GaiResolver, Resolve};
|
||||||
use super::{Connected, Destination};
|
use super::{Connected};
|
||||||
//#[cfg(feature = "runtime")] use super::dns::TokioThreadpoolGaiResolver;
|
//#[cfg(feature = "runtime")] use super::dns::TokioThreadpoolGaiResolver;
|
||||||
|
|
||||||
|
|
||||||
@@ -229,7 +229,7 @@ impl<R: fmt::Debug> fmt::Debug for HttpConnector<R> {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
impl<R> tower_service::Service<Destination> for HttpConnector<R>
|
impl<R> tower_service::Service<Uri> for HttpConnector<R>
|
||||||
where
|
where
|
||||||
R: Resolve + Clone + Send + Sync + 'static,
|
R: Resolve + Clone + Send + Sync + 'static,
|
||||||
R::Future: Send,
|
R::Future: Send,
|
||||||
@@ -243,7 +243,7 @@ where
|
|||||||
Poll::Ready(Ok(()))
|
Poll::Ready(Ok(()))
|
||||||
}
|
}
|
||||||
|
|
||||||
fn call(&mut self, dst: Destination) -> Self::Future {
|
fn call(&mut self, dst: Uri) -> Self::Future {
|
||||||
let mut self_ = self.clone();
|
let mut self_ = self.clone();
|
||||||
HttpConnecting {
|
HttpConnecting {
|
||||||
fut: Box::pin(async move { self_.call_async(dst).await }),
|
fut: Box::pin(async move { self_.call_async(dst).await }),
|
||||||
@@ -258,30 +258,30 @@ where
|
|||||||
{
|
{
|
||||||
async fn call_async(
|
async fn call_async(
|
||||||
&mut self,
|
&mut self,
|
||||||
dst: Destination,
|
dst: Uri,
|
||||||
) -> Result<(TcpStream, Connected), ConnectError> {
|
) -> Result<(TcpStream, Connected), ConnectError> {
|
||||||
trace!(
|
trace!(
|
||||||
"Http::connect; scheme={}, host={}, port={:?}",
|
"Http::connect; scheme={:?}, host={:?}, port={:?}",
|
||||||
dst.scheme(),
|
dst.scheme(),
|
||||||
dst.host(),
|
dst.host(),
|
||||||
dst.port(),
|
dst.port(),
|
||||||
);
|
);
|
||||||
|
|
||||||
if self.config.enforce_http {
|
if self.config.enforce_http {
|
||||||
if dst.uri.scheme() != Some(&Scheme::HTTP) {
|
if dst.scheme() != Some(&Scheme::HTTP) {
|
||||||
return Err(ConnectError {
|
return Err(ConnectError {
|
||||||
msg: INVALID_NOT_HTTP.into(),
|
msg: INVALID_NOT_HTTP.into(),
|
||||||
cause: None,
|
cause: None,
|
||||||
});
|
});
|
||||||
}
|
}
|
||||||
} else if dst.uri.scheme().is_none() {
|
} else if dst.scheme().is_none() {
|
||||||
return Err(ConnectError {
|
return Err(ConnectError {
|
||||||
msg: INVALID_MISSING_SCHEME.into(),
|
msg: INVALID_MISSING_SCHEME.into(),
|
||||||
cause: None,
|
cause: None,
|
||||||
});
|
});
|
||||||
}
|
}
|
||||||
|
|
||||||
let host = match dst.uri.host() {
|
let host = match dst.host() {
|
||||||
Some(s) => s,
|
Some(s) => s,
|
||||||
None => {
|
None => {
|
||||||
return Err(ConnectError {
|
return Err(ConnectError {
|
||||||
@@ -290,9 +290,9 @@ where
|
|||||||
})
|
})
|
||||||
}
|
}
|
||||||
};
|
};
|
||||||
let port = match dst.uri.port() {
|
let port = match dst.port() {
|
||||||
Some(port) => port.as_u16(),
|
Some(port) => port.as_u16(),
|
||||||
None => if dst.uri.scheme() == Some(&Scheme::HTTPS) { 443 } else { 80 },
|
None => if dst.scheme() == Some(&Scheme::HTTPS) { 443 } else { 80 },
|
||||||
};
|
};
|
||||||
|
|
||||||
let config = &self.config;
|
let config = &self.config;
|
||||||
@@ -351,26 +351,6 @@ where
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
impl<R> tower_service::Service<Uri> for HttpConnector<R>
|
|
||||||
where
|
|
||||||
R: Resolve + Clone + Send + Sync + 'static,
|
|
||||||
R::Future: Send,
|
|
||||||
{
|
|
||||||
type Response = TcpStream;
|
|
||||||
type Error = ConnectError;
|
|
||||||
type Future =
|
|
||||||
Pin<Box<dyn Future<Output = Result<Self::Response, Self::Error>> + Send + 'static>>;
|
|
||||||
|
|
||||||
fn poll_ready(&mut self, cx: &mut task::Context<'_>) -> Poll<Result<(), Self::Error>> {
|
|
||||||
tower_service::Service::<Destination>::poll_ready(self, cx)
|
|
||||||
}
|
|
||||||
|
|
||||||
fn call(&mut self, uri: Uri) -> Self::Future {
|
|
||||||
let mut self_ = self.clone();
|
|
||||||
Box::pin(async move { self_.call_async(Destination { uri }).await.map(|(s, _)| s) })
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
impl HttpInfo {
|
impl HttpInfo {
|
||||||
/// Get the remote address of the transport used.
|
/// Get the remote address of the transport used.
|
||||||
pub fn remote_addr(&self) -> SocketAddr {
|
pub fn remote_addr(&self) -> SocketAddr {
|
||||||
@@ -661,12 +641,14 @@ impl ConnectingTcp {
|
|||||||
mod tests {
|
mod tests {
|
||||||
use std::io;
|
use std::io;
|
||||||
|
|
||||||
|
use ::http::Uri;
|
||||||
|
|
||||||
use super::super::sealed::Connect;
|
use super::super::sealed::Connect;
|
||||||
use super::{Connected, Destination, HttpConnector};
|
use super::{Connected, HttpConnector};
|
||||||
|
|
||||||
async fn connect<C>(
|
async fn connect<C>(
|
||||||
connector: C,
|
connector: C,
|
||||||
dst: Destination,
|
dst: Uri,
|
||||||
) -> Result<(C::Transport, Connected), C::Error>
|
) -> Result<(C::Transport, Connected), C::Error>
|
||||||
where
|
where
|
||||||
C: Connect,
|
C: Connect,
|
||||||
@@ -676,8 +658,7 @@ mod tests {
|
|||||||
|
|
||||||
#[tokio::test]
|
#[tokio::test]
|
||||||
async fn test_errors_enforce_http() {
|
async fn test_errors_enforce_http() {
|
||||||
let uri = "https://example.domain/foo/bar?baz".parse().unwrap();
|
let dst = "https://example.domain/foo/bar?baz".parse().unwrap();
|
||||||
let dst = Destination { uri };
|
|
||||||
let connector = HttpConnector::new();
|
let connector = HttpConnector::new();
|
||||||
|
|
||||||
let err = connect(connector, dst).await.unwrap_err();
|
let err = connect(connector, dst).await.unwrap_err();
|
||||||
@@ -686,8 +667,7 @@ mod tests {
|
|||||||
|
|
||||||
#[tokio::test]
|
#[tokio::test]
|
||||||
async fn test_errors_missing_scheme() {
|
async fn test_errors_missing_scheme() {
|
||||||
let uri = "example.domain".parse().unwrap();
|
let dst = "example.domain".parse().unwrap();
|
||||||
let dst = Destination { uri };
|
|
||||||
let mut connector = HttpConnector::new();
|
let mut connector = HttpConnector::new();
|
||||||
connector.enforce_http(false);
|
connector.enforce_http(false);
|
||||||
|
|
||||||
|
|||||||
@@ -5,24 +5,14 @@
|
|||||||
//! - A default [`HttpConnector`](HttpConnector) that does DNS resolution and
|
//! - A default [`HttpConnector`](HttpConnector) that does DNS resolution and
|
||||||
//! establishes connections over TCP.
|
//! establishes connections over TCP.
|
||||||
//! - Types to build custom connectors.
|
//! - Types to build custom connectors.
|
||||||
use std::convert::TryFrom;
|
use std::fmt;
|
||||||
use std::{fmt, mem};
|
|
||||||
|
|
||||||
use bytes::{BufMut, Bytes, BytesMut};
|
use ::http::{Response};
|
||||||
use ::http::{uri, Response, Uri};
|
|
||||||
|
|
||||||
#[cfg(feature = "tcp")] pub mod dns;
|
#[cfg(feature = "tcp")] pub mod dns;
|
||||||
#[cfg(feature = "tcp")] mod http;
|
#[cfg(feature = "tcp")] mod http;
|
||||||
#[cfg(feature = "tcp")] pub use self::http::{HttpConnector, HttpInfo};
|
#[cfg(feature = "tcp")] pub use self::http::{HttpConnector, HttpInfo};
|
||||||
|
|
||||||
/// A set of properties to describe where and how to try to connect.
|
|
||||||
///
|
|
||||||
/// This type is passed an argument to connectors.
|
|
||||||
#[derive(Clone, Debug)]
|
|
||||||
pub struct Destination {
|
|
||||||
pub(super) uri: Uri,
|
|
||||||
}
|
|
||||||
|
|
||||||
/// Extra information about the connected transport.
|
/// Extra information about the connected transport.
|
||||||
///
|
///
|
||||||
/// This can be used to inform recipients about things like if ALPN
|
/// This can be used to inform recipients about things like if ALPN
|
||||||
@@ -42,205 +32,6 @@ pub(super) enum Alpn {
|
|||||||
None,
|
None,
|
||||||
}
|
}
|
||||||
|
|
||||||
impl Destination {
|
|
||||||
/// Try to convert a `Uri` into a `Destination`
|
|
||||||
///
|
|
||||||
/// # Error
|
|
||||||
///
|
|
||||||
/// Returns an error if the uri contains no authority or
|
|
||||||
/// no scheme.
|
|
||||||
pub fn try_from_uri(uri: Uri) -> crate::Result<Self> {
|
|
||||||
uri.authority().ok_or(crate::error::Parse::Uri)?;
|
|
||||||
uri.scheme().ok_or(crate::error::Parse::Uri)?;
|
|
||||||
Ok(Destination { uri })
|
|
||||||
}
|
|
||||||
|
|
||||||
/// Get the protocol scheme.
|
|
||||||
#[inline]
|
|
||||||
pub fn scheme(&self) -> &str {
|
|
||||||
self.uri
|
|
||||||
.scheme_str()
|
|
||||||
.unwrap_or("")
|
|
||||||
}
|
|
||||||
|
|
||||||
/// Get the hostname.
|
|
||||||
#[inline]
|
|
||||||
pub fn host(&self) -> &str {
|
|
||||||
self.uri
|
|
||||||
.host()
|
|
||||||
.unwrap_or("")
|
|
||||||
}
|
|
||||||
|
|
||||||
/// Get the port, if specified.
|
|
||||||
#[inline]
|
|
||||||
pub fn port(&self) -> Option<u16> {
|
|
||||||
self.uri.port_u16()
|
|
||||||
}
|
|
||||||
|
|
||||||
/// Update the scheme of this destination.
|
|
||||||
///
|
|
||||||
/// # Example
|
|
||||||
///
|
|
||||||
/// ```rust
|
|
||||||
/// # use hyper::client::connect::Destination;
|
|
||||||
/// # fn with_dst(mut dst: Destination) {
|
|
||||||
/// // let mut dst = some_destination...
|
|
||||||
/// // Change from "http://"...
|
|
||||||
/// assert_eq!(dst.scheme(), "http");
|
|
||||||
///
|
|
||||||
/// // to "ws://"...
|
|
||||||
/// dst.set_scheme("ws");
|
|
||||||
/// assert_eq!(dst.scheme(), "ws");
|
|
||||||
/// # }
|
|
||||||
/// ```
|
|
||||||
///
|
|
||||||
/// # Error
|
|
||||||
///
|
|
||||||
/// Returns an error if the string is not a valid scheme.
|
|
||||||
pub fn set_scheme(&mut self, scheme: &str) -> crate::Result<()> {
|
|
||||||
let scheme = scheme.parse().map_err(crate::error::Parse::from)?;
|
|
||||||
self.update_uri(move |parts| {
|
|
||||||
parts.scheme = Some(scheme);
|
|
||||||
})
|
|
||||||
}
|
|
||||||
|
|
||||||
/// Update the host of this destination.
|
|
||||||
///
|
|
||||||
/// # Example
|
|
||||||
///
|
|
||||||
/// ```rust
|
|
||||||
/// # use hyper::client::connect::Destination;
|
|
||||||
/// # fn with_dst(mut dst: Destination) {
|
|
||||||
/// // let mut dst = some_destination...
|
|
||||||
/// // Change from "hyper.rs"...
|
|
||||||
/// assert_eq!(dst.host(), "hyper.rs");
|
|
||||||
///
|
|
||||||
/// // to "some.proxy"...
|
|
||||||
/// dst.set_host("some.proxy");
|
|
||||||
/// assert_eq!(dst.host(), "some.proxy");
|
|
||||||
/// # }
|
|
||||||
/// ```
|
|
||||||
///
|
|
||||||
/// # Error
|
|
||||||
///
|
|
||||||
/// Returns an error if the string is not a valid hostname.
|
|
||||||
pub fn set_host(&mut self, host: &str) -> crate::Result<()> {
|
|
||||||
// Prevent any userinfo setting, it's bad!
|
|
||||||
if host.contains('@') {
|
|
||||||
return Err(crate::error::Parse::Uri.into());
|
|
||||||
}
|
|
||||||
let auth = if let Some(port) = self.port() {
|
|
||||||
let bytes = Bytes::from(format!("{}:{}", host, port));
|
|
||||||
uri::Authority::from_maybe_shared(bytes)
|
|
||||||
.map_err(crate::error::Parse::from)?
|
|
||||||
} else {
|
|
||||||
let auth = host.parse::<uri::Authority>().map_err(crate::error::Parse::from)?;
|
|
||||||
if auth.port().is_some() { // std::uri::Authority::Uri
|
|
||||||
return Err(crate::error::Parse::Uri.into());
|
|
||||||
}
|
|
||||||
auth
|
|
||||||
};
|
|
||||||
self.update_uri(move |parts| {
|
|
||||||
parts.authority = Some(auth);
|
|
||||||
})
|
|
||||||
}
|
|
||||||
|
|
||||||
/// Update the port of this destination.
|
|
||||||
///
|
|
||||||
/// # Example
|
|
||||||
///
|
|
||||||
/// ```rust
|
|
||||||
/// # use hyper::client::connect::Destination;
|
|
||||||
/// # fn with_dst(mut dst: Destination) {
|
|
||||||
/// // let mut dst = some_destination...
|
|
||||||
/// // Change from "None"...
|
|
||||||
/// assert_eq!(dst.port(), None);
|
|
||||||
///
|
|
||||||
/// // to "4321"...
|
|
||||||
/// dst.set_port(4321);
|
|
||||||
/// assert_eq!(dst.port(), Some(4321));
|
|
||||||
///
|
|
||||||
/// // Or remove the port...
|
|
||||||
/// dst.set_port(None);
|
|
||||||
/// assert_eq!(dst.port(), None);
|
|
||||||
/// # }
|
|
||||||
/// ```
|
|
||||||
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();
|
|
||||||
// Need space to copy the hostname, plus ':',
|
|
||||||
// plus max 5 port digits...
|
|
||||||
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_maybe_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) -> crate::Result<()>
|
|
||||||
where
|
|
||||||
F: FnOnce(&mut uri::Parts)
|
|
||||||
{
|
|
||||||
// Need to store a default Uri while we modify the current one...
|
|
||||||
let old_uri = mem::replace(&mut self.uri, Uri::default());
|
|
||||||
// However, mutate a clone, so we can revert if there's an error...
|
|
||||||
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(crate::error::Parse::from(err).into())
|
|
||||||
},
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
/*
|
|
||||||
/// Returns whether this connection must negotiate HTTP/2 via ALPN.
|
|
||||||
pub fn must_h2(&self) -> bool {
|
|
||||||
match self.alpn {
|
|
||||||
Alpn::Http1 => false,
|
|
||||||
Alpn::H2 => true,
|
|
||||||
}
|
|
||||||
}
|
|
||||||
*/
|
|
||||||
}
|
|
||||||
|
|
||||||
impl TryFrom<Uri> for Destination {
|
|
||||||
type Error = crate::error::Error;
|
|
||||||
|
|
||||||
fn try_from(uri: Uri) -> Result<Self, Self::Error> {
|
|
||||||
Destination::try_from_uri(uri)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
impl Connected {
|
impl Connected {
|
||||||
/// Create new `Connected` type with empty metadata.
|
/// Create new `Connected` type with empty metadata.
|
||||||
pub fn new() -> Connected {
|
pub fn new() -> Connected {
|
||||||
@@ -372,22 +163,22 @@ where
|
|||||||
pub(super) mod sealed {
|
pub(super) mod sealed {
|
||||||
use std::error::Error as StdError;
|
use std::error::Error as StdError;
|
||||||
|
|
||||||
|
use ::http::Uri;
|
||||||
use tokio::io::{AsyncRead, AsyncWrite};
|
use tokio::io::{AsyncRead, AsyncWrite};
|
||||||
|
|
||||||
use crate::common::{Future, Unpin};
|
use crate::common::{Future, Unpin};
|
||||||
use super::{Connected, Destination};
|
use super::{Connected};
|
||||||
|
|
||||||
/// Connect to a destination, returning an IO transport.
|
/// Connect to a destination, returning an IO transport.
|
||||||
///
|
///
|
||||||
/// A connector receives a [`Destination`](Destination) describing how a
|
/// A connector receives a [`Uri`](::http::Uri) and returns a `Future` of the
|
||||||
/// connection should be estabilished, and returns a `Future` of the
|
|
||||||
/// ready connection.
|
/// ready connection.
|
||||||
///
|
///
|
||||||
/// # Trait Alias
|
/// # Trait Alias
|
||||||
///
|
///
|
||||||
/// This is really just an *alias* for the `tower::Service` trait, with
|
/// This is really just an *alias* for the `tower::Service` trait, with
|
||||||
/// additional bounds set for convenience *inside* hyper. You don't actually
|
/// additional bounds set for convenience *inside* hyper. You don't actually
|
||||||
/// implement this trait, but `tower::Service<Destination>` instead.
|
/// implement this trait, but `tower::Service<Uri>` instead.
|
||||||
// The `Sized` bound is to prevent creating `dyn Connect`, since they cannot
|
// The `Sized` bound is to prevent creating `dyn Connect`, since they cannot
|
||||||
// fit the `Connect` bounds because of the blanket impl for `Service`.
|
// fit the `Connect` bounds because of the blanket impl for `Service`.
|
||||||
pub trait Connect: Sealed + Sized {
|
pub trait Connect: Sealed + Sized {
|
||||||
@@ -398,27 +189,27 @@ pub(super) mod sealed {
|
|||||||
/// A Future that will resolve to the connected Transport.
|
/// A Future that will resolve to the connected Transport.
|
||||||
type Future: Future<Output=Result<(Self::Transport, Connected), Self::Error>>;
|
type Future: Future<Output=Result<(Self::Transport, Connected), Self::Error>>;
|
||||||
#[doc(hidden)]
|
#[doc(hidden)]
|
||||||
fn connect(self, internal_only: Internal, dst: Destination) -> Self::Future;
|
fn connect(self, internal_only: Internal, dst: Uri) -> Self::Future;
|
||||||
}
|
}
|
||||||
|
|
||||||
impl<S, T> Connect for S
|
impl<S, T> Connect for S
|
||||||
where
|
where
|
||||||
S: tower_service::Service<Destination, Response=(T, Connected)> + Send,
|
S: tower_service::Service<Uri, Response=(T, Connected)> + Send,
|
||||||
S::Error: Into<Box<dyn StdError + Send + Sync>>,
|
S::Error: Into<Box<dyn StdError + Send + Sync>>,
|
||||||
S::Future: Unpin + Send,
|
S::Future: Unpin + Send,
|
||||||
T: AsyncRead + AsyncWrite + Unpin + Send + 'static,
|
T: AsyncRead + AsyncWrite + Unpin + Send + 'static,
|
||||||
{
|
{
|
||||||
type Transport = T;
|
type Transport = T;
|
||||||
type Error = S::Error;
|
type Error = S::Error;
|
||||||
type Future = crate::service::Oneshot<S, Destination>;
|
type Future = crate::service::Oneshot<S, Uri>;
|
||||||
fn connect(self, _: Internal, dst: Destination) -> Self::Future {
|
fn connect(self, _: Internal, dst: Uri) -> Self::Future {
|
||||||
crate::service::oneshot(self, dst)
|
crate::service::oneshot(self, dst)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
impl<S, T> Sealed for S
|
impl<S, T> Sealed for S
|
||||||
where
|
where
|
||||||
S: tower_service::Service<Destination, Response=(T, Connected)> + Send,
|
S: tower_service::Service<Uri, Response=(T, Connected)> + Send,
|
||||||
S::Error: Into<Box<dyn StdError + Send + Sync>>,
|
S::Error: Into<Box<dyn StdError + Send + Sync>>,
|
||||||
S::Future: Unpin + Send,
|
S::Future: Unpin + Send,
|
||||||
T: AsyncRead + AsyncWrite + Unpin + Send + 'static,
|
T: AsyncRead + AsyncWrite + Unpin + Send + 'static,
|
||||||
@@ -431,165 +222,7 @@ pub(super) mod sealed {
|
|||||||
|
|
||||||
#[cfg(test)]
|
#[cfg(test)]
|
||||||
mod tests {
|
mod tests {
|
||||||
use super::{Connected, Destination, TryFrom};
|
use super::{Connected};
|
||||||
|
|
||||||
#[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");
|
|
||||||
|
|
||||||
// Check port isn't snuck into `set_host`.
|
|
||||||
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(), None, "error doesn't modify dst");
|
|
||||||
|
|
||||||
// Check userinfo isn't snuck into `set_host`.
|
|
||||||
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(), None, "error doesn't modify dst");
|
|
||||||
|
|
||||||
// Allow IPv6 hosts
|
|
||||||
dst.set_host("[::1]").expect("set_host with IPv6");
|
|
||||||
assert_eq!(dst.host(), "[::1]");
|
|
||||||
assert_eq!(dst.port(), None, "IPv6 didn't affect port");
|
|
||||||
|
|
||||||
// However, IPv6 with a port is rejected.
|
|
||||||
dst.set_host("[::2]:1337").expect_err("set_host with IPv6 and sneaky port");
|
|
||||||
assert_eq!(dst.host(), "[::1]");
|
|
||||||
assert_eq!(dst.port(), None);
|
|
||||||
|
|
||||||
// -----------------
|
|
||||||
|
|
||||||
// Also test that an exist port is set correctly.
|
|
||||||
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");
|
|
||||||
|
|
||||||
// Check port isn't snuck into `set_host`.
|
|
||||||
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");
|
|
||||||
|
|
||||||
// Check userinfo isn't snuck into `set_host`.
|
|
||||||
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");
|
|
||||||
|
|
||||||
// Allow IPv6 hosts
|
|
||||||
dst.set_host("[::1]").expect("set_host with IPv6");
|
|
||||||
assert_eq!(dst.host(), "[::1]");
|
|
||||||
assert_eq!(dst.port(), Some(8080), "IPv6 didn't affect port");
|
|
||||||
|
|
||||||
// However, IPv6 with a port is rejected.
|
|
||||||
dst.set_host("[::2]:1337").expect_err("set_host with IPv6 and sneaky port");
|
|
||||||
assert_eq!(dst.host(), "[::1]");
|
|
||||||
assert_eq!(dst.port(), Some(8080));
|
|
||||||
}
|
|
||||||
|
|
||||||
#[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));
|
|
||||||
|
|
||||||
// Also test that an exist port is set correctly.
|
|
||||||
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);
|
|
||||||
}
|
|
||||||
|
|
||||||
#[test]
|
|
||||||
fn test_try_from_destination() {
|
|
||||||
let uri: http::Uri = "http://hyper.rs".parse().expect("initial parse");
|
|
||||||
let result = Destination::try_from(uri);
|
|
||||||
assert_eq!(result.is_ok(), true);
|
|
||||||
}
|
|
||||||
|
|
||||||
#[test]
|
|
||||||
fn test_try_from_no_scheme() {
|
|
||||||
let uri: http::Uri = "hyper.rs".parse().expect("initial parse error");
|
|
||||||
let result = Destination::try_from(uri);
|
|
||||||
assert_eq!(result.is_err(), true);
|
|
||||||
}
|
|
||||||
|
|
||||||
#[derive(Clone, Debug, PartialEq)]
|
#[derive(Clone, Debug, PartialEq)]
|
||||||
struct Ex1(usize);
|
struct Ex1(usize);
|
||||||
|
|||||||
@@ -71,7 +71,7 @@ use http::uri::Scheme;
|
|||||||
|
|
||||||
use crate::body::{Body, Payload};
|
use crate::body::{Body, Payload};
|
||||||
use crate::common::{lazy as hyper_lazy, BoxSendFuture, Executor, Lazy, Future, Pin, Poll, task};
|
use crate::common::{lazy as hyper_lazy, BoxSendFuture, Executor, Lazy, Future, Pin, Poll, task};
|
||||||
use self::connect::{Alpn, sealed::Connect, Connected, Destination};
|
use self::connect::{Alpn, sealed::Connect, Connected};
|
||||||
use self::pool::{Key as PoolKey, Pool, Poolable, Pooled, Reservation};
|
use self::pool::{Key as PoolKey, Pool, Poolable, Pooled, Reservation};
|
||||||
|
|
||||||
#[cfg(feature = "tcp")] pub use self::connect::HttpConnector;
|
#[cfg(feature = "tcp")] pub use self::connect::HttpConnector;
|
||||||
@@ -462,9 +462,7 @@ where C: Connect + Clone + Send + Sync + 'static,
|
|||||||
let ver = self.config.ver;
|
let ver = self.config.ver;
|
||||||
let is_ver_h2 = ver == Ver::Http2;
|
let is_ver_h2 = ver == Ver::Http2;
|
||||||
let connector = self.connector.clone();
|
let connector = self.connector.clone();
|
||||||
let dst = Destination {
|
let dst = uri;
|
||||||
uri,
|
|
||||||
};
|
|
||||||
hyper_lazy(move || {
|
hyper_lazy(move || {
|
||||||
// Try to take a "connecting lock".
|
// Try to take a "connecting lock".
|
||||||
//
|
//
|
||||||
|
|||||||
@@ -940,11 +940,12 @@ mod dispatch_impl {
|
|||||||
use futures_channel::{mpsc, oneshot};
|
use futures_channel::{mpsc, oneshot};
|
||||||
use futures_util::future::{FutureExt, TryFutureExt};
|
use futures_util::future::{FutureExt, TryFutureExt};
|
||||||
use futures_util::stream::{StreamExt};
|
use futures_util::stream::{StreamExt};
|
||||||
|
use http::Uri;
|
||||||
use tokio::runtime::Runtime;
|
use tokio::runtime::Runtime;
|
||||||
use tokio::io::{AsyncRead, AsyncWrite};
|
use tokio::io::{AsyncRead, AsyncWrite};
|
||||||
use tokio::net::TcpStream;
|
use tokio::net::TcpStream;
|
||||||
|
|
||||||
use hyper::client::connect::{Connected, Destination, HttpConnector};
|
use hyper::client::connect::{Connected, HttpConnector};
|
||||||
use hyper::Client;
|
use hyper::Client;
|
||||||
|
|
||||||
#[test]
|
#[test]
|
||||||
@@ -1738,19 +1739,19 @@ mod dispatch_impl {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
impl hyper::service::Service<Destination> for DebugConnector {
|
impl hyper::service::Service<Uri> for DebugConnector {
|
||||||
type Response = (DebugStream, Connected);
|
type Response = (DebugStream, Connected);
|
||||||
type Error = <HttpConnector as hyper::service::Service<Destination>>::Error;
|
type Error = <HttpConnector as hyper::service::Service<Uri>>::Error;
|
||||||
type Future = Pin<Box<dyn Future<
|
type Future = Pin<Box<dyn Future<
|
||||||
Output = Result<Self::Response, Self::Error>
|
Output = Result<Self::Response, Self::Error>
|
||||||
> + Send>>;
|
> + Send>>;
|
||||||
|
|
||||||
fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
|
fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
|
||||||
// don't forget to check inner service is ready :)
|
// don't forget to check inner service is ready :)
|
||||||
hyper::service::Service::<Destination>::poll_ready(&mut self.http, cx)
|
hyper::service::Service::<Uri>::poll_ready(&mut self.http, cx)
|
||||||
}
|
}
|
||||||
|
|
||||||
fn call(&mut self, dst: Destination) -> Self::Future {
|
fn call(&mut self, dst: Uri) -> Self::Future {
|
||||||
self.connects.fetch_add(1, Ordering::SeqCst);
|
self.connects.fetch_add(1, Ordering::SeqCst);
|
||||||
let closes = self.closes.clone();
|
let closes = self.closes.clone();
|
||||||
let is_proxy = self.is_proxy;
|
let is_proxy = self.is_proxy;
|
||||||
|
|||||||
Reference in New Issue
Block a user