Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
12 changes: 7 additions & 5 deletions Cargo.toml
Original file line number Diff line number Diff line change
Expand Up @@ -42,13 +42,13 @@ deflate = ["tower-http/decompression-deflate"]
json = ["dep:serde_json"]

# Enable multipart/form-data support.
multipart = ["dep:mime_guess"]
multipart = ["dep:mime_guess", "dep:sync_wrapper"]

# Enable hickory DNS resolver.
hickory-dns = ["dep:hickory-resolver"]

# Enable streaming support.
stream = ["tokio/fs", "dep:tokio-util"]
stream = ["tokio/fs", "dep:tokio-util", "dep:sync_wrapper"]

# Enable SOCKS proxy support.
socks = ["dep:tokio-socks"]
Expand All @@ -71,7 +71,6 @@ url = "2.5"
serde = { version = "1.0", features = ["derive"] }
serde_urlencoded = "0.7.1"
tower = { version = "0.5.2", default-features = false, features = ["timeout", "util", "retry"] }
sync_wrapper = { version = "1.0", features = ["futures"] }

bytes = "1.2"
http = "1"
Expand Down Expand Up @@ -109,8 +108,11 @@ mime_guess = { version = "2.0", default-features = false, optional = true }
encoding_rs = { version = "0.8", optional = true }
mime = { version = "0.3.17", optional = true }

## root certs
webpki-root-certs = { version = "1.0.0", optional = true }
## sync wrapper
sync_wrapper = { version = "1.0.2", features = ["futures"], optional = true }

## webpki root certs
webpki-root-certs = { version = "1.0.2", optional = true }

## cookies
cookie_crate = { version = "0.18", package = "cookie", optional = true }
Expand Down
2 changes: 0 additions & 2 deletions src/client/http/aliases.rs
Original file line number Diff line number Diff line change
Expand Up @@ -59,8 +59,6 @@ pub type RedirectLayer = FollowRedirect<
FollowRedirectPolicy,
>;

pub type CoreResponseFuture = crate::core::client::ResponseFuture;

pub type GenericClientService =
MapErr<Timeout<Retry<Http2RetryPolicy, RedirectLayer>>, fn(BoxError) -> BoxError>;

Expand Down
4 changes: 2 additions & 2 deletions src/client/http/future.rs
Original file line number Diff line number Diff line change
Expand Up @@ -10,12 +10,12 @@ use url::Url;

use super::{
Response,
aliases::{BoxedClientService, CoreResponseFuture, GenericClientService},
aliases::{BoxedClientService, GenericClientService},
};
use crate::{
Body, Error,
client::{body, layer::redirect::RequestUri},
core::body::Incoming,
core::{body::Incoming, client::future::ResponseFuture as CoreResponseFuture},
error::BoxError,
into_url::IntoUrlSealed,
};
Expand Down
43 changes: 43 additions & 0 deletions src/core/client/future.rs
Original file line number Diff line number Diff line change
@@ -0,0 +1,43 @@
use std::{
fmt,
future::Future,
pin::Pin,
task::{self, Poll},
};

use http::Response;

use crate::core::{body::Incoming, client::Error};

/// A `Future` that will resolve to an HTTP Response.
#[must_use = "futures do nothing unless polled"]
pub struct ResponseFuture {
inner: Pin<Box<dyn Future<Output = Result<Response<Incoming>, Error>> + Send>>,
Comment thread
0x676e67 marked this conversation as resolved.
}

impl ResponseFuture {
#[inline]
pub(super) fn new<F>(value: F) -> ResponseFuture
where
F: Future<Output = Result<Response<Incoming>, Error>> + Send + 'static,
{
ResponseFuture {
inner: Box::pin(value),
}
}
}

impl fmt::Debug for ResponseFuture {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
f.pad("Future<Response>")
}
}

impl Future for ResponseFuture {
Comment thread
0x676e67 marked this conversation as resolved.
type Output = Result<Response<Incoming>, Error>;

#[inline]
fn poll(mut self: Pin<&mut Self>, cx: &mut task::Context<'_>) -> Poll<Self::Output> {
self.inner.as_mut().poll(cx)
}
}
63 changes: 14 additions & 49 deletions src/core/client/mod.rs
Original file line number Diff line number Diff line change
Expand Up @@ -7,6 +7,7 @@ mod pool;
pub mod conn;
pub mod connect;
pub(super) mod dispatch;
pub mod future;
pub mod options;
pub mod proxy;

