use super::Channel;
#[cfg(feature = "_tls-any")]
use super::ClientTlsConfig;
#[cfg(feature = "_tls-any")]
use super::service::TlsConnector;
use super::service::{self, Executor, SharedExec};
use super::uds_connector::UdsConnector;
use crate::transport::Error;
#[cfg(feature = "_tls-any")]
use crate::transport::error;
use bytes::Bytes;
use http::{HeaderValue, uri::Uri};
use hyper::rt;
use hyper_util::client::legacy::connect::HttpConnector;
#[cfg(feature = "_tls-any")]
use std::sync::Arc;
use std::{fmt, future::Future, net::IpAddr, pin::Pin, str, str::FromStr, time::Duration};
#[cfg(feature = "_tls-any")]
use tokio_rustls::rustls::client::danger::ServerCertVerifier;
use tower_service::Service;
#[derive(Clone, PartialEq, Eq, Hash)]
pub(crate) enum EndpointType {
Uri(Uri),
Uds(String),
}
#[derive(Clone)]
pub struct Endpoint {
pub(crate) uri: EndpointType,
fallback_uri: Uri,
pub(crate) origin: Option<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-any")]
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) max_frame_size: Option<u32>,
pub(crate) tcp_keepalive: Option<Duration>,
pub(crate) tcp_keepalive_interval: Option<Duration>,
pub(crate) tcp_keepalive_retries: Option<u32>,
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>,
pub(crate) http2_header_table_size: Option<u32>,
pub(crate) http2_max_header_list_size: Option<u32>,
pub(crate) connect_timeout: Option<Duration>,
pub(crate) http2_adaptive_window: Option<bool>,
pub(crate) local_address: Option<IpAddr>,
pub(crate) executor: SharedExec,
}
impl Endpoint {
#[doc(hidden)]
pub fn new<D>(dst: D) -> Result<Self, Error>
where
D: TryInto<Self>,
D::Error: Into<crate::BoxError>,
{
let me = dst.try_into().map_err(|e| Error::from_source(e.into()))?;
#[cfg(feature = "_tls-any")]
if let EndpointType::Uri(uri) = &me.uri
&& me.tls.is_none()
&& uri.scheme() == Some(&http::uri::Scheme::HTTPS)
{
return me.tls_config(ClientTlsConfig::new().with_enabled_roots());
}
Ok(me)
}
fn new_uri(uri: Uri) -> Self {
Self {
uri: EndpointType::Uri(uri.clone()),
fallback_uri: uri,
origin: None,
user_agent: None,
concurrency_limit: None,
rate_limit: None,
timeout: None,
#[cfg(feature = "_tls-any")]
tls: None,
buffer_size: None,
init_stream_window_size: None,
init_connection_window_size: None,
max_frame_size: None,
tcp_keepalive: None,
tcp_keepalive_interval: None,
tcp_keepalive_retries: None,
tcp_nodelay: true,
http2_keep_alive_interval: None,
http2_keep_alive_timeout: None,
http2_keep_alive_while_idle: None,
http2_header_table_size: None,
http2_max_header_list_size: None,
connect_timeout: None,
http2_adaptive_window: None,
executor: SharedExec::tokio(),
local_address: None,
}
}
fn new_uds(uds_filepath: &str) -> Self {
Self {
uri: EndpointType::Uds(uds_filepath.to_string()),
fallback_uri: Uri::from_static("http://tonic"),
origin: None,
user_agent: None,
concurrency_limit: None,
rate_limit: None,
timeout: None,
#[cfg(feature = "_tls-any")]
tls: None,
buffer_size: None,
init_stream_window_size: None,
init_connection_window_size: None,
max_frame_size: None,
tcp_keepalive: None,
tcp_keepalive_interval: None,
tcp_keepalive_retries: None,
tcp_nodelay: true,
http2_keep_alive_interval: None,
http2_keep_alive_timeout: None,
http2_keep_alive_while_idle: None,
http2_header_table_size: None,
http2_max_header_list_size: None,
connect_timeout: None,
http2_adaptive_window: None,
executor: SharedExec::tokio(),
local_address: None,
}
}
pub fn from_static(s: &'static str) -> Self {
if s.starts_with("unix:") {
let uds_filepath = s
.strip_prefix("unix://")
.or_else(|| s.strip_prefix("unix:"))
.expect("Invalid unix domain socket URI");
Self::new_uds(uds_filepath)
} else {
let uri = Uri::from_static(s);
Self::new_uri(uri)
}
}
pub fn from_shared(s: impl Into<Bytes>) -> Result<Self, Error> {
let s = str::from_utf8(&s.into())
.map_err(|e| Error::new_invalid_uri().with(e))?
.to_string();
if s.starts_with("unix:") {
let uds_filepath = s
.strip_prefix("unix://")
.or_else(|| s.strip_prefix("unix:"))
.ok_or(Error::new_invalid_uri())?;
Ok(Self::new_uds(uds_filepath))
} else {
let uri = Uri::from_maybe_shared(s).map_err(|e| Error::new_invalid_uri().with(e))?;
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 origin(self, origin: Uri) -> Self {
Endpoint {
origin: Some(origin),
..self
}
}
pub fn timeout(self, dur: Duration) -> Self {
Endpoint {
timeout: Some(dur),
..self
}
}
pub fn connect_timeout(self, dur: Duration) -> Self {
Endpoint {
connect_timeout: Some(dur),
..self
}
}
pub fn tcp_keepalive(self, tcp_keepalive: Option<Duration>) -> Self {
Endpoint {
tcp_keepalive,
..self
}
}
pub fn tcp_keepalive_interval(self, tcp_keepalive_interval: Option<Duration>) -> Self {
Endpoint {
tcp_keepalive_interval,
..self
}
}
pub fn tcp_keepalive_retries(self, tcp_keepalive_retries: Option<u32>) -> Self {
Endpoint {
tcp_keepalive_retries,
..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
}
}
pub fn buffer_size(self, sz: impl Into<Option<usize>>) -> Self {
Endpoint {
buffer_size: sz.into(),
..self
}
}
#[cfg(feature = "_tls-any")]
pub fn tls_config(self, tls_config: ClientTlsConfig) -> Result<Self, Error> {
match &self.uri {
EndpointType::Uri(uri) => Ok(Endpoint {
tls: Some(
tls_config
.into_tls_connector(uri)
.map_err(Error::from_source)?,
),
..self
}),
EndpointType::Uds(_) => Err(Error::new(error::Kind::InvalidTlsConfigForUds)),
}
}
#[cfg(feature = "_tls-any")]
pub fn tls_config_with_verifier(
self,
tls_config: ClientTlsConfig,
verifier: Arc<dyn ServerCertVerifier>,
) -> Result<Self, Error> {
match &self.uri {
EndpointType::Uri(uri) => Ok(Endpoint {
tls: Some(
tls_config
.into_tls_connector_with_verifier(uri, verifier)
.map_err(Error::from_source)?,
),
..self
}),
EndpointType::Uds(_) => Err(Error::new(error::Kind::InvalidTlsConfigForUds)),
}
}
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 fn http2_adaptive_window(self, enabled: bool) -> Self {
Endpoint {
http2_adaptive_window: Some(enabled),
..self
}
}
pub fn http2_header_table_size(self, size: u32) -> Self {
Endpoint {
http2_header_table_size: Some(size),
..self
}
}
pub fn http2_max_header_list_size(self, size: u32) -> Self {
Endpoint {
http2_max_header_list_size: Some(size),
..self
}
}
#[must_use]
pub fn max_frame_size(self, frame_size: impl Into<Option<u32>>) -> Self {
Endpoint {
max_frame_size: frame_size.into(),
..self
}
}
pub fn executor<E>(mut self, executor: E) -> Self
where
E: Executor<Pin<Box<dyn Future<Output = ()> + Send>>> + Send + Sync + 'static,
{
self.executor = SharedExec::new(executor);
self
}
pub(crate) fn connector<C>(&self, c: C) -> service::Connector<C> {
service::Connector::new(
c,
#[cfg(feature = "_tls-any")]
self.tls.clone(),
)
}
pub fn local_address(self, addr: Option<IpAddr>) -> Self {
Endpoint {
local_address: addr,
..self
}
}
pub(crate) fn http_connector(&self) -> service::Connector<HttpConnector> {
let mut http = HttpConnector::new();
http.enforce_http(false);
http.set_nodelay(self.tcp_nodelay);
http.set_keepalive(self.tcp_keepalive);
http.set_keepalive_interval(self.tcp_keepalive_interval);
http.set_keepalive_retries(self.tcp_keepalive_retries);
http.set_connect_timeout(self.connect_timeout);
http.set_local_address(self.local_address);
self.connector(http)
}
pub(crate) fn uds_connector(&self, uds_filepath: &str) -> service::Connector<UdsConnector> {
self.connector(UdsConnector::new(uds_filepath))
}
pub async fn connect(&self) -> Result<Channel, Error> {
match &self.uri {
EndpointType::Uri(_) => Channel::connect(self.http_connector(), self.clone()).await,
EndpointType::Uds(uds_filepath) => {
Channel::connect(self.uds_connector(uds_filepath.as_str()), self.clone()).await
}
}
}
pub fn connect_lazy(&self) -> Channel {
match &self.uri {
EndpointType::Uri(_) => Channel::new(self.http_connector(), self.clone()),
EndpointType::Uds(uds_filepath) => {
Channel::new(self.uds_connector(uds_filepath.as_str()), self.clone())
}
}
}
pub async fn connect_with_connector<C>(&self, connector: C) -> Result<Channel, Error>
where
C: Service<Uri> + Send + 'static,
C::Response: rt::Read + rt::Write + Send + Unpin,
C::Future: Send,
crate::BoxError: From<C::Error> + Send,
{
let connector = self.connector(connector);
if let Some(connect_timeout) = self.connect_timeout {
let mut connector = hyper_timeout::TimeoutConnector::new(connector);
connector.set_connect_timeout(Some(connect_timeout));
Channel::connect(connector, self.clone()).await
} else {
Channel::connect(connector, self.clone()).await
}
}
pub fn connect_with_connector_lazy<C>(&self, connector: C) -> Channel
where
C: Service<Uri> + Send + 'static,
C::Response: rt::Read + rt::Write + Send + Unpin,
C::Future: Send,
crate::BoxError: From<C::Error> + Send,
{
let connector = self.connector(connector);
if let Some(connect_timeout) = self.connect_timeout {
let mut connector = hyper_timeout::TimeoutConnector::new(connector);
connector.set_connect_timeout(Some(connect_timeout));
Channel::new(connector, self.clone())
} else {
Channel::new(connector, self.clone())
}
}
pub fn uri(&self) -> &Uri {
match &self.uri {
EndpointType::Uri(uri) => uri,
EndpointType::Uds(_) => &self.fallback_uri,
}
}
pub fn get_tcp_nodelay(&self) -> bool {
self.tcp_nodelay
}
pub fn get_connect_timeout(&self) -> Option<Duration> {
self.connect_timeout
}
pub fn get_tcp_keepalive(&self) -> Option<Duration> {
self.tcp_keepalive
}
pub fn get_tcp_keepalive_interval(&self) -> Option<Duration> {
self.tcp_keepalive_interval
}
pub fn get_tcp_keepalive_retries(&self) -> Option<u32> {
self.tcp_keepalive_retries
}
}
impl From<Uri> for Endpoint {
fn from(uri: Uri) -> Self {
Self::new_uri(uri)
}
}
impl TryFrom<Bytes> for Endpoint {
type Error = Error;
fn try_from(t: Bytes) -> Result<Self, Self::Error> {
Self::from_shared(t)
}
}
impl TryFrom<String> for Endpoint {
type Error = Error;
fn try_from(t: String) -> Result<Self, Self::Error> {
Self::from_shared(t.into_bytes())
}
}
impl TryFrom<&'static str> for Endpoint {
type Error = Error;
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 = Error;
fn from_str(s: &str) -> Result<Self, Self::Err> {
Self::try_from(s.to_string())
}
}