eva_sdk/
http.rs

1use eva_common::op::Op;
2use eva_common::prelude::*;
3use http_body_util::{BodyExt, Empty, Full};
4use hyper::body::{Bytes, Incoming};
5use hyper::{StatusCode, Uri};
6use hyper_tls::HttpsConnector;
7use hyper_util::client::legacy::{Client as HyperClient, connect::HttpConnector};
8use hyper_util::rt::TokioExecutor;
9use serde::{Deserialize, Serialize};
10use simple_pool::ResourcePool;
11use std::collections::BTreeMap;
12use std::time::Duration;
13
14type Resource = HyperClient<HttpsConnector<HttpConnector>, Empty<Bytes>>;
15
16pub type RawResponse = hyper::Response<Incoming>;
17pub type BufferedResponse = hyper::Response<Full<Bytes>>;
18
19pub const MAX_REDIRECTS: usize = 10;
20
21pub struct Client {
22    pool: ResourcePool<Resource>,
23    timeout: Duration,
24    max_redirects: usize,
25    follow_redirects: bool,
26}
27
28#[derive(Serialize, Deserialize, Debug, Clone)]
29pub struct Response {
30    status: u16,
31    headers: BTreeMap<String, String>,
32    body: Vec<u8>,
33}
34
35impl Response {
36    #[inline]
37    pub fn status(&self) -> u16 {
38        self.status
39    }
40    #[inline]
41    pub fn headers(&self) -> &BTreeMap<String, String> {
42        &self.headers
43    }
44    #[inline]
45    pub fn body(&self) -> &[u8] {
46        &self.body
47    }
48}
49
50impl TryFrom<Response> for BufferedResponse {
51    type Error = Error;
52    fn try_from(resp: Response) -> EResult<Self> {
53        let mut r = hyper::http::Response::builder();
54        for (header, value) in resp.headers {
55            r = r.header(header, value);
56        }
57        r.status(StatusCode::from_u16(resp.status).map_err(Error::failed)?)
58            .body(Full::new(Bytes::from(resp.body)))
59            .map_err(Error::failed)
60    }
61}
62
63impl Client {
64    pub fn new(pool_size: usize, timeout: Duration) -> Self {
65        let pool: ResourcePool<Resource> = <_>::default();
66        for _ in 0..=pool_size {
67            let https = HttpsConnector::new();
68            let client = HyperClient::builder(TokioExecutor::new())
69                .pool_idle_timeout(timeout)
70                .build(https);
71            pool.append(client);
72        }
73        Self {
74            pool,
75            timeout,
76            max_redirects: MAX_REDIRECTS,
77            follow_redirects: true,
78        }
79    }
80    #[inline]
81    pub fn max_redirects(mut self, max_redirects: usize) -> Self {
82        self.max_redirects = max_redirects;
83        self
84    }
85    #[inline]
86    pub fn follow_redirects(mut self, follow: bool) -> Self {
87        self.follow_redirects = follow;
88        self
89    }
90    pub async fn get(&self, url: &str) -> EResult<RawResponse> {
91        let op = Op::new(self.timeout);
92        let mut target_uri: Uri = {
93            if url.starts_with("http://") || url.starts_with("https://") {
94                url.parse()
95            } else {
96                format!("http://{url}").parse()
97            }
98        }
99        .map_err(|e| Error::invalid_params(format!("invalid url {}: {}", url, e)))?;
100        let client = tokio::time::timeout(op.timeout()?, self.pool.get()).await?;
101        let mut rdr = 0;
102        loop {
103            let res = tokio::time::timeout(op.timeout()?, client.get(target_uri.clone()))
104                .await?
105                .map_err(Error::io)?;
106            if self.follow_redirects
107                && (res.status() == StatusCode::MOVED_PERMANENTLY
108                    || res.status() == StatusCode::TEMPORARY_REDIRECT
109                    || res.status() == StatusCode::FOUND)
110            {
111                if rdr > self.max_redirects {
112                    return Err(Error::io("too many redirects"));
113                }
114                rdr += 1;
115                if let Some(loc) = res.headers().get(hyper::header::LOCATION) {
116                    let location_uri: Uri = loc
117                        .to_str()
118                        .map_err(|e| Error::invalid_params(format!("invalid redirect url: {e}")))?
119                        .parse()
120                        .map_err(|e| Error::invalid_params(format!("invalid redirect url: {e}")))?;
121                    let loc_parts = location_uri.into_parts();
122                    let mut parts = target_uri.into_parts();
123                    if loc_parts.scheme.is_some() {
124                        parts.scheme = loc_parts.scheme;
125                    }
126                    if loc_parts.authority.is_some() {
127                        parts.authority = loc_parts.authority;
128                    }
129                    parts.path_and_query = loc_parts.path_and_query;
130                    target_uri = Uri::from_parts(parts)
131                        .map_err(|e| Error::invalid_params(format!("invalid redirect url: {e}")))?;
132                } else {
133                    return Err(Error::io("invalid redirect"));
134                }
135            } else {
136                return Ok(res);
137            }
138        }
139    }
140    pub async fn get_response(&self, url: &str) -> EResult<Response> {
141        let op = Op::new(self.timeout);
142        let resp = self.get(url).await?;
143        let status = resp.status().as_u16();
144        let mut headers = BTreeMap::new();
145        for (header, value) in resp.headers() {
146            headers.insert(
147                header.to_string(),
148                value.to_str().unwrap_or_default().to_owned(),
149            );
150        }
151        let body = tokio::time::timeout(op.timeout()?, resp.into_body().collect())
152            .await?
153            .map_err(Error::io)?
154            .to_bytes()
155            .to_vec();
156        Ok(Response {
157            status,
158            headers,
159            body,
160        })
161    }
162}
eva_sdk/http.rs

eva_sdk/
http.rs