Expand All @@ -21,15 +22,15 @@ use std::{
time::Duration,
};

use futures_util::future::{self, Either, FutureExt, TryFutureExt};
use future::ResponseFuture;
use futures_util::future::{Either, FutureExt, TryFutureExt};
use http::{
HeaderValue, Method, Request, Response, Uri, Version,
header::HOST,
uri::{Authority, PathAndQuery, Scheme},
};
use http_body::Body;
use pool::Ver;
use sync_wrapper::SyncWrapper;
use tower::util::Oneshot;

use crate::{
Expand Down Expand Up @@ -225,15 +226,6 @@ enum TrySendError<B> {
Nope(Error),
}

type ResponseWrapper =
SyncWrapper<Pin<Box<dyn Future<Output = Result<Response<Incoming>, Error>> + Send>>>;

/// A `Future` that will resolve to an HTTP Response.
#[must_use = "futures do nothing unless polled"]
pub struct ResponseFuture {
inner: ResponseWrapper,
}

// ===== impl HttpClient =====

impl HttpClient<(), ()> {
Expand Down Expand Up @@ -263,17 +255,22 @@ where
match req.version() {
Version::HTTP_10 if is_http_connect => {
warn!("CONNECT is not allowed for HTTP/1.0");
return ResponseFuture::new(future::err(e!(UserUnsupportedRequestMethod)));
return ResponseFuture::new(futures_util::future::err(e!(
UserUnsupportedRequestMethod
)));
}
Version::HTTP_10 | Version::HTTP_11 | Version::HTTP_2 => {}
// completely unsupported HTTP version (like HTTP/0.9)!
unsupported => return ResponseFuture::error_version(unsupported),
_unsupported => {
warn!("Request has unsupported version \"{:?}\"", _unsupported);
return ResponseFuture::new(futures_util::future::err(e!(UserUnsupportedVersion)));
}
};

// Extract and normalize URI
let uri = match normalize_uri(&mut req, is_http_connect) {
Ok(uri) => uri,
Err(err) => return ResponseFuture::new(future::err(err)),
Err(err) => return ResponseFuture::new(futures_util::future::err(err)),
};

// Extract config extensions
Expand Down Expand Up @@ -490,7 +487,7 @@ where

// The order of the `select` is depended on below...

match future::select(checkout, connect).await {
match futures_util::future::select(checkout, connect).await {
// Checkout won, connect future may have been started or not.
//
// If it has, let it finish and insert back into the pool,
Expand Down Expand Up @@ -577,7 +574,7 @@ where
None => {
let canceled = e!(Canceled);
// HTTP/2 connection in progress.
return Either::Right(future::err(canceled));
return Either::Right(futures_util::future::err(canceled));
}
};
Either::Left(
Expand All @@ -598,7 +595,7 @@ where
// Another connection has already upgraded,
// the pool checkout should finish up for us.
let canceled = e!(Canceled, "ALPN upgraded to HTTP/2");
return Either::Right(future::err(canceled));
return Either::Right(futures_util::future::err(canceled));
}
}
} else {
Expand Down Expand Up @@ -796,38 +793,6 @@ impl<C, B> fmt::Debug for HttpClient<C, B> {
}
}

// ===== impl ResponseFuture =====

impl ResponseFuture {
fn new<F>(value: F) -> Self
where
F: Future<Output = Result<Response<Incoming>, Error>> + Send + 'static,
{
Self {
inner: SyncWrapper::new(Box::pin(value)),
}
}

fn error_version(_ver: Version) -> Self {
warn!("Request has unsupported version \"{:?}\"", _ver);
ResponseFuture::new(Box::pin(future::err(e!(UserUnsupportedVersion))))
}
}

impl fmt::Debug for ResponseFuture {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
f.pad("Future<Response>")
}
}

impl Future for ResponseFuture {
type Output = Result<Response<Incoming>, Error>;

fn poll(mut self: Pin<&mut Self>, cx: &mut task::Context<'_>) -> Poll<Self::Output> {
self.inner.get_mut().as_mut().poll(cx)
}
}

/// A pooled HTTP connection that can send requests
struct PoolClient<B> {
conn_info: Connected,
Expand Down
Loading