cyfs_lib/requestor/
tcp.rs

1use super::requestor::*;
2use cyfs_base::*;
3
4use async_std::net::{SocketAddr, TcpStream};
5use http_types::{Request, Response};
6use std::str::FromStr;
7
8#[derive(Clone)]
9pub struct TcpHttpRequestor {
10    service_addr: SocketAddr,
11}
12
13impl TcpHttpRequestor {
14    pub fn new(service_addr: &str) -> Self {
15        let service_addr = SocketAddr::from_str(&service_addr).unwrap();
16        Self { service_addr }
17    }
18}
19
20#[async_trait::async_trait]
21impl HttpRequestor for TcpHttpRequestor {
22    async fn request_ext(
23        &self,
24        req: &mut Option<Request>,
25        conn_info: Option<&mut HttpRequestConnectionInfo>,
26    ) -> BuckyResult<Response> {
27        debug!(
28            "will http-local request to {}, url={}",
29            self.remote_addr(),
30            req.as_ref().unwrap().url()
31        );
32
33        let begin = std::time::Instant::now();
34        let tcp_stream = TcpStream::connect(self.service_addr).await.map_err(|e| {
35            let msg = format!(
36                "tcp connect to {} error! during={}ms, {}",
37                self.service_addr,
38                begin.elapsed().as_millis(),
39                e
40            );
41            error!("{}", msg);
42
43            BuckyError::new(BuckyErrorCode::ConnectFailed, msg)
44        })?;
45
46        info!(
47            "tcp connect to {} success, during={}ms",
48            self.remote_addr(),
49            begin.elapsed().as_millis(),
50        );
51
52        if let Some(conn_info) = conn_info {
53            *conn_info = HttpRequestConnectionInfo::Tcp((
54                tcp_stream.local_addr().unwrap(),
55                tcp_stream.peer_addr().unwrap(),
56            ));
57        }
58
59        let req = req.take().unwrap();
60        let req = self.add_default_headers(req);
61        match async_h1::connect(tcp_stream, req).await {
62            Ok(resp) => {
63                info!(
64                    "http-tcp request to {} success! during={}ms",
65                    self.remote_addr(),
66                    begin.elapsed().as_millis()
67                );
68                Ok(resp)
69            }
70            Err(e) => {
71                let msg = format!(
72                    "http-tcp request to {} failed! during={}ms, {}",
73                    self.remote_addr(),
74                    begin.elapsed().as_millis(),
75                    e,
76                );
77                error!("{}", msg);
78                Err(BuckyError::from(msg))
79            }
80        }
81    }
82
83    fn remote_addr(&self) -> String {
84        self.service_addr.to_string()
85    }
86
87    fn remote_device(&self) -> Option<DeviceId> {
88        None
89    }
90
91    fn clone_requestor(&self) -> Box<dyn HttpRequestor> {
92        Box::new(self.clone())
93    }
94
95    async fn stop(&self) {
96        // do nothing
97    }
98}
cyfs_lib/requestor/tcp.rs

cyfs_lib/requestor/
tcp.rs
