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