From b66e39df177699f1629334c24b46d5da67084487 Mon Sep 17 00:00:00 2001 From: 0x676e67 Date: Sat, 19 Jul 2025 12:55:20 +0800 Subject: [PATCH 1/2] build(feature): drop redundant `sync_wrapper` --- Cargo.toml | 12 +++++--- src/client/http/aliases.rs | 2 -- src/client/http/future.rs | 4 +-- src/core/client/future.rs | 42 +++++++++++++++++++++++++ src/core/client/mod.rs | 63 +++++++++----------------------------- 5 files changed, 65 insertions(+), 58 deletions(-) create mode 100644 src/core/client/future.rs diff --git a/Cargo.toml b/Cargo.toml index d45928c3f..eea433812 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -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"] @@ -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" @@ -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 } diff --git a/src/client/http/aliases.rs b/src/client/http/aliases.rs index 302844b06..9830c7f85 100644 --- a/src/client/http/aliases.rs +++ b/src/client/http/aliases.rs @@ -59,8 +59,6 @@ pub type RedirectLayer = FollowRedirect< FollowRedirectPolicy, >; -pub type CoreResponseFuture = crate::core::client::ResponseFuture; - pub type GenericClientService = MapErr>, fn(BoxError) -> BoxError>; diff --git a/src/client/http/future.rs b/src/client/http/future.rs index 2709d3fe6..05e0bbfb9 100644 --- a/src/client/http/future.rs +++ b/src/client/http/future.rs @@ -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, }; diff --git a/src/core/client/future.rs b/src/core/client/future.rs new file mode 100644 index 000000000..d92924c13 --- /dev/null +++ b/src/core/client/future.rs @@ -0,0 +1,42 @@ +use std::{ + fmt, + 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, Error>> + Send>>, +} + +impl ResponseFuture { + #[inline] + pub(super) fn new(value: F) -> ResponseFuture + where + F: Future, 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") + } +} + +impl Future for ResponseFuture { + type Output = Result, Error>; + + #[inline] + fn poll(mut self: Pin<&mut Self>, cx: &mut task::Context<'_>) -> Poll { + self.inner.as_mut().poll(cx) + } +} diff --git a/src/core/client/mod.rs b/src/core/client/mod.rs index 24d42cfed..9782ff6b5 100644 --- a/src/core/client/mod.rs +++ b/src/core/client/mod.rs @@ -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; @@ -21,7 +22,8 @@ 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, @@ -29,7 +31,6 @@ use http::{ }; use http_body::Body; use pool::Ver; -use sync_wrapper::SyncWrapper; use tower::util::Oneshot; use crate::{ @@ -225,15 +226,6 @@ enum TrySendError { Nope(Error), } -type ResponseWrapper = - SyncWrapper, 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<(), ()> { @@ -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 @@ -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, @@ -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( @@ -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 { @@ -796,38 +793,6 @@ impl fmt::Debug for HttpClient { } } -// ===== impl ResponseFuture ===== - -impl ResponseFuture { - fn new(value: F) -> Self - where - F: Future, 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") - } -} - -impl Future for ResponseFuture { - type Output = Result, Error>; - - fn poll(mut self: Pin<&mut Self>, cx: &mut task::Context<'_>) -> Poll { - self.inner.get_mut().as_mut().poll(cx) - } -} - /// A pooled HTTP connection that can send requests struct PoolClient { conn_info: Connected, From d85d8d90ba6127d008f08f52297d717b2edb3330 Mon Sep 17 00:00:00 2001 From: 0x676e67 Date: Sat, 19 Jul 2025 12:58:53 +0800 Subject: [PATCH 2/2] build(feature): drop redundant `sync_wrapper` --- src/core/client/future.rs | 1 + 1 file changed, 1 insertion(+) diff --git a/src/core/client/future.rs b/src/core/client/future.rs index d92924c13..3f0ce56b3 100644 --- a/src/core/client/future.rs +++ b/src/core/client/future.rs @@ -1,5 +1,6 @@ use std::{ fmt, + future::Future, pin::Pin, task::{self, Poll}, };