From 57c902203817a2f43abf4d8276bf377f77894662 Mon Sep 17 00:00:00 2001 From: Charles Sibbald Date: Tue, 27 Oct 2026 16:45:00 +0200 Subject: [PATCH] feat(client): Interceptor, a per-attempt hook with static dispatch `Client` and `RequestBuilder` are generic over an `Interceptor` with three hooks, all no-ops by default: - `before_send(&mut OutgoingRequest)`: once the attempt has a connection, before it is written. It can read the method, URL, attempt number, redirect hop and whether the connection is reused, and change that attempt's headers. It runs again on the pool's stale-connection retry and on each redirect hop, which is what a header that must be fresh per attempt needs. - `after_response(&ResponseHead, &Timing)`: at the response head, before the body; `Timing` gives the time to first byte and connection reuse. - `on_error(&io::Error, &Timing)`: a failed attempt, including one a retry then hides from the caller. `NoInterceptor` sets `Interceptor::ENABLED = false`, so the client skips building the hook arguments (and reading the clock): a client without an interceptor compiles to what it was. `headers_mut()` copies the request's headers only when called, so a read-only interceptor keeps the borrowed map. `ClientBuilder::build_with_interceptor` builds one; `Client::with_interceptor` gives the same pool, configuration and counters another interceptor. `BufferedResponse::reused_connection()` and `StreamingResponse::reused_connection()` say whether the response came over a pooled connection. bench_client gains `pool-hook`: the pool with empty hooks that do not opt out, to measure the plumbing against `pool`. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01Y3TSRmEZAYU6RBUXhkn7WF --- README.md | 31 ++++ examples/bench_client.rs | 23 ++- src/client/interceptor.rs | 186 ++++++++++++++++++++++++ src/client/mod.rs | 2 + src/client/rich.rs | 299 ++++++++++++++++++++++++++++++++++++-- 5 files changed, 522 insertions(+), 19 deletions(-) create mode 100644 src/client/interceptor.rs diff --git a/README.md b/README.md index 7552cd4..5a7ff50 100644 --- a/README.md +++ b/README.md @@ -97,6 +97,37 @@ cancellation interrupts may-aware waits and I/O, discards incomplete connections unsafe coroutine cancellation still performs pool cleanup but deliberately invokes no callbacks from `Drop`. +#### Per-attempt hooks + +`Client` is generic over an `Interceptor`, which defaults to `NoInterceptor` and then compiles +away. An interceptor's `before_send` runs on every attempt the client makes: the first try, the +pool's own retry on a fresh connection, and each redirect hop. It can add or change headers for +that attempt. `after_response` sees the response head with the attempt's timing (time to first +byte, whether the connection was reused), and `on_error` sees a failed attempt, including one a +retry then hid from the caller. Headers that are the same for every attempt are simpler to set on +the request builder. + +```rust,no_run +use may_minihttp::client::{Client, Interceptor, OutgoingRequest}; +use http::{HeaderName, HeaderValue}; + +#[derive(Clone)] +struct AttemptId; + +impl Interceptor for AttemptId { + fn before_send(&self, request: &mut OutgoingRequest<'_, '_>) { + let attempt = HeaderValue::from(request.attempt()); + request + .headers_mut() + .insert(HeaderName::from_static("x-attempt"), attempt); + } +} + +let client = Client::builder().build_with_interceptor(AttemptId)?; +// or, sharing an existing client's pool: existing.with_interceptor(AttemptId) +# Ok::<(), Box>(()) +``` + Multipart filesystem helpers are deliberately named `blocking_*` and must be called outside latency-sensitive may coroutines. Automatic stale-socket retry is limited to one attempt for idempotent requests with replayable bodies. diff --git a/examples/bench_client.rs b/examples/bench_client.rs index e78ce37..253e800 100644 --- a/examples/bench_client.rs +++ b/examples/bench_client.rs @@ -11,6 +11,9 @@ //! per call pays //! - `reuse` one `HttpClient` per worker, kept alive across its requests //! - `pool` one shared pooled [`Client`], `max_connections_per_origin = concurrency` +//! - `pool-hook` the same pool with an interceptor whose hooks are empty but which does not opt +//! out (`Interceptor::ENABLED` left true): the cost of building the hook arguments, against +//! `pool` (no interceptor) //! //! `header-bytes` adds an `X-Pad` header of that size (a bearer token with claims is 1-6 KiB); //! `body-bytes > 0` makes the request a POST with a body of that size (`/echo` on bench_server). @@ -23,7 +26,7 @@ use std::sync::{Arc, Mutex}; use std::time::{Duration, Instant}; use http::{HeaderName, HeaderValue, Method, Uri}; -use may_minihttp::client::{Client, HttpClient}; +use may_minihttp::client::{Client, HttpClient, Interceptor}; struct Counting; @@ -53,6 +56,7 @@ enum Mode { Fresh, Reuse, Pool, + PoolHook, } struct Job { @@ -90,7 +94,7 @@ impl Job { Ok(sink.len()) } - fn pooled(&self, client: &Client) -> io::Result { + fn pooled(&self, client: &Client) -> io::Result { let mut builder = client.request(self.method(), &self.url)?; if let Some(pad) = &self.pad { builder = builder.header(HeaderName::from_static("x-pad"), pad.clone()); @@ -106,6 +110,12 @@ impl Job { } } +/// Empty hooks, still called: measures the interceptor plumbing itself. +#[derive(Clone)] +struct EmptyHooks; + +impl Interceptor for EmptyHooks {} + fn arg(value: Option, default: T) -> T { value.and_then(|v| v.parse().ok()).unwrap_or(default) } @@ -116,9 +126,10 @@ fn main() -> io::Result<()> { Some("fresh") => Mode::Fresh, Some("reuse") => Mode::Reuse, Some("pool") => Mode::Pool, + Some("pool-hook") => Mode::PoolHook, _ => { eprintln!( - "usage: bench_client fresh|reuse|pool [c] [n] [header-bytes] [body-bytes]" + "usage: bench_client fresh|reuse|pool|pool-hook [c] [n] [header-bytes] [body-bytes]" ); std::process::exit(2); } @@ -152,6 +163,7 @@ fn main() -> io::Result<()> { .request_timeout(Duration::from_secs(30)) .build()?, ); + let hooked = Arc::new(pool.with_interceptor(EmptyHooks)); // warm-up: one request per worker's worth of connections, not measured { @@ -168,6 +180,7 @@ fn main() -> io::Result<()> { .map(|_| { let job = Arc::clone(&job); let pool = Arc::clone(&pool); + let hooked = Arc::clone(&hooked); let latencies = Arc::clone(&latencies); let errors = Arc::clone(&errors); may::go!(move || { @@ -189,6 +202,7 @@ fn main() -> io::Result<()> { } } Mode::Pool => job.pooled(&pool), + Mode::PoolHook => job.pooled(&hooked), }; match result { Ok(_) => local.push(t.elapsed()), @@ -224,6 +238,7 @@ fn main() -> io::Result<()> { Mode::Fresh => "fresh", Mode::Reuse => "reuse", Mode::Pool => "pool", + Mode::PoolHook => "pool-hook", }, concurrency, per_worker, @@ -237,7 +252,7 @@ fn main() -> io::Result<()> { pct(0.99), pct(1.0), allocs as f64 / (ok.max(1)) as f64, - if mode == Mode::Pool { + if matches!(mode, Mode::Pool | Mode::PoolHook) { format!(" pool={stats:?}") } else { String::new() diff --git a/src/client/interceptor.rs b/src/client/interceptor.rs new file mode 100644 index 0000000..f5534af --- /dev/null +++ b/src/client/interceptor.rs @@ -0,0 +1,186 @@ +//! Per-attempt hooks on the pooled [`Client`](super::Client). +//! +//! An [`Interceptor`] sees every attempt the client makes on the wire: the first try, a retry on a +//! fresh connection after a reused one turned out to be dead, and each redirect hop. It is the +//! extension point for anything that has to happen *per attempt* rather than once per logical +//! request, such as: +//! +//! - headers whose value must be fresh on every attempt (a signature over the request line and a +//! timestamp, a per-attempt id), including on the retry the pool makes by itself; +//! - per-attempt timing or connection-reuse accounting; +//! - recording failures that the caller never sees because a retry succeeded. +//! +//! Headers that are the same for every attempt (a static token, a trace context for the logical +//! request) are simpler to set once on the [`RequestBuilder`](super::RequestBuilder). +//! +//! The client is generic over its interceptor and defaults to [`NoInterceptor`], whose methods are +//! empty and inlined, so a client without an interceptor compiles to the same code as before. An +//! interceptor is attached with [`ClientBuilder::build_with_interceptor`](super::ClientBuilder::build_with_interceptor) +//! or [`Client::with_interceptor`](super::Client::with_interceptor); the latter shares the original +//! client's connection pool. +//! +//! Hooks run synchronously on the request's coroutine and never while the client holds its pool +//! lock. They cannot alter control flow. They should return quickly; a panic propagates to the +//! caller like any panic in the request path. + +use std::borrow::Cow; +use std::io; +use std::time::{Duration, Instant}; + +use http::{HeaderMap, Method, StatusCode, Version}; +use url::Url; + +/// Hooks called for every attempt the client makes. All methods default to doing nothing. +/// +/// Implementations are cloned into every [`RequestBuilder`](super::RequestBuilder), so keep them +/// cheap to clone (a unit struct, or an `Arc` around shared state). +pub trait Interceptor: Clone + Send + Sync + 'static { + /// False only for an interceptor whose hooks all do nothing ([`NoInterceptor`]): the client + /// then skips building the hook arguments, including reading the clock. + const ENABLED: bool = true; + + /// Called once the attempt has a connection and before the request is written to it. + /// + /// Headers added here are sent on this attempt only; a retry calls this again with the + /// request's original headers. + #[inline(always)] + fn before_send(&self, request: &mut OutgoingRequest<'_, '_>) { + let _ = request; + } + + /// Called when the response head has been read, before any of the body. + #[inline(always)] + fn after_response(&self, response: &ResponseHead<'_>, timing: &Timing) { + let _ = (response, timing); + } + + /// Called when an attempt fails after [`before_send`](Self::before_send), whether or not the + /// client then retries it. + #[inline(always)] + fn on_error(&self, error: &io::Error, timing: &Timing) { + let _ = (error, timing); + } +} + +/// The default interceptor: does nothing, costs nothing. +#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)] +pub struct NoInterceptor; + +impl Interceptor for NoInterceptor { + const ENABLED: bool = false; +} + +/// One attempt, as seen by [`Interceptor::before_send`]. +pub struct OutgoingRequest<'a, 'h> { + pub(super) method: &'a Method, + pub(super) url: &'a Url, + pub(super) headers: &'a mut Cow<'h, HeaderMap>, + pub(super) attempt: u32, + pub(super) redirect_hop: usize, + pub(super) stale_retry: bool, + pub(super) reused_connection: bool, +} + +impl OutgoingRequest<'_, '_> { + pub fn method(&self) -> &Method { + self.method + } + + /// The URL of this attempt (after any redirects). + pub fn url(&self) -> &Url { + self.url + } + + pub fn headers(&self) -> &HeaderMap { + self.headers + } + + /// Headers for this attempt. The first call copies the request's headers when they were + /// borrowed, so an interceptor that only reads pays nothing. + pub fn headers_mut(&mut self) -> &mut HeaderMap { + self.headers.to_mut() + } + + /// 1 for the first attempt; counts across stale-connection retries and redirect hops. + pub fn attempt(&self) -> u32 { + self.attempt + } + + /// 0 for the original URL, then 1, 2, ... for each redirect followed. + pub fn redirect_hop(&self) -> usize { + self.redirect_hop + } + + /// True when this attempt replaces one that failed on a reused, stale connection. + pub fn stale_retry(&self) -> bool { + self.stale_retry + } + + /// True when the attempt goes out on a pooled connection rather than a new one. + pub fn reused_connection(&self) -> bool { + self.reused_connection + } +} + +/// The response head, as seen by [`Interceptor::after_response`]. +#[derive(Debug, Clone, Copy)] +pub struct ResponseHead<'a> { + pub(super) status: StatusCode, + pub(super) version: Version, + pub(super) headers: &'a HeaderMap, + pub(super) url: &'a Url, +} + +impl ResponseHead<'_> { + pub fn status(&self) -> StatusCode { + self.status + } + + pub fn version(&self) -> Version { + self.version + } + + pub fn headers(&self) -> &HeaderMap { + self.headers + } + + pub fn url(&self) -> &Url { + self.url + } +} + +/// When things happened during one attempt. +#[derive(Debug, Clone, Copy)] +pub struct Timing { + pub(super) request_started: Instant, + pub(super) attempt_started: Instant, + pub(super) attempt: u32, + pub(super) reused_connection: bool, +} + +impl Timing { + /// When the logical request started (the `send` call). + pub fn request_started(&self) -> Instant { + self.request_started + } + + /// When this attempt had its connection and started writing. + pub fn attempt_started(&self) -> Instant { + self.attempt_started + } + + /// Time from the start of this attempt to the moment of the call: in `after_response`, the + /// time to the response head (time to first byte); in `on_error`, the time to the failure. + /// Read it inside the hook; a stored `Timing` keeps counting. + pub fn attempt_elapsed(&self) -> Duration { + self.attempt_started.elapsed() + } + + pub fn attempt(&self) -> u32 { + self.attempt + } + + pub fn reused_connection(&self) -> bool { + self.reused_connection + } +} diff --git a/src/client/mod.rs b/src/client/mod.rs index a1a0f0f..2d92f5c 100644 --- a/src/client/mod.rs +++ b/src/client/mod.rs @@ -7,6 +7,7 @@ mod body; mod buffer; mod cancellation; mod client_impl; +mod interceptor; mod metadata; mod multipart; mod observer; @@ -19,6 +20,7 @@ mod tls; pub use cancellation::CancellationToken; pub use client_impl::HttpClient; +pub use interceptor::{Interceptor, NoInterceptor, OutgoingRequest, ResponseHead, Timing}; pub use metadata::{RequestMetadata, RequestMetadataContext, RequestMetadataProvider}; pub use multipart::MultipartForm; pub use observer::{ClientEvent, ClientObserver, ObservedOrigin}; diff --git a/src/client/rich.rs b/src/client/rich.rs index 4a08f57..1797116 100644 --- a/src/client/rich.rs +++ b/src/client/rich.rs @@ -18,6 +18,7 @@ use rustls::ClientConfig; use url::Url; use super::cancellation::{cancelled_error, is_cancelled_error, run_cancellable}; +use super::interceptor::{Interceptor, NoInterceptor, OutgoingRequest, ResponseHead, Timing}; use super::{ CancellationToken, ClientEvent, ClientObserver, ObservedOrigin, RequestMetadata, RequestMetadataContext, RequestMetadataProvider, Resolver, SystemResolver, @@ -246,6 +247,11 @@ impl ClientBuilder { } pub fn build(self) -> io::Result { + self.build_with_interceptor(NoInterceptor) + } + + /// Build a client that calls `interceptor` on every attempt (see [`Interceptor`]). + pub fn build_with_interceptor(self, interceptor: I) -> io::Result> { if self.max_connections == 0 || self.max_connections_per_origin == 0 { return Err(io::Error::new( io::ErrorKind::InvalidInput, @@ -339,13 +345,17 @@ impl ClientBuilder { stats: ClientStatsInner::default(), next_request_id: AtomicU64::new(1), }), + interceptor, }) } } +/// Pooled HTTP/1.1 client. `I` is the per-attempt [`Interceptor`]; the default does nothing and +/// compiles away. #[derive(Clone)] -pub struct Client { +pub struct Client { inner: Arc, + interceptor: I, } /// Monotonic operational counters for a [`Client`]. @@ -469,7 +479,7 @@ impl ClientStatsInner { } } -impl fmt::Debug for Client { +impl fmt::Debug for Client { fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { formatter .debug_struct("Client") @@ -498,6 +508,16 @@ struct StreamingExecution<'a> { tls_snapshot: &'a mut Option, } +struct AttemptInfo<'a> { + method: &'a Method, + url: &'a Url, + trace: &'a RequestTrace, + attempt: u32, + redirect_hop: usize, + stale_retry: bool, + reused_connection: bool, +} + struct AttemptExecution<'a> { deadline: Instant, trace: &'a RequestTrace, @@ -516,8 +536,23 @@ impl Client { pub fn new() -> io::Result { Self::builder().build() } +} - pub fn request(&self, method: Method, url: &str) -> io::Result { +impl Client { + /// The same client (same connection pool, configuration and counters) with another + /// interceptor. + pub fn with_interceptor(&self, interceptor: J) -> Client { + Client { + inner: Arc::clone(&self.inner), + interceptor, + } + } + + pub fn interceptor(&self) -> &I { + &self.interceptor + } + + pub fn request(&self, method: Method, url: &str) -> io::Result> { let url = parse_url(url)?; Ok(RequestBuilder { client: self.clone(), @@ -530,11 +565,11 @@ impl Client { }) } - pub fn get(&self, url: &str) -> io::Result { + pub fn get(&self, url: &str) -> io::Result> { self.request(Method::GET, url) } - pub fn post(&self, url: &str) -> io::Result { + pub fn post(&self, url: &str) -> io::Result> { self.request(Method::POST, url) } @@ -543,7 +578,7 @@ impl Client { self.inner.stats.snapshot() } - fn execute(&self, request: RequestBuilder) -> io::Result { + fn execute(&self, request: RequestBuilder) -> io::Result { let trace = self.begin_request(&request.method, &request.url); let cancellation = request.cancellation.clone(); let result = match cancellation { @@ -563,7 +598,7 @@ impl Client { fn execute_buffered( &self, - request: RequestBuilder, + request: RequestBuilder, trace: &RequestTrace, ) -> io::Result { let deadline = Instant::now() @@ -691,7 +726,7 @@ impl Client { } } - fn execute_streaming(&self, request: RequestBuilder) -> io::Result { + fn execute_streaming(&self, request: RequestBuilder) -> io::Result { let trace = self.begin_request(&request.method, &request.url); let cancellation = request.cancellation.clone(); let result = match cancellation { @@ -706,7 +741,7 @@ impl Client { fn execute_streaming_inner( &self, - request: RequestBuilder, + request: RequestBuilder, trace: RequestTrace, ) -> io::Result { if self.inner.config.redirect_policy != RedirectPolicy::None { @@ -772,7 +807,7 @@ impl Client { "request attempt counter overflow", ) })?; - let attempt_headers = self.prepare_attempt_headers( + let mut attempt_headers = self.prepare_attempt_headers( headers, RequestMetadataContext { request_id: trace.request_id, @@ -787,11 +822,36 @@ impl Client { )?; let mut lease = self.checkout(&key, Arc::clone(&tls_config), deadline, trace)?; let reused_idle_connection = lease.reused_idle_connection; + let timing = self.before_send( + &mut attempt_headers, + AttemptInfo { + method, + url, + trace, + attempt: *attempt, + redirect_hop, + stale_retry, + reused_connection: reused_idle_connection, + }, + ); let result = (|| { let mut response = self.send_on_lease(&mut lease, method, url, &attempt_headers, body, deadline)?; let status = response.status(); + if I::ENABLED { + if let Some(timing) = &timing { + self.interceptor.after_response( + &ResponseHead { + status, + version: response.version(), + headers: response.headers(), + url, + }, + timing, + ); + } + } self.inner.observe(ClientEvent::ResponseHeaders { request_id: trace.request_id, origin: observed_url(url), @@ -852,10 +912,16 @@ impl Client { headers: response_headers, body: bytes, final_url: url.clone(), + reused_connection: reused_idle_connection, }, reusable, )) })(); + if I::ENABLED { + if let (Err(error), Some(timing)) = (&result, &timing) { + self.interceptor.on_error(error, timing); + } + } match result { Ok((response, true)) => { @@ -920,7 +986,7 @@ impl Client { "request attempt counter overflow", ) })?; - let attempt_headers = self.prepare_attempt_headers( + let mut attempt_headers = self.prepare_attempt_headers( headers, RequestMetadataContext { request_id: trace.request_id, @@ -935,7 +1001,37 @@ impl Client { )?; let mut lease = self.checkout(&key, Arc::clone(&tls_config), deadline, trace)?; let reused_idle_connection = lease.reused_idle_connection; - match self.send_on_lease(&mut lease, method, url, &attempt_headers, body, deadline) { + let timing = self.before_send( + &mut attempt_headers, + AttemptInfo { + method, + url, + trace, + attempt: *attempt, + redirect_hop: 0, + stale_retry, + reused_connection: reused_idle_connection, + }, + ); + let sent = + self.send_on_lease(&mut lease, method, url, &attempt_headers, body, deadline); + if I::ENABLED { + if let Some(timing) = &timing { + match &sent { + Ok(response) => self.interceptor.after_response( + &ResponseHead { + status: response.status(), + version: response.version(), + headers: response.headers(), + url, + }, + timing, + ), + Err(error) => self.interceptor.on_error(error, timing), + } + } + } + match sent { Ok(response) => { let status = response.status(); self.inner.observe(ClientEvent::ResponseHeaders { @@ -968,6 +1064,7 @@ impl Client { trace: *trace, cancellation, terminal_observed: false, + reused_connection: reused_idle_connection, }; if streaming .response @@ -1004,6 +1101,35 @@ impl Client { } } + /// Run the interceptor's `before_send` for one attempt; `None` (and no clock read) when the + /// interceptor is [`NoInterceptor`]. + #[inline(always)] + fn before_send<'h>( + &self, + headers: &mut Cow<'h, HeaderMap>, + info: AttemptInfo<'_>, + ) -> Option { + if !I::ENABLED { + return None; + } + let timing = Timing { + request_started: info.trace.started, + attempt_started: Instant::now(), + attempt: info.attempt, + reused_connection: info.reused_connection, + }; + self.interceptor.before_send(&mut OutgoingRequest { + method: info.method, + url: info.url, + headers, + attempt: info.attempt, + redirect_hop: info.redirect_hop, + stale_retry: info.stale_retry, + reused_connection: info.reused_connection, + }); + Some(timing) + } + fn tls_for_url( &self, url: &Url, @@ -1430,8 +1556,8 @@ impl Client { } /// Request builder whose body explicitly records whether it can be replayed. -pub struct RequestBuilder { - client: Client, +pub struct RequestBuilder { + client: Client, method: Method, url: Url, headers: HeaderMap, @@ -1440,7 +1566,7 @@ pub struct RequestBuilder { cancellation: Option, } -impl RequestBuilder { +impl RequestBuilder { pub fn header(mut self, name: HeaderName, value: HeaderValue) -> Self { self.headers.append(name, value); self @@ -1561,9 +1687,15 @@ pub struct BufferedResponse { headers: HeaderMap, body: Vec, final_url: Url, + reused_connection: bool, } impl BufferedResponse { + /// True when the response came over a pooled connection that an earlier request had used. + pub fn reused_connection(&self) -> bool { + self.reused_connection + } + pub fn status(&self) -> StatusCode { self.status } @@ -1617,6 +1749,7 @@ pub struct StreamingResponse { trace: RequestTrace, cancellation: Option, terminal_observed: bool, + reused_connection: bool, } impl fmt::Debug for StreamingResponse { @@ -1633,6 +1766,11 @@ impl fmt::Debug for StreamingResponse { } impl StreamingResponse { + /// True when the response came over a pooled connection that an earlier request had used. + pub fn reused_connection(&self) -> bool { + self.reused_connection + } + pub fn status(&self) -> StatusCode { self.status } @@ -3952,6 +4090,137 @@ mod tests { server.join().unwrap(); } + #[derive(Clone, Default)] + struct AttemptHeader { + events: Arc>>, + } + + impl Interceptor for AttemptHeader { + fn before_send(&self, request: &mut OutgoingRequest<'_, '_>) { + let attempt = request.attempt(); + self.events.lock().unwrap().push(format!( + "before:{}:{}:{}", + request.url().path(), + attempt, + request.reused_connection() + )); + request.headers_mut().insert( + HeaderName::from_static("x-attempt"), + HeaderValue::from(attempt), + ); + } + + fn after_response(&self, response: &ResponseHead<'_>, timing: &Timing) { + self.events.lock().unwrap().push(format!( + "after:{}:{}:{}", + response.status().as_u16(), + timing.attempt(), + timing.reused_connection() + )); + } + + fn on_error(&self, _error: &io::Error, timing: &Timing) { + self.events + .lock() + .unwrap() + .push(format!("error:{}", timing.attempt())); + } + } + + #[test] + fn interceptor_runs_on_every_attempt_including_the_stale_retry() { + let listener = TcpListener::bind("127.0.0.1:0").unwrap(); + let port = listener.local_addr().unwrap().port(); + let server = thread::spawn(move || { + let (mut stale, _) = listener.accept().unwrap(); + let first = read_head(&mut stale).to_ascii_lowercase(); + assert!(first.contains("x-attempt: 1\r\n"), "{first}"); + stale + .write_all(b"HTTP/1.1 200 OK\r\nContent-Length: 2\r\n\r\nok") + .unwrap(); + stale.flush().unwrap(); + let _ = read_head(&mut stale); + drop(stale); + + let (mut replacement, _) = listener.accept().unwrap(); + let retried = read_head(&mut replacement).to_ascii_lowercase(); + // The header set in before_send is per attempt: the retry carries its own value. + assert!(retried.contains("x-attempt: 2\r\n"), "{retried}"); + assert_eq!(retried.matches("x-attempt").count(), 1, "{retried}"); + replacement + .write_all( + b"HTTP/1.1 201 Created\r\nContent-Length: 5\r\nConnection: close\r\n\r\nfresh", + ) + .unwrap(); + }); + + let interceptor = AttemptHeader::default(); + let client = Client::builder() + .redirect_policy(RedirectPolicy::None) + .request_timeout(Duration::from_secs(2)) + .build_with_interceptor(interceptor.clone()) + .unwrap(); + let first = client + .get(&format!("http://127.0.0.1:{port}/first")) + .unwrap() + .send() + .unwrap(); + assert!(!first.reused_connection()); + let second = client + .get(&format!("http://127.0.0.1:{port}/second")) + .unwrap() + .send() + .unwrap(); + assert_eq!(second.body(), b"fresh"); + assert!(!second.reused_connection()); + server.join().unwrap(); + + let events = interceptor.events.lock().unwrap().clone(); + assert_eq!( + events, + [ + "before:/first:1:false", + "after:200:1:false", + "before:/second:1:true", + "error:1", + "before:/second:2:false", + "after:201:2:false", + ] + ); + } + + #[test] + fn with_interceptor_shares_the_pool_and_reports_reuse() { + let listener = TcpListener::bind("127.0.0.1:0").unwrap(); + let port = listener.local_addr().unwrap().port(); + let server = thread::spawn(move || { + let (mut stream, _) = listener.accept().unwrap(); + for _ in 0..2 { + let _ = read_head(&mut stream); + stream + .write_all(b"HTTP/1.1 200 OK\r\nContent-Length: 2\r\n\r\nok") + .unwrap(); + stream.flush().unwrap(); + } + }); + + let plain = test_client(RedirectPolicy::None); + let interceptor = AttemptHeader::default(); + let intercepted = plain.with_interceptor(interceptor.clone()); + let url = format!("http://127.0.0.1:{port}/"); + assert!(!plain.get(&url).unwrap().send().unwrap().reused_connection()); + let streamed = intercepted.get(&url).unwrap().send_streaming().unwrap(); + assert!(streamed.reused_connection()); + drop(streamed); + assert_eq!(plain.stats().connections_created, 1); + assert_eq!(plain.stats().connections_reused, 1); + assert_eq!( + interceptor.events.lock().unwrap().clone(), + ["before:/:1:true", "after:200:1:true"] + ); + server.join().unwrap(); + } + #[test] fn stale_connection_retry_refreshes_attempt_metadata() { let listener = TcpListener::bind("127.0.0.1:0").unwrap();