feat: support listeners & tunnel

.cargo/config.toml

@@ -3,4 +3,4 @@ target-dir = "target"
 rustflags = ["--cfg", "tokio_unstable"]
 
 [env]
-RUST_LOG = { value = "clash=trace" }
+RUST_LOG = { value = "clash=trace" }

clash/tests/data/config/listeners/tunnel.yaml

@@ -0,0 +1,10 @@
+port: 8080
+socks-port: 8081
+log-level: trace
+listeners:
+- name: tunnel-in
+  type: tunnel
+  port: 15201
+  listen: 127.0.0.1
+  network: [tcp, udp]
+  target: 127.0.0.1:5201

clash_lib/Cargo.toml

@@ -61,6 +61,7 @@ uuid = { version = "1", features = ["v4", "fast-rng", "macro-diagnostics", "serd
 foreign-types-shared = "0.3"
 network-interface = { version = "2", git = "https://github.com/Watfaq/network-interface.git", rev = "db9f1bcd2faa5122f694283bd300bf42d166d2f3" } # temp workaround for https://github.com/cross-rs/cross/issues/1627
 base64 = "0.22"
+arc-swap = "1"
 
 httparse = "1"
 h2 = "0.4"
@@ -100,6 +101,11 @@ serde_yaml = "0.9"
 serde_json = "1"
 erased-serde = "0.4"
 
+
+# Macro Magic
+educe = "0.6"
+enum_dispatch = "0.3"
+
 # DNS
 watfaq-dns = { version = "0.1.1" }
 hickory-client = "0.25.0-alpha.2"

clash_lib/src/app/api/handlers/config.rs

@@ -18,21 +18,21 @@ use crate::{
         api::AppState,
         dispatcher,
         dns::ThreadSafeDNSResolver,
-        inbound::manager::{Ports, ThreadSafeInboundManager},
+        inbound::manager::{InboundManager, Ports},
     },
     config::{def, internal::config::BindAddress},
 };
 
 #[derive(Clone)]
 struct ConfigState {
-    inbound_manager: ThreadSafeInboundManager,
+    inbound_manager: Arc<InboundManager>,
     dispatcher: Arc<dispatcher::Dispatcher>,
     global_state: Arc<Mutex<GlobalState>>,
     dns_resolver: ThreadSafeDNSResolver,
 }
 
 pub fn routes(
-    inbound_manager: ThreadSafeInboundManager,
+    inbound_manager: Arc<InboundManager>,
     dispatcher: Arc<dispatcher::Dispatcher>,
     global_state: Arc<Mutex<GlobalState>>,
     dns_resolver: ThreadSafeDNSResolver,
@@ -51,31 +51,24 @@ pub fn routes(
 }
 
 async fn get_configs(State(state): State<ConfigState>) -> impl IntoResponse {
-    let inbound_manager = state.inbound_manager.lock().await;
     let run_mode = state.dispatcher.get_mode().await;
     let global_state = state.global_state.lock().await;
     let dns_resolver = state.dns_resolver;
 
-    let ports = inbound_manager.get_ports();
+    let ports = state.inbound_manager.get_ports().await;
 
     axum::response::Json(PatchConfigRequest {
         port: ports.port,
         socks_port: ports.socks_port,
         redir_port: ports.redir_port,
         tproxy_port: ports.tproxy_port,
         mixed_port: ports.mixed_port,
-        bind_address: Some(inbound_manager.get_bind_address().to_string()),
+        bind_address: Some(state.inbound_manager.get_bind_address().0.to_string()),
 
         mode: Some(run_mode),
         log_level: Some(global_state.log_level),
         ipv6: Some(dns_resolver.ipv6()),
-        allow_lan: Some(match inbound_manager.get_bind_address() {
-            BindAddress::Any => true,
-            BindAddress::One(one) => match one {
-                crate::proxy::utils::Interface::IpAddr(ip) => !ip.is_loopback(),
-                crate::proxy::utils::Interface::Name(iface) => iface != "lo",
-            },
-        }),
+        allow_lan: Some(state.inbound_manager.get_bind_address().0.is_unspecified()),
     })
 }
 
@@ -187,12 +180,13 @@ async fn patch_configs(
         );
     }
 
-    let mut inbound_manager = state.inbound_manager.lock().await;
-
+    let inbound_manager = state.inbound_manager.clone();
+    let mut need_restart = false;
     if let Some(bind_address) = payload.bind_address.clone() {
         match bind_address.parse::<BindAddress>() {
             Ok(bind_address) => {
-                inbound_manager.set_bind_address(bind_address);
+                inbound_manager.set_bind_address(bind_address).await;
+                need_restart = true;
             }
             Err(_) => {
                 return (
@@ -208,7 +202,7 @@ async fn patch_configs(
 
     if payload.rebuild_listeners() {
         // TODO: maybe buggy
-        let current_ports = inbound_manager.get_ports();
+        let current_ports = inbound_manager.get_ports().await;
 
         let ports = Ports {
             port: payload.port.or(current_ports.port),
@@ -217,14 +211,11 @@ async fn patch_configs(
             tproxy_port: payload.tproxy_port.or(current_ports.tproxy_port),
             mixed_port: payload.mixed_port.or(current_ports.mixed_port),
         };
-
-        inbound_manager.rebuild_listeners(ports);
-
-        global_state.inbound_listener_handle.abort();
-
-        let r = inbound_manager.get_runner().unwrap();
-
-        global_state.inbound_listener_handle = tokio::spawn(r);
+        inbound_manager.change_ports(ports).await;
+        need_restart = true;
+    }
+    if need_restart {
+        inbound_manager.restart().await;
     }
 
     if let Some(mode) = payload.mode {