use std::{
convert::Infallible,
error::Error as StdError,
future::Future,
sync::{Arc, RwLock},
time::Duration,
};
use cookie::Cookie;
use cookie_store::CookieStore;
use http::{HeaderMap, HeaderName, HeaderValue, Method, Request, Uri, header};
use http_body_util::{BodyExt, Full, combinators::BoxBody};
use hyper::{Response as HyperResponse, body::Incoming};
use hyper_rustls::HttpsConnector;
use hyper_util::{
client::legacy::{self, connect::HttpConnector},
rt::TokioExecutor,
};
use serde::{Serialize, de::DeserializeOwned};
use url::Url;
pub use http::{StatusCode, Version};
pub use hyper::body::Bytes;
pub type BoxError = Box<dyn StdError + Send + Sync>;
pub type StandardConnector = HttpsConnector<HttpConnector>;
pub type StandardHyperClient = legacy::Client<StandardConnector, Body>;
#[derive(Clone)]
pub struct StandardService(StandardHyperClient);
impl StandardService {
#[must_use]
pub const fn hyper_client(&self) -> &StandardHyperClient {
&self.0
}
}
impl HttpService for StandardService {
type Error = legacy::Error;
async fn send(&self, request: Request<Body>) -> Result<HyperResponse<Incoming>, Self::Error> {
self.0.request(request).await
}
}
pub type StandardClient = Client<StandardService>;
pub struct StandardClientBuilder {
cookie_store: bool,
default_headers: HeaderMap,
timeout: Option<Duration>,
pool_idle_timeout: Option<Option<Duration>>,
pool_max_idle_per_host: Option<usize>,
}
impl StandardClientBuilder {
#[must_use]
pub fn cookie_store(mut self, enabled: bool) -> Self {
self.cookie_store = enabled;
self
}
#[must_use]
pub fn default_headers(mut self, headers: HeaderMap) -> Self {
self.default_headers = headers;
self
}
#[must_use]
pub fn timeout(mut self, timeout: Duration) -> Self {
self.timeout = Some(timeout);
self
}
#[must_use]
pub fn pool_idle_timeout<D>(mut self, timeout: D) -> Self
where
D: Into<Option<Duration>>,
{
self.pool_idle_timeout = Some(timeout.into());
self
}
#[must_use]
pub fn pool_max_idle_per_host(mut self, max: usize) -> Self {
self.pool_max_idle_per_host = Some(max);
self
}
#[must_use]
pub fn build(self) -> StandardClient {
let mut hyper_builder = legacy::Client::builder(TokioExecutor::new());
if let Some(timeout) = self.pool_idle_timeout {
hyper_builder.pool_idle_timeout(timeout);
}
if let Some(max) = self.pool_max_idle_per_host {
hyper_builder.pool_max_idle_per_host(max);
}
ClientBuilder {
service: StandardService(hyper_builder.build(standard_connector())),
cookie_store: self.cookie_store,
default_headers: self.default_headers,
timeout: self.timeout,
}
.build()
}
}
#[must_use]
pub fn standard_connector() -> StandardConnector {
let _ = rustls::crypto::ring::default_provider().install_default();
hyper_rustls::HttpsConnectorBuilder::new()
.with_webpki_roots()
.https_or_http()
.enable_http1()
.enable_http2()
.build()
}
#[must_use]
pub fn standard_hyper_client() -> StandardHyperClient {
legacy::Client::builder(TokioExecutor::new()).build(standard_connector())
}
#[must_use]
pub fn standard_service() -> StandardService {
StandardService(standard_hyper_client())
}
#[must_use]
pub fn standard_client() -> StandardClient {
standard_client_builder().build()
}
#[must_use]
pub fn standard_client_builder() -> StandardClientBuilder {
StandardClientBuilder {
cookie_store: true,
default_headers: HeaderMap::new(),
timeout: None,
pool_idle_timeout: None,
pool_max_idle_per_host: None,
}
}
pub struct Body(BoxBody<Bytes, BoxError>);
impl Body {
#[must_use]
pub fn empty() -> Self {
Self::from(Bytes::new())
}
}
impl Default for Body {
fn default() -> Self {
Self::empty()
}
}
impl From<Bytes> for Body {
fn from(value: Bytes) -> Self {
Self(
Full::new(value)
.map_err(|error: Infallible| match error {})
.boxed(),
)
}
}
impl From<Vec<u8>> for Body {
fn from(value: Vec<u8>) -> Self {
Self::from(Bytes::from(value))
}
}
impl From<String> for Body {
fn from(value: String) -> Self {
Self::from(Bytes::from(value))
}
}
impl From<&'static str> for Body {
fn from(value: &'static str) -> Self {
Self::from(Bytes::from_static(value.as_bytes()))
}
}
impl hyper::body::Body for Body {
type Data = Bytes;
type Error = BoxError;
fn poll_frame(
mut self: std::pin::Pin<&mut Self>,
cx: &mut std::task::Context<'_>,
) -> std::task::Poll<Option<Result<hyper::body::Frame<Self::Data>, Self::Error>>> {
std::pin::Pin::new(&mut self.0).poll_frame(cx)
}
fn is_end_stream(&self) -> bool {
self.0.is_end_stream()
}
fn size_hint(&self) -> hyper::body::SizeHint {
self.0.size_hint()
}
}
pub trait HttpService: Clone + Send + Sync + 'static {
type Error: StdError + Send + Sync + 'static;
fn send(
&self,
request: Request<Body>,
) -> impl Future<Output = Result<HyperResponse<Incoming>, Self::Error>> + Send;
}
impl<S> HttpService for S
where
S: hyper::service::Service<Request<Body>, Response = HyperResponse<Incoming>>
+ Clone
+ Send
+ Sync
+ 'static,
S::Error: StdError + Send + Sync + 'static,
S::Future: Send,
{
type Error = S::Error;
fn send(
&self,
request: Request<Body>,
) -> impl Future<Output = Result<HyperResponse<Incoming>, Self::Error>> + Send {
self.call(request)
}
}
#[derive(Debug, thiserror::Error)]
#[non_exhaustive]
pub enum Error {
#[error("invalid URL: {0}")]
Url(#[from] url::ParseError),
#[error("invalid request: {0}")]
Request(#[from] http::Error),
#[error("invalid header: {0}")]
Header(String),
#[error("HTTP transport failed")]
Transport(#[source] BoxError),
#[error("request timed out")]
Timeout,
#[error("cookie store lock was poisoned")]
CookieStore,
#[error("response body failed")]
Body(#[source] BoxError),
#[error("JSON serialization failed: {0}")]
Json(#[from] serde_json::Error),
#[error("response was not valid UTF-8: {0}")]
Utf8(#[from] std::string::FromUtf8Error),
}
#[derive(Clone)]
pub struct Client<S> {
service: S,
cookies: Option<Arc<RwLock<CookieStore>>>,
default_headers: HeaderMap,
timeout: Option<Duration>,
}
impl<S: HttpService> Client<S> {
#[must_use]
pub fn new(service: S) -> Self {
Self::builder(service).build()
}
#[must_use]
pub fn builder(service: S) -> ClientBuilder<S> {
ClientBuilder {
service,
cookie_store: true,
default_headers: HeaderMap::new(),
timeout: None,
}
}
pub fn request<U>(&self, method: Method, url: U) -> Result<RequestBuilder<S>, Error>
where
U: AsRef<str>,
{
let url = Url::parse(url.as_ref())?;
Ok(RequestBuilder {
client: self.clone(),
method,
url,
headers: HeaderMap::new(),
body: Body::empty(),
})
}
pub fn get<U: AsRef<str>>(&self, url: U) -> Result<RequestBuilder<S>, Error> {
self.request(Method::GET, url)
}
pub fn head<U: AsRef<str>>(&self, url: U) -> Result<RequestBuilder<S>, Error> {
self.request(Method::HEAD, url)
}
pub fn post<U: AsRef<str>>(&self, url: U) -> Result<RequestBuilder<S>, Error> {
self.request(Method::POST, url)
}
pub fn put<U: AsRef<str>>(&self, url: U) -> Result<RequestBuilder<S>, Error> {
self.request(Method::PUT, url)
}
pub fn patch<U: AsRef<str>>(&self, url: U) -> Result<RequestBuilder<S>, Error> {
self.request(Method::PATCH, url)
}
pub fn delete<U: AsRef<str>>(&self, url: U) -> Result<RequestBuilder<S>, Error> {
self.request(Method::DELETE, url)
}
pub async fn execute(&self, request: Request<Body>) -> Result<Response, Error> {
let url = Url::parse(&request.uri().to_string())?;
self.execute_url(url, request).await
}
async fn execute_url(&self, url: Url, mut request: Request<Body>) -> Result<Response, Error> {
for (name, value) in &self.default_headers {
if !request.headers().contains_key(name) {
request.headers_mut().insert(name, value.clone());
}
}
if !request.headers().contains_key(header::COOKIE) {
if let Some(store) = &self.cookies {
let store = store.read().map_err(|_| Error::CookieStore)?;
if let Some(value) = cookie_header(&store, &url) {
request.headers_mut().insert(
header::COOKIE,
HeaderValue::from_str(&value)
.map_err(|error| Error::Header(error.to_string()))?,
);
}
}
}
let sent = self.service.send(request);
let response = if let Some(timeout) = self.timeout {
tokio::time::timeout(timeout, sent)
.await
.map_err(|_| Error::Timeout)?
.map_err(|error| Error::Transport(Box::new(error)))?
} else {
sent.await
.map_err(|error| Error::Transport(Box::new(error)))?
};
if let Some(store) = &self.cookies {
let mut store = store.write().map_err(|_| Error::CookieStore)?;
store_response_cookies(&mut store, response.headers(), &url);
}
Ok(Response(response))
}
}
fn cookie_header(store: &CookieStore, url: &Url) -> Option<String> {
let value = store
.get_request_values(url)
.map(|(name, value)| format!("{name}={value}"))
.collect::<Vec<_>>()
.join("; ");
(!value.is_empty()).then_some(value)
}
fn store_response_cookies(store: &mut CookieStore, headers: &HeaderMap, url: &Url) {
let parsed = headers
.get_all(header::SET_COOKIE)
.iter()
.filter_map(|value| value.to_str().ok())
.filter_map(|value| Cookie::parse(value.to_owned()).ok())
.map(Cookie::into_owned);
store.store_response_cookies(parsed, url);
}
pub struct ClientBuilder<S> {
service: S,
cookie_store: bool,
default_headers: HeaderMap,
timeout: Option<Duration>,
}
impl<S> ClientBuilder<S> {
#[must_use]
pub fn cookie_store(mut self, enabled: bool) -> Self {
self.cookie_store = enabled;
self
}
#[must_use]
pub fn default_headers(mut self, headers: HeaderMap) -> Self {
self.default_headers = headers;
self
}
#[must_use]
pub fn timeout(mut self, timeout: Duration) -> Self {
self.timeout = Some(timeout);
self
}
#[must_use]
pub fn build(self) -> Client<S> {
Client {
service: self.service,
cookies: self
.cookie_store
.then(|| Arc::new(RwLock::new(CookieStore::default()))),
default_headers: self.default_headers,
timeout: self.timeout,
}
}
}
pub struct RequestBuilder<S> {
client: Client<S>,
method: Method,
url: Url,
headers: HeaderMap,
body: Body,
}
impl<S: HttpService> RequestBuilder<S> {
#[must_use]
pub fn body(mut self, body: impl Into<Body>) -> Self {
self.body = body.into();
self
}
pub fn header<K, V>(mut self, key: K, value: V) -> Result<Self, Error>
where
HeaderName: TryFrom<K>,
<HeaderName as TryFrom<K>>::Error: std::fmt::Display,
HeaderValue: TryFrom<V>,
<HeaderValue as TryFrom<V>>::Error: std::fmt::Display,
{
let key = HeaderName::try_from(key).map_err(|e| Error::Header(e.to_string()))?;
let value = HeaderValue::try_from(value).map_err(|e| Error::Header(e.to_string()))?;
self.headers.insert(key, value);
Ok(self)
}
#[must_use]
pub fn headers(mut self, headers: HeaderMap) -> Self {
self.headers.extend(headers);
self
}
pub fn json(mut self, value: &impl Serialize) -> Result<Self, Error> {
self.body = Body::from(serde_json::to_vec(value)?);
self.headers.insert(
header::CONTENT_TYPE,
HeaderValue::from_static("application/json"),
);
Ok(self)
}
pub fn build(self) -> Result<Request<Body>, Error> {
let uri: Uri = self
.url
.as_str()
.parse()
.map_err(|error: http::uri::InvalidUri| Error::Header(error.to_string()))?;
let mut request = Request::builder()
.method(self.method)
.uri(uri)
.body(self.body)?;
*request.headers_mut() = self.headers;
Ok(request)
}
pub async fn send(self) -> Result<Response, Error> {
let client = self.client.clone();
let url = self.url.clone();
client.execute_url(url, self.build()?).await
}
}
pub struct Response(HyperResponse<Incoming>);
impl Response {
#[must_use]
pub fn status(&self) -> StatusCode {
self.0.status()
}
#[must_use]
pub fn headers(&self) -> &HeaderMap {
self.0.headers()
}
pub async fn bytes(self) -> Result<Bytes, Error> {
self.0
.into_body()
.collect()
.await
.map(|body| body.to_bytes())
.map_err(|error| Error::Body(Box::new(error)))
}
pub async fn text(self) -> Result<String, Error> {
Ok(String::from_utf8(self.bytes().await?.to_vec())?)
}
pub async fn json<T: DeserializeOwned>(self) -> Result<T, Error> {
Ok(serde_json::from_slice(&self.bytes().await?)?)
}
#[must_use]
pub fn into_inner(self) -> HyperResponse<Incoming> {
self.0
}
}
#[cfg(test)]
mod tests {
use tokio::{
io::{AsyncReadExt, AsyncWriteExt},
net::TcpListener,
};
use super::*;
#[test]
fn stores_and_scopes_response_cookies() {
let url = Url::parse("https://example.com/account/login").unwrap();
let mut headers = HeaderMap::new();
headers.append(
header::SET_COOKIE,
HeaderValue::from_static("session=secret; Path=/account; Secure; HttpOnly"),
);
let mut store = CookieStore::default();
store_response_cookies(&mut store, &headers, &url);
assert_eq!(
cookie_header(&store, &url).as_deref(),
Some("session=secret")
);
assert_eq!(
cookie_header(&store, &Url::parse("https://example.com/public").unwrap()),
None
);
assert_eq!(
cookie_header(&store, &Url::parse("http://example.com/account").unwrap()),
None
);
}
#[tokio::test]
async fn standard_client_uses_tokio_http_connector() {
let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
let address = listener.local_addr().unwrap();
let server = tokio::spawn(async move {
let (mut stream, _) = listener.accept().await.unwrap();
stream
.write_all(b"HTTP/1.1 200 OK\r\nContent-Length: 2\r\nConnection: close\r\n\r\nok")
.await
.unwrap();
});
let response = standard_client()
.get(format!("http://{address}/"))
.unwrap()
.send()
.await
.unwrap();
assert_eq!(response.status(), StatusCode::OK);
assert_eq!(response.text().await.unwrap(), "ok");
server.await.unwrap();
}
#[tokio::test]
async fn standard_client_reuses_pooled_connections() {
let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
let address = listener.local_addr().unwrap();
let server = tokio::spawn(async move {
let (mut stream, _) = listener.accept().await.unwrap();
for _ in 0..2 {
let mut request = Vec::new();
while !request.ends_with(b"\r\n\r\n") {
let mut byte = [0];
stream.read_exact(&mut byte).await.unwrap();
request.push(byte[0]);
}
stream
.write_all(b"HTTP/1.1 200 OK\r\nContent-Length: 2\r\n\r\nok")
.await
.unwrap();
}
});
let client = standard_client_builder()
.pool_idle_timeout(Duration::from_secs(30))
.pool_max_idle_per_host(1)
.build();
for _ in 0..2 {
let response = client
.get(format!("http://{address}/"))
.unwrap()
.send()
.await
.unwrap();
assert_eq!(response.text().await.unwrap(), "ok");
}
server.await.unwrap();
}
}