refactor(liveman): proxy plugin

libs/net4mqtt/src/kxdns.rs

@@ -1,11 +1,11 @@
 const SPLIT: char = '.';
 
 pub struct Kxdns {
-    domain: &'static str,
+    domain: String,
 }
 
 impl Kxdns {
-    pub fn new(domain: &'static str) -> Self {
+    pub fn new(domain: String) -> Self {
         Self { domain }
     }
 
@@ -21,7 +21,7 @@ impl Kxdns {
 
 #[test]
 fn test_kxdns() {
-    let kxdns = Kxdns::new("kxdns.com");
+    let kxdns = Kxdns::new("kxdns.com".to_string());
     let key = "test";
     let domain = kxdns.registry(key);
     assert_eq!(domain, "test.kxdns.com");

libs/net4mqtt/src/tests.rs

@@ -640,7 +640,7 @@ async fn test_socks_simple() {
     let res = client
         .get(format!(
             "http://{}/",
-            kxdns::Kxdns::new(DOMAIN_SUFFIX).registry("0")
+            kxdns::Kxdns::new(DOMAIN_SUFFIX.to_string()).registry("0")
         ))
         .send()
         .await
@@ -660,7 +660,7 @@ async fn test_socks_simple() {
     let res = client
         .get(format!(
             "http://{}/",
-            kxdns::Kxdns::new(DOMAIN_SUFFIX).registry("0")
+            kxdns::Kxdns::new(DOMAIN_SUFFIX.to_string()).registry("0")
         ))
         .send()
         .await
@@ -706,7 +706,7 @@ async fn test_socks_restart() {
         let res = client
             .get(format!(
                 "http://{}/",
-                kxdns::Kxdns::new(DOMAIN_SUFFIX).registry("0")
+                kxdns::Kxdns::new(DOMAIN_SUFFIX.to_string()).registry("0")
             ))
             .send()
             .await
@@ -726,7 +726,7 @@ async fn test_socks_restart() {
         let res = client
             .get(format!(
                 "http://{}/",
-                kxdns::Kxdns::new(DOMAIN_SUFFIX).registry("0")
+                kxdns::Kxdns::new(DOMAIN_SUFFIX.to_string()).registry("0")
             ))
             .send()
             .await
@@ -795,7 +795,7 @@ async fn test_socks_multiple_server() {
         let res = client
             .get(format!(
                 "http://{}/",
-                kxdns::Kxdns::new(DOMAIN_SUFFIX).registry(&id.to_string())
+                kxdns::Kxdns::new(DOMAIN_SUFFIX.to_string()).registry(&id.to_string())
             ))
             .send()
             .await

liveman/src/config.rs

@@ -39,8 +39,10 @@ pub struct Net4mqtt {
     pub mqtt_url: String,
     #[serde(default)]
     pub alias: String,
-    #[serde(default = "default_socks_listen")]
+    #[serde(default = "default_net4mqtt_listen")]
     pub listen: SocketAddr,
+    #[serde(default = "default_net4mqtt_domain")]
+    pub domain: String,
 }
 
 #[cfg(feature = "net4mqtt")]
@@ -49,16 +51,22 @@ impl Default for Net4mqtt {
         Self {
             mqtt_url: String::new(),
             alias: String::new(),
-            listen: default_socks_listen(),
+            listen: default_net4mqtt_listen(),
+            domain: default_net4mqtt_domain(),
         }
     }
 }
 
 #[cfg(feature = "net4mqtt")]
-fn default_socks_listen() -> SocketAddr {
+fn default_net4mqtt_listen() -> SocketAddr {
     SocketAddr::from_str("0.0.0.0:1077").expect("invalid listen socks address")
 }
 
+#[cfg(feature = "net4mqtt")]
+fn default_net4mqtt_domain() -> String {
+    String::from("net4mqtt.local")
+}
+
 #[derive(Debug, Clone, Deserialize, Serialize)]
 pub struct Http {
     #[serde(default = "default_http_listen")]

liveman/src/lib.rs

@@ -44,40 +44,37 @@ where
     F: Future<Output = ()> + Send + 'static,
 {
     #[cfg(feature = "net4mqtt")]
-    let net4mqtt_domain = "net4mqtt.local";
-
-    #[cfg(feature = "net4mqtt")]
-    let proxy_addr = match cfg.net4mqtt.clone() {
-        Some(c) => Some((c.listen, net4mqtt_domain.to_string())),
+    let proxy_plugin = match cfg.net4mqtt.clone() {
+        Some(c) => {
+            // References: https://github.com/seanmonstar/reqwest/issues/899
+            let target = reqwest::Url::parse(&format!("socks5h://{}", c.listen)).unwrap();
+            Some(reqwest::Proxy::custom(move |url| match url.host_str() {
+                Some(host) => {
+                    if host.ends_with(c.domain.as_str()) {
+                        Some(target.clone())
+                    } else {
+                        None
+                    }
+                }
+                None => None,
+            }))
+        }
         None => None,
     };
     #[cfg(not(feature = "net4mqtt"))]
-    let proxy_addr = None;
-
-    let client_builder = reqwest::Client::builder()
-        .connect_timeout(Duration::from_millis(1000))
-        .timeout(Duration::from_millis(10000));
-
-    let client = if let Some((addr, domain)) = proxy_addr {
-        // References: https://github.com/seanmonstar/reqwest/issues/899
-        let target = reqwest::Url::parse(format!("socks5h://{}", addr).as_str()).unwrap();
-        client_builder.proxy(reqwest::Proxy::custom(move |url| match url.host_str() {
-            Some(host) => {
-                if host.ends_with(domain.as_str()) {
-                    Some(target.clone())
-                } else {
-                    None
-                }
-            }
-            None => None,
-        }))
+    let proxy_plugin = None;
+
+    let client_req = reqwest::Client::builder();
+    let client_mem = reqwest::Client::builder()
+        .connect_timeout(Duration::from_millis(500))
+        .timeout(Duration::from_millis(1000));
+    let (client_req, client_mem) = if let Some(proxy) = proxy_plugin {
+        (client_req.proxy(proxy.clone()), client_mem.proxy(proxy))
     } else {
-        client_builder
-    }
-    .build()
-    .unwrap();
+        (client_req, client_mem)
+    };
 
-    let store = MemStorage::new(client.clone());
+    let store = MemStorage::new(client_mem.build().unwrap());
     let nodes = store.get_map_nodes_mut();
     for v in cfg.nodes.clone() {
         nodes
@@ -104,6 +101,8 @@ where
             let (sender, mut receiver) =
                 tokio::sync::mpsc::channel::<(String, String, Vec<u8>)>(10);
 
+            let domain = c.domain.clone();
+
             std::thread::spawn(move || {
                 tokio::runtime::Runtime::new()
                     .unwrap()
@@ -113,7 +112,7 @@ where
                             &c.mqtt_url,
                             listener,
                             ("-", &c.alias.clone()),
-                            Some(net4mqtt_domain.to_string()),
+                            Some(c.domain),
                             Some(net4mqtt::proxy::VDataConfig {
                                 receiver: Some(sender),
                                 ..Default::default()
@@ -126,7 +125,7 @@ where
             });
 
             std::thread::spawn(move || {
-                let dns = net4mqtt::kxdns::Kxdns::new(net4mqtt_domain);
+                let dns = net4mqtt::kxdns::Kxdns::new(domain);
                 tokio::runtime::Runtime::new()
                     .unwrap()
                     .block_on(async move {
@@ -159,7 +158,7 @@ where
 
     let app_state = AppState {
         config: cfg.clone(),
-        client,
+        client: client_req.build().unwrap(),
         storage: store,
     };