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();