1use std::pin::Pin;
2use std::sync::{Arc, Mutex, MutexGuard, PoisonError};
3use std::task::{Context, Poll};
4
5use bytes::Bytes;
6use futures_util::Stream;
7
8use super::RequestBuilder;
9use super::route::Route;
10use crate::core::Result;
11use crate::core::body::{Body, BodyKind, BoxedStream};
12use crate::core::deadline::Deadline;
13use crate::core::error::{Error, Kind};
14use crate::core::response::Response;
15use crate::core::session::execute::Attempt;
16use crate::core::transport::ResponseMode;
17use crate::util::is_idempotent;
18
19impl RequestBuilder {
20 pub async fn send(self) -> Result<Response> {
21 Box::pin(self.run()).await
22 }
23
24 async fn run(mut self) -> Result<Response> {
25 self.prepare()?;
26 let retry_policy = self.retry_policy.clone();
27 let session = self.session.clone();
28 let deadline = session.deadline(self.timeouts.as_ref());
29 let tag = self.tag.take();
30 let status_errors = self.status_errors;
31 let attempt = self.into_attempt(deadline);
32 let method = attempt.method.clone();
33 let url = attempt.url.clone();
34 let streamed = attempt.response == ResponseMode::Streamed;
35 let mut exhausted = false;
36 let response = session
37 .traced(
38 method.as_str(),
39 &url,
40 streamed,
41 run_attempts(
42 &session,
43 attempt,
44 &retry_policy,
45 deadline,
46 tag,
47 &mut exhausted,
48 ),
49 )
50 .await?;
51 if !status_errors {
52 return Ok(response);
53 }
54 let policy_wait = retry_policy.server_wait(&response);
55 response
56 .error_for_status_with_body(&deadline)
57 .await
58 .map_err(|mut error| {
59 error.set_retry_outcome(policy_wait, exhausted);
60 error
61 })
62 }
63
64 fn into_attempt(mut self, deadline: Deadline) -> Attempt {
65 let headers = if self.headers.is_empty() {
66 None
67 } else {
68 Some(std::mem::take(&mut self.headers))
69 };
70 Attempt {
71 method: std::mem::take(&mut self.method),
72 url: std::mem::take(&mut self.url),
73 preset: self.preset,
74 body: std::mem::take(&mut self.body),
75 headers,
76 deadline,
77 response: ResponseMode::new(self.stream_response, self.status_errors),
78 proxy: self.proxy.take(),
79 header_order: self.header_order.take(),
80 redirect: self.redirect.take(),
81 digest: self.digest_auth.take(),
82 initiator: self.initiator.take(),
83 trusted_origin: self.trusted_origin.take(),
84 }
85 }
86
87 pub(crate) fn prepare(&mut self) -> Result<()> {
88 if let Some(err) = self.builder_error.take() {
89 return Err(err);
90 }
91 self.infer_from_content_type();
92
93 if !self.query_params.is_empty() {
94 let mut url = url::Url::parse(&self.url).map_err(crate::core::Error::from_url_parse)?;
95 {
96 let mut pairs = url.query_pairs_mut();
97 for (k, v) in &self.query_params {
98 pairs.append_pair(k, v);
99 }
100 }
101 self.url = url.to_string();
102 }
103
104 if let Some(encoding) = self.compress {
105 match &self.body.0 {
106 BodyKind::Bytes(b) => {
107 let compressed = encoding.encode(b)?;
108 self.headers
109 .set("content-encoding", encoding.header_value())?;
110 self.body = Body::from(compressed);
111 }
112 BodyKind::Empty => {}
113 BodyKind::Stream { .. } => {
114 return Err(Error::new(Kind::Body).with_message(
115 "request-body compression is not supported for streaming bodies; \
116 buffer the body into bytes before calling `.compress(..)`",
117 ));
118 }
119 }
120 }
121 Ok(())
122 }
123}
124
125async fn run_attempts(
126 session: &crate::core::session::Session,
127 mut attempt: Attempt,
128 retry_policy: &crate::core::retry::RetryPolicy,
129 deadline: Deadline,
130 tag: Option<String>,
131 exhausted: &mut bool,
132) -> (Result<Response>, u32) {
133 crate::trace::note_tag(tag);
134 let mut route = Route::new(session, &attempt, retry_policy);
135 if retry_policy.is_none() {
136 return (counted(route.send(attempt, None).await, 1), 1);
137 }
138 let retryable_method =
139 retry_policy.allow_non_idempotent || is_idempotent(attempt.method.as_str());
140 let replay = attempt.body.replay();
141 let mut rewind = Rewind::wrap(&mut attempt.body);
142 let base_headers = attempt.headers.clone();
143
144 let mut n: u32 = 0;
145 loop {
146 if let Err(err) = deadline.check() {
147 return (Err(err), n);
148 }
149 let attempt_body = std::mem::take(&mut attempt.body);
150 let this = attempt.again(attempt_body, base_headers.clone());
151 let result = route.send(this, retry_policy.proxy_for_retry(n)).await;
152
153 let body_retryable = || replay.is_some() || rewind.as_mut().is_some_and(Rewind::reclaim);
154 let sleep = match plan_retry(&result, retry_policy, n, retryable_method, body_retryable) {
155 RetryPlan::Backoff(sleep) if sleep < deadline.remaining() => sleep,
156 plan => {
157 *exhausted = matches!(
158 plan,
159 RetryPlan::Exhausted | RetryPlan::Declined | RetryPlan::Backoff(_)
160 );
161 return (counted(result, n + 1), n + 1);
162 }
163 };
164 drop(result);
165
166 let slept = session
167 .unless_shut_down(async {
168 deadline.sleep(sleep).await;
169 Ok(())
170 })
171 .await;
172 if let Err(err) = slept {
173 return (counted(Err(err), n + 1), n + 1);
174 }
175 n += 1;
176 attempt.body = replay
177 .as_ref()
178 .and_then(Body::replay)
179 .or_else(|| rewind.as_mut().and_then(Rewind::again))
180 .unwrap_or_default();
181 }
182}
183
184fn counted(result: Result<Response>, attempts: u32) -> Result<Response> {
185 result
186 .map(|mut response| {
187 response.set_attempts(attempts);
188 response
189 })
190 .map_err(|mut error| {
191 error.set_attempts(attempts);
192 error
193 })
194}
195
196enum RetryPlan {
197 Stop,
198 Exhausted,
199 Declined,
200 Backoff(std::time::Duration),
201}
202
203fn plan_retry(
204 result: &Result<Response>,
205 retry_policy: &crate::core::retry::RetryPolicy,
206 attempt: u32,
207 retryable_method: bool,
208 body_retryable: impl FnOnce() -> bool,
209) -> RetryPlan {
210 if retry_policy.is_none() {
211 return RetryPlan::Stop;
212 }
213 if attempt >= retry_policy.max_retries {
214 return match result {
215 Ok(response) if retry_policy.matches_response(response) => RetryPlan::Exhausted,
216 _ => RetryPlan::Stop,
217 };
218 }
219 if !retry_wanted(result, retry_policy)
220 || !(retryable_method || retry_policy.retries_unsent(result))
221 || !body_retryable()
222 {
223 return RetryPlan::Stop;
224 }
225 match result
226 .as_ref()
227 .ok()
228 .and_then(|r| retry_policy.server_wait(r))
229 {
230 Some(wait) if wait > retry_policy.max_retry_after => RetryPlan::Declined,
231 Some(wait) => RetryPlan::Backoff(wait),
232 None => RetryPlan::Backoff(retry_policy.backoff(attempt)),
233 }
234}
235
236fn retry_wanted(result: &Result<Response>, retry_policy: &crate::core::retry::RetryPolicy) -> bool {
237 match result {
238 Ok(resp) => retry_policy.matches_response(resp),
239 Err(err) if err.is_timeout() => retry_policy.matches_timeout(),
240 Err(err) if err.is_retryable() => retry_policy.matches_connection_error(),
241 Err(_) => false,
242 }
243}
244
245struct Slot {
246 stream: Option<BoxedStream>,
247 polled: bool,
248}
249
250type SharedSlot = Arc<Mutex<Slot>>;
251
252fn lock(slot: &SharedSlot) -> MutexGuard<'_, Slot> {
253 slot.lock().unwrap_or_else(PoisonError::into_inner)
254}
255
256struct Rewind {
257 slot: SharedSlot,
258 length_hint: Option<u64>,
259 held: Option<BoxedStream>,
260}
261
262impl Rewind {
263 fn wrap(body: &mut Body) -> Option<Self> {
264 if !matches!(body.0, BodyKind::Stream { .. }) {
265 return None;
266 }
267 let BodyKind::Stream {
268 stream,
269 length_hint,
270 } = std::mem::take(&mut body.0)
271 else {
272 return None;
273 };
274 let rewind = Self {
275 slot: Arc::new(Mutex::new(Slot {
276 stream: Some(stream),
277 polled: false,
278 })),
279 length_hint,
280 held: None,
281 };
282 *body = rewind.body();
283 Some(rewind)
284 }
285
286 fn body(&self) -> Body {
287 Body(BodyKind::Stream {
288 stream: Box::pin(Reclaim(Arc::clone(&self.slot))),
289 length_hint: self.length_hint,
290 })
291 }
292
293 fn reclaim(&mut self) -> bool {
294 let mut slot = lock(&self.slot);
295 if slot.polled {
296 return false;
297 }
298 self.held = slot.stream.take();
299 self.held.is_some()
300 }
301
302 fn again(&mut self) -> Option<Body> {
303 let stream = self.held.take()?;
304 self.slot = Arc::new(Mutex::new(Slot {
305 stream: Some(stream),
306 polled: false,
307 }));
308 Some(self.body())
309 }
310}
311
312struct Reclaim(SharedSlot);
313
314impl Stream for Reclaim {
315 type Item = std::io::Result<Bytes>;
316
317 fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
318 let mut slot = lock(&self.0);
319 slot.polled = true;
320 match slot.stream.as_mut() {
321 Some(stream) => stream.as_mut().poll_next(cx),
322 None => Poll::Ready(None),
323 }
324 }
325}
326
327impl std::future::IntoFuture for RequestBuilder {
328 type Output = Result<Response>;
329 type IntoFuture = std::pin::Pin<Box<dyn std::future::Future<Output = Self::Output> + Send>>;
330
331 fn into_future(self) -> Self::IntoFuture {
332 Box::pin(async move { self.send().await })
333 }
334}
335
336#[cfg(test)]
337#[path = "send_tests.rs"]
338mod tests;