use crate::{
agent::{self, AgentBuilder},
auth::{Authentication, Credentials},
config::internal::{ConfigurableBase, SetOpt},
config::*,
handler::{RequestHandler, ResponseBodyReader},
middleware::Middleware,
task::Join,
Body, Error,
};
use futures_io::AsyncRead;
use futures_util::{future::BoxFuture, pin_mut};
use http::{Request, Response};
use lazy_static::lazy_static;
use std::{
convert::TryFrom,
fmt,
future::Future,
io,
pin::Pin,
sync::Arc,
task::{Context, Poll},
};
lazy_static! {
static ref USER_AGENT: String = format!(
"curl/{} isahc/{}",
curl::Version::get().version(),
env!("CARGO_PKG_VERSION")
);
}
pub struct HttpClientBuilder {
agent_builder: AgentBuilder,
defaults: http::Extensions,
middleware: Vec<Box<dyn Middleware>>,
}
impl Default for HttpClientBuilder {
fn default() -> Self {
Self::new()
}
}
impl HttpClientBuilder {
pub fn new() -> Self {
let mut defaults = http::Extensions::new();
defaults.insert(VersionNegotiation::default());
defaults.insert(Authentication::default());
Self {
agent_builder: AgentBuilder::default(),
defaults,
middleware: Vec::new(),
}
}
#[cfg(feature = "cookies")]
pub fn cookies(self) -> Self {
self.middleware_impl(crate::cookies::CookieJar::default())
}
#[cfg(feature = "middleware-api-preview")]
pub fn middleware(self, middleware: impl Middleware) -> Self {
self.middleware_impl(middleware)
}
#[allow(unused)]
fn middleware_impl(mut self, middleware: impl Middleware) -> Self {
self.middleware.push(Box::new(middleware));
self
}
pub fn max_connections(mut self, max: usize) -> Self {
self.agent_builder = self.agent_builder.max_connections(max);
self
}
pub fn max_connections_per_host(mut self, max: usize) -> Self {
self.agent_builder = self.agent_builder.max_connections_per_host(max);
self
}
pub fn connection_cache_size(mut self, size: usize) -> Self {
self.agent_builder = self.agent_builder.connection_cache_size(size);
self.defaults.insert(CloseConnection(size == 0));
self
}
pub fn dns_cache(self, cache: impl Into<DnsCache>) -> Self {
self.configure(cache.into())
}
pub fn dns_resolve(self, map: ResolveMap) -> Self {
self.configure(map)
}
pub fn build(self) -> Result<HttpClient, Error> {
Ok(HttpClient {
agent: Arc::new(self.agent_builder.spawn()?),
defaults: self.defaults,
middleware: self.middleware,
})
}
}
impl Configurable for HttpClientBuilder {}
impl ConfigurableBase for HttpClientBuilder {
fn configure(mut self, option: impl Send + Sync + 'static) -> Self {
self.defaults.insert(option);
self
}
}
impl fmt::Debug for HttpClientBuilder {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
f.debug_struct("HttpClientBuilder").finish()
}
}
pub struct HttpClient {
agent: Arc<agent::Handle>,
defaults: http::Extensions,
middleware: Vec<Box<dyn Middleware>>,
}
impl HttpClient {
pub fn new() -> Result<Self, Error> {
HttpClientBuilder::default().build()
}
pub(crate) fn shared() -> &'static Self {
lazy_static! {
static ref SHARED: HttpClient =
HttpClient::new().expect("shared client failed to initialize");
}
&SHARED
}
pub fn builder() -> HttpClientBuilder {
HttpClientBuilder::default()
}
#[inline]
pub fn get<U>(&self, uri: U) -> Result<Response<Body>, Error>
where
http::Uri: TryFrom<U>,
<http::Uri as TryFrom<U>>::Error: Into<http::Error>,
{
self.get_async(uri).join()
}
pub fn get_async<U>(&self, uri: U) -> ResponseFuture<'_>
where
http::Uri: TryFrom<U>,
<http::Uri as TryFrom<U>>::Error: Into<http::Error>,
{
self.send_builder_async(http::Request::get(uri), Body::empty())
}
#[inline]
pub fn head<U>(&self, uri: U) -> Result<Response<Body>, Error>
where
http::Uri: TryFrom<U>,
<http::Uri as TryFrom<U>>::Error: Into<http::Error>,
{
self.head_async(uri).join()
}
pub fn head_async<U>(&self, uri: U) -> ResponseFuture<'_>
where
http::Uri: TryFrom<U>,
<http::Uri as TryFrom<U>>::Error: Into<http::Error>,
{
self.send_builder_async(http::Request::head(uri), Body::empty())
}
#[inline]
pub fn post<U>(&self, uri: U, body: impl Into<Body>) -> Result<Response<Body>, Error>
where
http::Uri: TryFrom<U>,
<http::Uri as TryFrom<U>>::Error: Into<http::Error>,
{
self.post_async(uri, body).join()
}
pub fn post_async<U>(&self, uri: U, body: impl Into<Body>) -> ResponseFuture<'_>
where
http::Uri: TryFrom<U>,
<http::Uri as TryFrom<U>>::Error: Into<http::Error>,
{
self.send_builder_async(http::Request::post(uri), body.into())
}
#[inline]
pub fn put<U>(&self, uri: U, body: impl Into<Body>) -> Result<Response<Body>, Error>
where
http::Uri: TryFrom<U>,
<http::Uri as TryFrom<U>>::Error: Into<http::Error>,
{
self.put_async(uri, body).join()
}
pub fn put_async<U>(&self, uri: U, body: impl Into<Body>) -> ResponseFuture<'_>
where
http::Uri: TryFrom<U>,
<http::Uri as TryFrom<U>>::Error: Into<http::Error>,
{
self.send_builder_async(http::Request::put(uri), body.into())
}
#[inline]
pub fn delete<U>(&self, uri: U) -> Result<Response<Body>, Error>
where
http::Uri: TryFrom<U>,
<http::Uri as TryFrom<U>>::Error: Into<http::Error>,
{
self.delete_async(uri).join()
}
pub fn delete_async<U>(&self, uri: U) -> ResponseFuture<'_>
where
http::Uri: TryFrom<U>,
<http::Uri as TryFrom<U>>::Error: Into<http::Error>,
{
self.send_builder_async(http::Request::delete(uri), Body::empty())
}
#[inline]
pub fn send<B: Into<Body>>(&self, request: Request<B>) -> Result<Response<Body>, Error> {
self.send_async(request).join()
}
pub fn send_async<B: Into<Body>>(&self, request: Request<B>) -> ResponseFuture<'_> {
let request = request.map(Into::into);
ResponseFuture::new(self.send_async_inner(request))
}
fn send_builder_async(
&self,
builder: http::request::Builder,
body: Body,
) -> ResponseFuture<'_> {
ResponseFuture::new(async move { self.send_async_inner(builder.body(body)?).await })
}
async fn send_async_inner(&self, mut request: Request<Body>) -> Result<Response<Body>, Error> {
request
.headers_mut()
.entry(http::header::USER_AGENT)
.or_insert(USER_AGENT.parse().unwrap());
for middleware in self.middleware.iter().rev() {
request = middleware.filter_request(request);
}
let (easy, future) = self.create_easy_handle(request)?;
self.agent.submit_request(easy)?;
let response = future.await?;
let content_length = response
.headers()
.get(http::header::CONTENT_LENGTH)
.and_then(|v| v.to_str().ok())
.and_then(|v| v.parse().ok());
let mut response = response.map(|reader| {
let body = ResponseBody {
inner: reader,
_agent: self.agent.clone(),
};
if let Some(len) = content_length {
Body::from_reader_sized(body, len)
} else {
Body::from_reader(body)
}
});
for middleware in self.middleware.iter() {
response = middleware.filter_response(response);
}
Ok(response)
}
#[allow(clippy::cognitive_complexity)]
fn create_easy_handle(
&self,
request: Request<Body>,
) -> Result<
(
curl::easy::Easy2<RequestHandler>,
impl Future<Output = Result<Response<ResponseBodyReader>, Error>>,
),
Error,
> {
let (mut parts, body) = request.into_parts();
let has_body = !body.is_empty();
let body_length = body.len();
let (handler, future) = RequestHandler::new(body);
let mut easy = curl::easy::Easy2::new(handler);
easy.verbose(log::log_enabled!(log::Level::Debug))?;
easy.signal(false)?;
macro_rules! set_opts {
($easy:expr, $extensions:expr, $defaults:expr, [$($option:ty,)*]) => {{
$(
if let Some(extension) = $extensions.get::<$option>().or_else(|| $defaults.get()) {
extension.set_opt($easy)?;
}
)*
}};
}
set_opts!(
&mut easy,
parts.extensions,
self.defaults,
[
Timeout,
ConnectTimeout,
TcpKeepAlive,
TcpNoDelay,
NetworkInterface,
RedirectPolicy,
redirect::AutoReferer,
Authentication,
Credentials,
MaxUploadSpeed,
MaxDownloadSpeed,
VersionNegotiation,
proxy::Proxy<Option<http::Uri>>,
proxy::Blacklist,
proxy::Proxy<Authentication>,
proxy::Proxy<Credentials>,
DnsCache,
dns::ResolveMap,
dns::Servers,
ssl::Ciphers,
ClientCertificate,
CaCertificate,
SslOption,
CloseConnection,
EnableMetrics,
]
);
easy.accept_encoding(
parts
.headers
.get("Accept-Encoding")
.and_then(|value| value.to_str().ok())
.unwrap_or(""),
)?;
#[allow(indirect_structural_match)]
match (&parts.method, has_body) {
(&http::Method::GET, false) => {
easy.get(true)?;
}
(&http::Method::HEAD, has_body) => {
easy.upload(has_body)?;
easy.nobody(true)?;
easy.custom_request("HEAD")?;
}
(&http::Method::POST, _) => {
easy.post(true)?;
}
(&http::Method::PUT, _) => {
easy.upload(true)?;
}
(method, has_body) => {
easy.upload(has_body)?;
easy.custom_request(method.as_str())?;
}
}
easy.url(&parts.uri.to_string())?;
if has_body {
let body_length = parts
.headers
.get("Content-Length")
.and_then(|value| value.to_str().ok())
.and_then(|value| value.parse().ok())
.or(body_length);
if let Some(len) = body_length {
if parts.method == http::Method::POST {
easy.post_field_size(len)?;
} else {
easy.in_filesize(len)?;
}
} else {
parts.headers.insert(
"Transfer-Encoding",
http::header::HeaderValue::from_static("chunked"),
);
}
}
parts.headers.set_opt(&mut easy)?;
Ok((easy, future))
}
}
impl fmt::Debug for HttpClient {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
f.debug_struct("HttpClient").finish()
}
}
pub struct ResponseFuture<'c>(BoxFuture<'c, Result<Response<Body>, Error>>);
impl<'c> ResponseFuture<'c> {
fn new(future: impl Future<Output = Result<Response<Body>, Error>> + Send + 'c) -> Self {
ResponseFuture(Box::pin(future))
}
}
impl Future for ResponseFuture<'_> {
type Output = Result<Response<Body>, Error>;
fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
use futures_util::future::FutureExt;
self.0.poll_unpin(cx)
}
}
impl<'c> fmt::Debug for ResponseFuture<'c> {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
f.debug_struct("ResponseFuture").finish()
}
}
struct ResponseBody {
inner: ResponseBodyReader,
_agent: Arc<agent::Handle>,
}
impl AsyncRead for ResponseBody {
fn poll_read(
mut self: Pin<&mut Self>,
cx: &mut Context<'_>,
buf: &mut [u8],
) -> Poll<io::Result<usize>> {
let inner = &mut self.inner;
pin_mut!(inner);
inner.poll_read(cx, buf)
}
}
#[cfg(test)]
mod tests {
use super::*;
static_assertions::assert_impl_all!(HttpClient: Send, Sync);
static_assertions::assert_impl_all!(HttpClientBuilder: Send);
}