use std::collections::HashMap;
use std::sync::{Arc, Mutex};
use apollo_opentelemetry::metrics::Clock;
use apollo_opentelemetry::tower::ServiceBuilderExt as _;
use hyper_util::client::legacy::connect::HttpConnector;
use rustls::ClientConfig;
use tower::ServiceBuilder;
use tower::util::BoxCloneSyncService;
use crate::client::{HttpClient, HttpClientDispatch, HttpClientState};
use crate::config::{HttpClientConfig, Protocol};
use crate::error::HttpClientError;
use crate::metrics::HttpMetrics;
use crate::protocol::h1::H1Connector;
use crate::protocol::h2::H2Connector;
use crate::protocol::{HostSender, Origin};
use crate::spans::HttpSpans;
use crate::tls::{build_https_connector, build_tls_config, with_protocol_alpn};
pub struct HttpClientBuilder {
config: HttpClientConfig,
tls_override: Option<Arc<ClientConfig>>,
}
impl HttpClientBuilder {
pub(crate) fn new(config: HttpClientConfig) -> Self {
Self {
config,
tls_override: None,
}
}
pub fn with_tls_config(mut self, tls: Arc<ClientConfig>) -> Self {
self.tls_override = Some(tls);
self
}
pub fn build(self) -> Result<HttpClient, HttpClientError> {
self.build_with_clock(Clock::new())
}
pub(crate) fn build_with_clock(self, clock: Clock) -> Result<HttpClient, HttpClientError> {
let HttpClientBuilder {
config,
tls_override,
} = self;
apollo_configuration::validate(config.clone())
.map_err(|source| HttpClientError::InvalidConfig { source })?;
let mut http = HttpConnector::new();
http.enforce_http(false);
http.set_nodelay(config.tcp.nodelay);
http.set_keepalive(Some(*config.tcp.keepalive));
let tls_config: Arc<ClientConfig> = match tls_override {
Some(ref base) => Arc::new(with_protocol_alpn(base, config.protocol)),
None => build_tls_config(config.protocol, &config.tls)?,
};
let connect_timeout = *config.connect_timeout;
let configured_protocol_version = match config.protocol {
Protocol::Http2 => http::Version::HTTP_2,
Protocol::Http1 | Protocol::Alpn => http::Version::HTTP_11,
};
let metrics = HttpMetrics::new(
clock,
config.telemetry.metrics.request_body_size,
config.telemetry.metrics.response_body_size,
config.telemetry.metrics.url_scheme,
);
let conn_metrics = metrics.connection_metrics();
let h1_max_idle = config.http1.pool_max_idle_per_host.get();
let h1_idle_timeout = *config.http1.pool_idle_timeout;
let h2_idle_timeout = *config.http2.connection_idle_timeout;
#[cfg(unix)]
let unix_config = Arc::new(crate::protocol::unix::UnixSenderConfig {
protocol: config.protocol,
connect_timeout,
h1_max_idle,
h1_idle_timeout,
h2_idle_timeout,
keep_alive_interval: *config.http2.keep_alive_interval,
keep_alive_timeout: *config.http2.keep_alive_timeout,
keep_alive_while_idle: config.http2.keep_alive_while_idle,
conn_metrics: conn_metrics.clone(),
});
let tcp_make_sender: Box<dyn Fn(Origin) -> HostSender + Send + Sync> =
if let Some(proxy) = &config.proxy {
let proxy_url = proxy.url.unredact();
let proxy_host = proxy_url
.host_str()
.expect("validate above rejects URLs with no host")
.to_string();
let proxy_port = proxy_url.port_or_known_default().expect(
"validate above rejects non-http(s) schemes, both of which have default ports",
);
let conn_metrics = conn_metrics.with_peer(proxy_host.as_str(), proxy_port);
let proxy_config = Arc::new(crate::proxy::ProxySenderConfig {
proxy_host,
proxy_port,
proxy_auth: crate::proxy::proxy_authorization(proxy_url),
tls_config,
protocol: config.protocol,
connect_timeout,
h1_max_idle,
h1_idle_timeout,
h2_idle_timeout,
keep_alive_interval: *config.http2.keep_alive_interval,
keep_alive_timeout: *config.http2.keep_alive_timeout,
keep_alive_while_idle: config.http2.keep_alive_while_idle,
conn_metrics,
});
Box::new(move |key| crate::proxy::new_sender(&proxy_config, key))
} else {
let https = build_https_connector(http, tls_config, config.protocol);
let h1_connector =
H1Connector::new(https.clone(), conn_metrics.clone(), connect_timeout);
let h2_connector = H2Connector::new(
https.clone(),
conn_metrics.clone(),
connect_timeout,
*config.http2.keep_alive_interval,
*config.http2.keep_alive_timeout,
config.http2.keep_alive_while_idle,
);
match config.protocol {
Protocol::Http2 => Box::new(move |key| {
crate::protocol::h2::new_sender(
h2_connector.clone(),
h2_idle_timeout,
key,
|_| {},
)
}),
Protocol::Http1 => Box::new(move |key| {
crate::protocol::h1::new_sender(
h1_connector.clone(),
h1_max_idle,
h1_idle_timeout,
key,
|_req| {},
)
}),
Protocol::Alpn => Box::new(move |key| {
crate::protocol::alpn::new_sender(
h1_connector.clone(),
h1_max_idle,
h1_idle_timeout,
h2_connector.clone(),
h2_idle_timeout,
key,
)
}),
}
};
let make_sender: Box<dyn Fn(Origin) -> HostSender + Send + Sync> = Box::new(move |key| {
if key.scheme.as_str() == "unix" {
#[cfg(unix)]
{
crate::protocol::unix::new_sender(&unix_config, key)
}
#[cfg(not(unix))]
{
crate::client::unsupported_scheme_sender()
}
} else {
tcp_make_sender(key)
}
});
let spans = HttpSpans::new(config.telemetry.spans.clone());
let dispatch = HttpClientDispatch {
state: Arc::new(Mutex::new(HttpClientState {
pools: HashMap::new(),
make_sender,
})),
metrics,
spans: spans.clone(),
configured_protocol_version,
};
let service = ServiceBuilder::new()
.traced(
apollo_opentelemetry::default_instrumentation_scope!(),
spans,
)
.http_client_propagation()
.service(dispatch);
let inner = BoxCloneSyncService::new(service);
Ok(HttpClient::from_service(inner))
}
}