use super::super::service;
use super::Channel;
#[cfg(feature = "tls")]
use super::ClientTlsConfig;
#[cfg(feature = "tls")]
use crate::transport::service::TlsConnector;
use crate::transport::Error;
use bytes::Bytes;
use http::{
uri::{InvalidUri, Uri},
HeaderValue,
};
use std::{
convert::{TryFrom, TryInto},
fmt,
str::FromStr,
time::Duration,
};
use tower::make::MakeConnection;
#[derive(Clone)]
pub struct Endpoint {
pub(crate) uri: Uri,
pub(crate) user_agent: Option<HeaderValue>,
pub(crate) timeout: Option<Duration>,
pub(crate) concurrency_limit: Option<usize>,
pub(crate) rate_limit: Option<(u64, Duration)>,
#[cfg(feature = "tls")]
pub(crate) tls: Option<TlsConnector>,
pub(crate) buffer_size: Option<usize>,
pub(crate) init_stream_window_size: Option<u32>,
pub(crate) init_connection_window_size: Option<u32>,
pub(crate) tcp_keepalive: Option<Duration>,
pub(crate) tcp_nodelay: bool,
pub(crate) http2_keep_alive_interval: Option<Duration>,
pub(crate) http2_keep_alive_timeout: Option<Duration>,
pub(crate) http2_keep_alive_while_idle: Option<bool>,
}
impl Endpoint {
#[doc(hidden)]
pub fn new<D>(dst: D) -> Result<Self, Error>
where
D: TryInto<Self>,
D::Error: Into<crate::Error>,
{
let me = dst.try_into().map_err(|e| Error::from_source(e.into()))?;
Ok(me)
}
pub fn from_static(s: &'static str) -> Self {
let uri = Uri::from_static(s);
Self::from(uri)
}
pub fn from_shared(s: impl Into<Bytes>) -> Result<Self, InvalidUri> {
let uri = Uri::from_maybe_shared(s.into())?;
Ok(Self::from(uri))
}
pub fn user_agent<T>(self, user_agent: T) -> Result<Self, Error>
where
T: TryInto<HeaderValue>,
{
user_agent
.try_into()
.map(|ua| Endpoint {
user_agent: Some(ua),
..self
})
.map_err(|_| Error::new_invalid_user_agent())
}
pub fn timeout(self, dur: Duration) -> Self {
Endpoint {
timeout: Some(dur),
..self
}
}
pub fn tcp_keepalive(self, tcp_keepalive: Option<Duration>) -> Self {
Endpoint {
tcp_keepalive,
..self
}
}
pub fn concurrency_limit(self, limit: usize) -> Self {
Endpoint {
concurrency_limit: Some(limit),
..self
}
}
pub fn rate_limit(self, limit: u64, duration: Duration) -> Self {
Endpoint {
rate_limit: Some((limit, duration)),
..self
}
}
pub fn initial_stream_window_size(self, sz: impl Into<Option<u32>>) -> Self {
Endpoint {
init_stream_window_size: sz.into(),
..self
}
}
pub fn initial_connection_window_size(self, sz: impl Into<Option<u32>>) -> Self {
Endpoint {
init_connection_window_size: sz.into(),
..self
}
}
#[cfg(feature = "tls")]
#[cfg_attr(docsrs, doc(cfg(feature = "tls")))]
pub fn tls_config(self, tls_config: ClientTlsConfig) -> Result<Self, Error> {
Ok(Endpoint {
tls: Some(
tls_config
.tls_connector(self.uri.clone())
.map_err(Error::from_source)?,
),
..self
})
}
pub fn tcp_nodelay(self, enabled: bool) -> Self {
Endpoint {
tcp_nodelay: enabled,
..self
}
}
pub fn http2_keep_alive_interval(self, interval: Duration) -> Self {
Endpoint {
http2_keep_alive_interval: Some(interval),
..self
}
}
pub fn keep_alive_timeout(self, duration: Duration) -> Self {
Endpoint {
http2_keep_alive_timeout: Some(duration),
..self
}
}
pub fn keep_alive_while_idle(self, enabled: bool) -> Self {
Endpoint {
http2_keep_alive_while_idle: Some(enabled),
..self
}
}
pub async fn connect(&self) -> Result<Channel, Error> {
let mut http = hyper::client::connect::HttpConnector::new();
http.enforce_http(false);
http.set_nodelay(self.tcp_nodelay);
http.set_keepalive(self.tcp_keepalive);
#[cfg(feature = "tls")]
let connector = service::connector(http, self.tls.clone());
#[cfg(not(feature = "tls"))]
let connector = service::connector(http);
Channel::connect(connector, self.clone()).await
}
pub fn connect_lazy(&self) -> Result<Channel, Error> {
let mut http = hyper::client::connect::HttpConnector::new();
http.enforce_http(false);
http.set_nodelay(self.tcp_nodelay);
http.set_keepalive(self.tcp_keepalive);
#[cfg(feature = "tls")]
let connector = service::connector(http, self.tls.clone());
#[cfg(not(feature = "tls"))]
let connector = service::connector(http);
Ok(Channel::new(connector, self.clone()))
}
pub async fn connect_with_connector<C>(&self, connector: C) -> Result<Channel, Error>
where
C: MakeConnection<Uri> + Send + 'static,
C::Connection: Unpin + Send + 'static,
C::Future: Send + 'static,
crate::Error: From<C::Error> + Send + 'static,
{
#[cfg(feature = "tls")]
let connector = service::connector(connector, self.tls.clone());
#[cfg(not(feature = "tls"))]
let connector = service::connector(connector);
Channel::connect(connector, self.clone()).await
}
pub fn uri(&self) -> &Uri {
&self.uri
}
}
impl From<Uri> for Endpoint {
fn from(uri: Uri) -> Self {
Self {
uri,
user_agent: None,
concurrency_limit: None,
rate_limit: None,
timeout: None,
#[cfg(feature = "tls")]
tls: None,
buffer_size: None,
init_stream_window_size: None,
init_connection_window_size: None,
tcp_keepalive: None,
tcp_nodelay: true,
http2_keep_alive_interval: None,
http2_keep_alive_timeout: None,
http2_keep_alive_while_idle: None,
}
}
}
impl TryFrom<Bytes> for Endpoint {
type Error = InvalidUri;
fn try_from(t: Bytes) -> Result<Self, Self::Error> {
Self::from_shared(t)
}
}
impl TryFrom<String> for Endpoint {
type Error = InvalidUri;
fn try_from(t: String) -> Result<Self, Self::Error> {
Self::from_shared(t.into_bytes())
}
}
impl TryFrom<&'static str> for Endpoint {
type Error = InvalidUri;
fn try_from(t: &'static str) -> Result<Self, Self::Error> {
Self::from_shared(t.as_bytes())
}
}
impl fmt::Debug for Endpoint {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
f.debug_struct("Endpoint").finish()
}
}
impl FromStr for Endpoint {
type Err = InvalidUri;
fn from_str(s: &str) -> Result<Self, Self::Err> {
Self::try_from(s.to_string())
}
}