s3/request/
tokio_backend.rs

1extern crate base64;
2extern crate md5;
3
4use bytes::Bytes;
5use futures::TryStreamExt;
6use http_body_util::BodyExt;
7use http_body_util::BodyStream;
8use http_body_util::Full;
9use hyper::body::Incoming;
10use std::collections::HashMap;
11use time::OffsetDateTime;
12use tokio::io;
13
14use super::request_trait::{Request, ResponseData};
15use crate::bucket::Bucket;
16use crate::command::Command;
17use crate::command::HttpMethod;
18use crate::error::S3Error;
19
20pub use crate::request::tokio_backend::HyperRequest as RequestImpl;
21pub use tokio::io::AsyncWriteExt;
22pub use tokio_stream::Stream;
23
24#[cfg(feature = "tracing")]
25use tracing::{event, span, Level};
26
27use crate::request::request_trait::ResponseDataStream;
28
29// Temporary structure for making a request
30pub struct HyperRequest<'a> {
31    pub bucket: &'a Bucket,
32    pub path: &'a str,
33    pub command: Command<'a>,
34    pub datetime: OffsetDateTime,
35}
36
37impl<'a> Request for HyperRequest<'a> {
38    type Response = http::Response<Incoming>;
39    type HeaderMap = http::header::HeaderMap;
40
41    async fn response(&self) -> Result<http::Response<Incoming>, S3Error> {
42        // Build headers
43        let headers = match self.headers() {
44            Ok(headers) => headers,
45            Err(e) => return Err(e),
46        };
47        let client = self.bucket.http_client();
48
49        let method = match self.command.http_verb() {
50            HttpMethod::Delete => http::Method::DELETE,
51            HttpMethod::Get => http::Method::GET,
52            HttpMethod::Post => http::Method::POST,
53            HttpMethod::Put => http::Method::PUT,
54            HttpMethod::Head => http::Method::HEAD,
55        };
56
57        let request: http::Request<Full<Bytes>> = {
58            let mut request = http::Request::builder()
59                .method(method)
60                .uri(self.url()?.as_str());
61
62            for (header, value) in headers.iter() {
63                request = request.header(header, value);
64            }
65            request.body(Full::from(Bytes::from(self.request_body())))?
66        };
67        #[cfg(feature = "tracing")]
68        let span = span!(
69            Level::DEBUG,
70            "rust-s3-async",
71            bucket = self.bucket.name(),
72            command = self.command.to_string(),
73            path = self.path,
74            second = self.datetime.second(),
75            minute = self.datetime.minute(),
76            hour = self.datetime.hour(),
77            day = self.datetime.day(),
78            month = self.datetime.month() as u8,
79            year = self.datetime.year()
80        );
81        #[cfg(feature = "tracing")]
82        let _enter = span.enter();
83        let response = client.request(request).await?;
84
85        #[cfg(feature = "tracing")]
86        event!(Level::DEBUG, status_code = response.status().as_u16(),);
87
88        if cfg!(feature = "fail-on-err") && !response.status().is_success() {
89            let status = response.status().as_u16();
90            let data = response.collect().await?;
91            let text = String::from_utf8(data.to_bytes().to_vec())?;
92            return Err(S3Error::HttpFailWithBody(status, text));
93        }
94
95        Ok(response)
96    }
97
98    async fn response_data(&self, etag: bool) -> Result<ResponseData, S3Error> {
99        let response = self.response().await?;
100        let status_code = response.status().as_u16();
101        let mut headers = response.headers().clone();
102        let response_headers = headers
103            .clone()
104            .iter()
105            .map(|(k, v)| {
106                (
107                    k.to_string(),
108                    v.to_str()
109                        .unwrap_or("could-not-decode-header-value")
110                        .to_string(),
111                )
112            })
113            .collect::<HashMap<String, String>>();
114        let body_vec = if etag {
115            if let Some(etag) = headers.remove("ETag") {
116                Bytes::from(etag.to_str()?.to_string())
117            } else {
118                Bytes::from("")
119            }
120        } else {
121            response.collect().await?.to_bytes()
122        };
123        Ok(ResponseData::new(body_vec, status_code, response_headers))
124    }
125
126    async fn response_data_to_writer<T: tokio::io::AsyncWrite + Send + Unpin>(
127        &self,
128        writer: &mut T,
129    ) -> Result<u16, S3Error> {
130        let response = self.response().await?;
131        let status_code = response.status();
132
133        let body: Incoming = response.into_body();
134        let stream: BodyStream<Incoming> = BodyStream::new(body);
135
136        let stream_of_bytes = stream
137            .try_filter_map(
138                |frame: hyper::body::Frame<Bytes>| async move { Ok(frame.into_data().ok()) },
139            )
140            .map_err(|err| io::Error::new(io::ErrorKind::Other, err));
141
142        let async_read = tokio_util::io::StreamReader::new(stream_of_bytes);
143        let buffer_size = 512 * 1024; //512 KiB
144        let async_read = tokio::io::BufReader::with_capacity(buffer_size, async_read);
145
146        let mut async_read = std::pin::pin!(async_read);
147
148        io::copy(&mut async_read, writer).await?;
149
150        Ok(status_code.as_u16())
151    }
152
153    async fn response_data_to_stream(&self) -> Result<ResponseDataStream, S3Error> {
154        let response = self.response().await?;
155        let status_code = response.status();
156
157        let body_stream = BodyStream::new(response.into_body());
158
159        Ok(ResponseDataStream {
160            body_stream,
161            status_code: status_code.as_u16(),
162        })
163    }
164
165    async fn response_header(&self) -> Result<(Self::HeaderMap, u16), S3Error> {
166        let response = self.response().await?;
167        let status_code = response.status().as_u16();
168        let headers = response.headers().clone();
169        Ok((headers, status_code))
170    }
171
172    fn datetime(&self) -> OffsetDateTime {
173        self.datetime
174    }
175
176    fn bucket(&self) -> Bucket {
177        self.bucket.clone()
178    }
179
180    fn command(&self) -> Command {
181        self.command.clone()
182    }
183
184    fn path(&self) -> String {
185        self.path.to_string()
186    }
187}
188
189impl<'a> HyperRequest<'a> {
190    pub fn new(
191        bucket: &'a Bucket,
192        path: &'a str,
193        command: Command<'a>,
194    ) -> Result<HyperRequest<'a>, S3Error> {
195        bucket.credentials_refresh()?;
196        Ok(Self {
197            bucket,
198            path,
199            command,
200            datetime: OffsetDateTime::now_utc(),
201        })
202    }
203}
204
205#[cfg(test)]
206mod tests {
207    use crate::bucket::Bucket;
208    use crate::command::Command;
209    use crate::request::tokio_backend::HyperRequest;
210    use crate::request::Request;
211    use awscreds::Credentials;
212    use http::header::{HOST, RANGE};
213
214    // Fake keys - otherwise using Credentials::default will use actual user
215    // credentials if they exist.
216    fn fake_credentials() -> Credentials {
217        let access_key = "AKIAIOSFODNN7EXAMPLE";
218        let secert_key = "wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY";
219        Credentials::new(Some(access_key), Some(secert_key), None, None, None).unwrap()
220    }
221
222    #[test]
223    fn url_uses_https_by_default() {
224        let region = "custom-region".parse().unwrap();
225        let bucket = Bucket::new("my-first-bucket", region, fake_credentials()).unwrap();
226        let path = "/my-first/path";
227        let request = HyperRequest::new(&bucket, path, Command::GetObject).unwrap();
228
229        assert_eq!(request.url().unwrap().scheme(), "https");
230
231        let headers = request.headers().unwrap();
232        let host = headers.get(HOST).unwrap();
233
234        assert_eq!(*host, "my-first-bucket.custom-region".to_string());
235    }
236
237    #[test]
238    fn url_uses_https_by_default_path_style() {
239        let region = "custom-region".parse().unwrap();
240        let bucket = Bucket::new("my-first-bucket", region, fake_credentials())
241            .unwrap()
242            .with_path_style();
243        let path = "/my-first/path";
244        let request = HyperRequest::new(&bucket, path, Command::GetObject).unwrap();
245
246        assert_eq!(request.url().unwrap().scheme(), "https");
247
248        let headers = request.headers().unwrap();
249        let host = headers.get(HOST).unwrap();
250
251        assert_eq!(*host, "custom-region".to_string());
252    }
253
254    #[test]
255    fn url_uses_scheme_from_custom_region_if_defined() {
256        let region = "http://custom-region".parse().unwrap();
257        let bucket = Bucket::new("my-second-bucket", region, fake_credentials()).unwrap();
258        let path = "/my-second/path";
259        let request = HyperRequest::new(&bucket, path, Command::GetObject).unwrap();
260
261        assert_eq!(request.url().unwrap().scheme(), "http");
262
263        let headers = request.headers().unwrap();
264        let host = headers.get(HOST).unwrap();
265        assert_eq!(*host, "my-second-bucket.custom-region".to_string());
266    }
267
268    #[test]
269    fn url_uses_scheme_from_custom_region_if_defined_with_path_style() {
270        let region = "http://custom-region".parse().unwrap();
271        let bucket = Bucket::new("my-second-bucket", region, fake_credentials())
272            .unwrap()
273            .with_path_style();
274        let path = "/my-second/path";
275        let request = HyperRequest::new(&bucket, path, Command::GetObject).unwrap();
276
277        assert_eq!(request.url().unwrap().scheme(), "http");
278
279        let headers = request.headers().unwrap();
280        let host = headers.get(HOST).unwrap();
281        assert_eq!(*host, "custom-region".to_string());
282    }
283
284    #[test]
285    fn test_get_object_range_header() {
286        let region = "http://custom-region".parse().unwrap();
287        let bucket = Bucket::new("my-second-bucket", region, fake_credentials())
288            .unwrap()
289            .with_path_style();
290        let path = "/my-second/path";
291
292        let request = HyperRequest::new(
293            &bucket,
294            path,
295            Command::GetObjectRange {
296                start: 0,
297                end: None,
298            },
299        )
300        .unwrap();
301        let headers = request.headers().unwrap();
302        let range = headers.get(RANGE).unwrap();
303        assert_eq!(range, "bytes=0-");
304
305        let request = HyperRequest::new(
306            &bucket,
307            path,
308            Command::GetObjectRange {
309                start: 0,
310                end: Some(1),
311            },
312        )
313        .unwrap();
314        let headers = request.headers().unwrap();
315        let range = headers.get(RANGE).unwrap();
316        assert_eq!(range, "bytes=0-1");
317    }
318}
s3/request/tokio_backend.rs

s3/request/
tokio_backend.rs
