leyline/core/request/
send.rs

1use super::RequestBuilder;
2use crate::core::Result;
3use crate::core::body::{Body, BodyKind};
4use crate::core::deadline::Deadline;
5use crate::core::error::{Error, Kind};
6use crate::core::response::Response;
7use crate::core::session::execute::Attempt;
8use crate::util::is_idempotent;
9
10impl RequestBuilder {
11    pub async fn send(mut self) -> Result<Response> {
12        self.prepare()?;
13        let retry_policy = self.retry_policy.clone();
14        let session = self.session.clone();
15        let deadline = session.deadline(self.timeouts.as_ref());
16        let mut attempt = self.into_attempt(deadline);
17
18        if retry_policy.is_none() {
19            return session.attempt(attempt).await;
20        }
21
22        let retryable_method =
23            retry_policy.allow_non_idempotent || is_idempotent(attempt.method.as_str());
24        let replay = attempt.body.replay();
25        let body_retryable = replay.is_some();
26        let base_headers = attempt.headers.clone();
27
28        let mut n: u32 = 0;
29        loop {
30            deadline.check()?;
31            let attempt_body = std::mem::take(&mut attempt.body);
32            let this = attempt.again(attempt_body, base_headers.clone());
33            let result = session.attempt(this).await;
34
35            let sleep =
36                match plan_retry(&result, &retry_policy, n, retryable_method, body_retryable) {
37                    RetryPlan::Backoff(sleep) if sleep < deadline.remaining() => sleep,
38                    _ => return result,
39                };
40
41            deadline.sleep(sleep).await;
42            n += 1;
43            attempt.body = replay.as_ref().and_then(Body::replay).unwrap_or_default();
44        }
45    }
46
47    fn into_attempt(mut self, deadline: Deadline) -> Attempt {
48        let headers = if self.headers.is_empty() {
49            None
50        } else {
51            Some(std::mem::take(&mut self.headers))
52        };
53        Attempt {
54            method: std::mem::take(&mut self.method),
55            url: std::mem::take(&mut self.url),
56            preset: self.preset,
57            body: std::mem::take(&mut self.body),
58            headers,
59            deadline,
60            stream_response: self.stream_response,
61            proxy: self.proxy.take(),
62            header_order: self.header_order.take(),
63            redirect: self.redirect.take(),
64            digest: self.digest_auth.take(),
65        }
66    }
67
68    pub(crate) fn prepare(&mut self) -> Result<()> {
69        if let Some(err) = self.builder_error.take() {
70            return Err(err);
71        }
72        self.infer_from_content_type();
73
74        if !self.query_params.is_empty() {
75            let mut url = url::Url::parse(&self.url).map_err(crate::core::Error::from_url_parse)?;
76            {
77                let mut pairs = url.query_pairs_mut();
78                for (k, v) in &self.query_params {
79                    pairs.append_pair(k, v);
80                }
81            }
82            self.url = url.to_string();
83        }
84
85        if let Some(encoding) = self.compress {
86            match &self.body.0 {
87                BodyKind::Bytes(b) => {
88                    let compressed = encoding.encode(b)?;
89                    self.headers
90                        .set("content-encoding", encoding.header_value())?;
91                    self.body = Body::from(compressed);
92                }
93                BodyKind::Empty => {}
94                BodyKind::Stream { .. } => {
95                    return Err(Error::new(Kind::Body).with_message(
96                        "request-body compression is not supported for streaming bodies; \
97                         buffer the body into bytes before calling `.compress(..)`",
98                    ));
99                }
100            }
101        }
102        Ok(())
103    }
104}
105
106enum RetryPlan {
107    Stop,
108    Backoff(std::time::Duration),
109}
110
111fn plan_retry(
112    result: &Result<Response>,
113    retry_policy: &crate::core::retry::RetryPolicy,
114    attempt: u32,
115    retryable_method: bool,
116    body_retryable: bool,
117) -> RetryPlan {
118    if retry_policy.is_none() || attempt >= retry_policy.max_retries {
119        return RetryPlan::Stop;
120    }
121    let should_retry = match result {
122        Ok(resp) => retry_policy.matches_status(resp.status().as_u16()),
123        Err(err) if err.is_timeout() => retry_policy.matches_timeout(),
124        Err(err) if err.is_retryable() => retry_policy.matches_connection_error(),
125        Err(_) => false,
126    };
127    if !should_retry || !retryable_method || !body_retryable {
128        return RetryPlan::Stop;
129    }
130    let retry_after = match result {
131        Ok(resp) => resp
132            .header("retry-after")
133            .and_then(crate::core::retry::parse_retry_after),
134        _ => None,
135    };
136    match retry_after {
137        Some(wait) if wait > retry_policy.max_retry_after => RetryPlan::Stop,
138        Some(wait) => RetryPlan::Backoff(wait),
139        None => RetryPlan::Backoff(retry_policy.delay(attempt)),
140    }
141}
142
143impl std::future::IntoFuture for RequestBuilder {
144    type Output = Result<Response>;
145    type IntoFuture = std::pin::Pin<Box<dyn std::future::Future<Output = Self::Output> + Send>>;
146
147    fn into_future(self) -> Self::IntoFuture {
148        Box::pin(async move { self.send().await })
149    }
150}
151
152#[cfg(test)]
153#[path = "send_tests.rs"]
154mod tests;
leyline/core/request/send.rs

leyline/core/request/
send.rs
