use anyspawn::Spawner;
use fetch_hyper::HyperTransportBuilder;
use fetch_options::TransportOptions;
use fetch_tls::{TlsBackend, TlsBackendBuilder};
use http_extensions::Result;
use hyper_util::rt::TokioIo;
use templated_uri::BaseUri;
use thread_aware::ThreadAware;
use tick::Clock;
use crate::custom::{CustomContext, CustomDeps, Isolation};
use crate::handlers::TransportHandler;
use crate::tls::TlsOptions;
use crate::{HttpClient, HttpClientBuilder};
#[derive(Debug, Clone, ThreadAware)]
#[fundle::deps]
pub struct TokioDeps {
pub clock: Clock,
pub global_pool: bytesbuf::mem::GlobalPool,
}
impl Default for TokioDeps {
fn default() -> Self {
Self::with_clock(&Clock::new_tokio())
}
}
impl TokioDeps {
#[must_use]
pub fn with_clock(clock: &Clock) -> Self {
Self {
global_pool: bytesbuf::mem::GlobalPool::new(),
clock: clock.clone(),
}
}
}
impl HttpClient {
pub fn builder_tokio(deps: impl Into<TokioDeps>) -> HttpClientBuilder {
let deps = deps.into();
let clock = deps.clock.clone();
let global_pool = deps.global_pool.clone();
Self::builder_custom_internal(
|cx| TransportHandler(build_tokio_handler(cx).into()),
Isolation::Shared,
CustomDeps {
clock,
global_pool,
extras: deps,
},
)
}
#[must_use]
pub fn new_tokio() -> Self {
Self::builder_tokio(TokioDeps::default()).build()
}
}
#[derive(Clone)]
struct TokioConnector;
impl layered::Service<BaseUri> for TokioConnector {
type Out = Result<TokioIo<::tokio::net::TcpStream>>;
async fn execute(&self, input: BaseUri) -> Self::Out {
let host = input.authority().host();
let port = input.try_effective_port()?;
let stream = ::tokio::net::TcpStream::connect((host, port)).await?;
Ok(TokioIo::new(stream))
}
}
fn build_tokio_handler(cx: CustomContext<TokioDeps>) -> fetch_hyper::HyperTransport {
let tls_backend = build_tls_backend(&cx.options, cx.tls);
HyperTransportBuilder::new(TokioConnector, Spawner::new_tokio(), cx.clock, cx.options)
.body_builder(cx.body_builder)
.pool_index(cx.pool_index)
.meter(cx.meter)
.build(tls_backend)
}
fn build_tls_backend(options: &TransportOptions, tls: TlsOptions) -> TlsBackend {
let mut builder = TlsBackendBuilder::new();
if !options.supported_http_versions.is_empty() {
builder = builder.supported_http_versions(&options.supported_http_versions);
}
#[cfg(any(feature = "rustls", test))]
{
let provider = std::sync::Arc::new(::rustls::crypto::aws_lc_rs::default_provider());
let verifier = std::sync::Arc::new(
rustls_platform_verifier::Verifier::new(std::sync::Arc::clone(&provider))
.expect("the platform certificate verifier must initialize with the aws-lc-rs crypto provider"),
);
builder = builder.configure_rustls(provider, verifier);
}
#[cfg(all(feature = "native-tls", not(any(feature = "rustls", test))))]
{
builder = builder.defaults_to_native_tls();
}
builder
.build_backend(tls)
.expect("TLS backend construction must succeed for the configured TlsOptions")
}
#[cfg(test)]
#[cfg_attr(coverage_nightly, coverage(off))]
mod tests {
use http::StatusCode;
use http_extensions::FakeHandler;
use thread_aware::ThreadAware;
use thread_aware::affinity::pinned_affinities;
use tick::Clock;
use super::TokioDeps;
use crate::pipeline::Pipeline;
use crate::{HttpClient, HttpResponseBuilder};
#[cfg_attr(miri, ignore)]
#[tokio::test]
async fn test_builder_tokio() {
let clock = Clock::new_tokio();
let client = HttpClient::builder_tokio(TokioDeps::with_clock(&clock)).minimal_pipeline().build();
assert!(matches!(client.pipeline(), Pipeline::Minimal(_)));
if let Pipeline::Minimal(dispatch) = client.pipeline() {
assert!(matches!(dispatch.mode, crate::handlers::DispatchMode::Single(_)));
}
}
#[cfg_attr(miri, ignore)]
#[tokio::test]
async fn test_new_tokio() {
let clock = Clock::new_tokio();
let client = HttpClient::builder_tokio(TokioDeps::with_clock(&clock)).build();
assert!(client.pipeline().is_standard());
}
#[cfg_attr(miri, ignore)]
#[tokio::test]
async fn new_tokio_uses_default_deps() {
let client = HttpClient::new_tokio();
assert!(client.pipeline().is_standard());
}
#[cfg_attr(miri, ignore)]
#[tokio::test]
async fn tokio_client_works_after_relocation() {
let affinities = pinned_affinities(&[2]);
let clock = Clock::new_tokio();
let mut client = HttpClient::builder_tokio(TokioDeps::with_clock(&clock))
.custom_pipeline(|_root, _ctx| FakeHandler::from_fn(|_request| HttpResponseBuilder::new_fake().status(StatusCode::OK).build()))
.build();
let response = client.get("https://example.com").fetch().await.unwrap();
assert_eq!(response.status(), StatusCode::OK);
client.relocate(None, affinities[0]);
let response = client.get("https://example.com/after-relocation").fetch().await.unwrap();
assert_eq!(response.status(), StatusCode::OK);
}
#[cfg_attr(miri, ignore)]
#[test]
fn build_tls_backend_skips_empty_supported_http_versions() {
use fetch_options::TransportOptions;
use crate::tls::TlsOptions;
let mut options = TransportOptions::default();
options.supported_http_versions = Vec::new();
let _backend = super::build_tls_backend(&options, TlsOptions::default());
}
}