[mio] 简化vnt创建

vnt/src/core/conn.rs

@@ -0,0 +1,354 @@
+use std::collections::HashMap;
+use std::io;
+use std::net::Ipv4Addr;
+use std::sync::atomic::{AtomicU64, Ordering};
+use std::sync::mpsc::{sync_channel, Receiver};
+use std::sync::Arc;
+use std::time::Duration;
+
+use crossbeam_utils::atomic::AtomicCell;
+use parking_lot::{Mutex, RwLock};
+use rand::Rng;
+
+use tun::device::IFace;
+
+use crate::channel::context::Context;
+use crate::channel::idle::Idle;
+use crate::channel::punch::{NatInfo, Punch};
+use crate::channel::{init_channel, init_context, Route, RouteKey};
+use crate::cipher::Cipher;
+#[cfg(feature = "server_encrypt")]
+use crate::cipher::RsaCipher;
+use crate::core::Config;
+use crate::external_route::{AllowExternalRoute, ExternalRoute};
+use crate::handle::recv_data::RecvDataHandler;
+use crate::handle::{
+    maintain, tun_tap, BaseConfigInfo, ConnectStatus, CurrentDeviceInfo, PeerDeviceInfo,
+};
+use crate::nat::NatTest;
+use crate::util::{Scheduler, StopManager, U64Adder, WatchU64Adder};
+use crate::{nat, tun_tap_device, DeviceInfo, VntCallback};
+
+#[derive(Clone)]
+pub struct Vnt {
+    stop_manager: StopManager,
+    config: Config,
+    current_device: Arc<AtomicCell<CurrentDeviceInfo>>,
+    nat_test: NatTest,
+    device_list: Arc<Mutex<(u16, Vec<PeerDeviceInfo>)>>,
+    context: Context,
+    peer_nat_info_map: Arc<RwLock<HashMap<Ipv4Addr, NatInfo>>>,
+    down_count_watcher: WatchU64Adder,
+    up_count_watcher: Arc<AtomicU64>,
+}
+
+impl Vnt {
+    pub fn new<Call: VntCallback>(config: Config, callback: Call) -> io::Result<Self> {
+        //服务端非对称加密
+        #[cfg(feature = "server_encrypt")]
+        let rsa_cipher: Arc<Mutex<Option<RsaCipher>>> = Arc::new(Mutex::new(None));
+        //服务端对称加密
+        let server_cipher: Cipher = if config.server_encrypt {
+            let mut key = [0u8; 32];
+            rand::thread_rng().fill(&mut key);
+            Cipher::new_key(key, config.token.clone())?
+        } else {
+            Cipher::None
+        };
+        let finger = if config.finger {
+            Some(config.token.clone())
+        } else {
+            None
+        };
+        //客户端对称加密
+        let client_cipher =
+            Cipher::new_password(config.cipher_model, config.password.clone(), finger);
+        //当前设备信息
+        let current_device = Arc::new(AtomicCell::new(CurrentDeviceInfo::new0(
+            config.server_address,
+        )));
+        //设备列表
+        let device_list: Arc<Mutex<(u16, Vec<PeerDeviceInfo>)>> =
+            Arc::new(Mutex::new((0, Vec::with_capacity(16))));
+        //基础信息
+        let config_info = BaseConfigInfo::new(
+            config.name.clone(),
+            config.token.clone(),
+            config.ip,
+            config.password.is_some(),
+            config.device_id.clone(),
+            config.server_address_str.clone(),
+        );
+        let ports = config.ports.as_ref().map_or(vec![0, 0], |v| {
+            if v.is_empty() {
+                vec![0, 0]
+            } else {
+                v.clone()
+            }
+        });
+        //通道上下文
+        let (context, tcp_listener) = init_context(ports, config.first_latency, config.tcp)?;
+        let local_ipv4 = nat::local_ipv4();
+        let local_ipv6 = nat::local_ipv6();
+        let udp_ports = context.main_local_udp_port()?;
+        let tcp_port = tcp_listener.local_addr()?.port();
+        //nat检测工具
+        let nat_test = NatTest::new(
+            config.stun_server.clone(),
+            local_ipv4,
+            local_ipv6,
+            udp_ports,
+            tcp_port,
+        );
+
+        // 虚拟网卡
+        let device = tun_tap_device::create_device(&config)?;
+        let tun_info = DeviceInfo::new(device.name()?, device.version()?);
+        callback.create_tun(tun_info);
+        // 服务停止管理器
+        let stop_manager = {
+            let callback = callback.clone();
+            StopManager::new(move || callback.stop())
+        };
+        // 定时器
+        let scheduler = Scheduler::new(stop_manager.clone())?;
+        let external_route = ExternalRoute::new(config.in_ips.clone());
+        let out_external_route = AllowExternalRoute::new(config.out_ips.clone());
+
+        #[cfg(feature = "ip_proxy")]
+        let proxy_map = if !config.out_ips.is_empty() && !config.no_proxy {
+            Some(crate::ip_proxy::init_proxy(
+                context.clone(),
+                scheduler.clone(),
+                stop_manager.clone(),
+                current_device.clone(),
+                client_cipher.clone(),
+            )?)
+        } else {
+            None
+        };
+        let (punch_sender, punch_receiver) = sync_channel(3);
+        let peer_nat_info_map: Arc<RwLock<HashMap<Ipv4Addr, NatInfo>>> =
+            Arc::new(RwLock::new(HashMap::with_capacity(16)));
+        let down_counter = U64Adder::with_capacity(8);
+        let down_count_watcher = down_counter.watch();
+        let handler = RecvDataHandler::new(
+            #[cfg(feature = "server_encrypt")]
+            rsa_cipher,
+            server_cipher.clone(),
+            client_cipher.clone(),
+            current_device.clone(),
+            device.clone(),
+            device_list.clone(),
+            config_info.clone(),
+            nat_test.clone(),
+            callback.clone(),
+            config.relay,
+            punch_sender,
+            peer_nat_info_map.clone(),
+            external_route.clone(),
+            out_external_route,
+            #[cfg(feature = "ip_proxy")]
+            proxy_map.clone(),
+            down_counter,
+        );
+
+        //初始化网络数据通道
+        let (udp_socket_sender, tcp_socket_sender) =
+            init_channel(tcp_listener, context.clone(), stop_manager.clone(), handler)?;
+        // 打洞逻辑
+        let punch = Punch::new(
+            context.clone(),
+            config.punch_model,
+            config.tcp,
+            tcp_socket_sender.clone(),
+        );
+        let up_counter = Arc::new(AtomicU64::new(0));
+        let up_count_watcher = up_counter.clone();
+        tun_tap::tun_handler::start(
+            stop_manager.clone(),
+            context.clone(),
+            device.clone(),
+            current_device.clone(),
+            external_route,
+            #[cfg(feature = "ip_proxy")]
+            proxy_map,
+            client_cipher.clone(),
+            server_cipher.clone(),
+            config.parallel,
+            up_counter,
+        )?;
+        maintain::idle_gateway(
+            &scheduler,
+            context.clone(),
+            current_device.clone(),
+            config_info.clone(),
+            tcp_socket_sender.clone(),
+            callback.clone(),
+            0,
+        );
+        {
+            let context = context.clone();
+            let nat_test = nat_test.clone();
+            let device_list = device_list.clone();
+            let current_device = current_device.clone();
+            let relay = config.relay;
+            if !relay {
+                // 定时nat探测
+                maintain::retrieve_nat_type(
+                    &scheduler,
+                    context.clone(),
+                    nat_test.clone(),
+                    udp_socket_sender,
+                );
+            }
+            //延迟启动
+            scheduler.timeout(Duration::from_secs(3), move |scheduler| {
+                start(
+                    scheduler,
+                    context,
+                    nat_test,
+                    device_list,
+                    current_device,
+                    client_cipher,
+                    server_cipher,
+                    punch_receiver,
+                    config_info,
+                    punch,
+                    callback,
+                    relay,
+                );
+            });
+        }
+
+        Ok(Self {
+            stop_manager,
+            config,
+            current_device,
+            nat_test,
+            device_list,
+            context,
+            peer_nat_info_map,
+            down_count_watcher,
+            up_count_watcher,
+        })
+    }
+}
+
+pub fn start<Call: VntCallback>(
+    scheduler: &Scheduler,
+    context: Context,
+    nat_test: NatTest,
+    device_list: Arc<Mutex<(u16, Vec<PeerDeviceInfo>)>>,
+    current_device: Arc<AtomicCell<CurrentDeviceInfo>>,
+    client_cipher: Cipher,
+    server_cipher: Cipher,
+    punch_receiver: Receiver<(Ipv4Addr, NatInfo)>,
+    config_info: BaseConfigInfo,
+    punch: Punch,
+    callback: Call,
+    relay: bool,
+) {
+    // 定时心跳
+    maintain::heartbeat(
+        &scheduler,
+        context.clone(),
+        current_device.clone(),
+        device_list.clone(),
+        client_cipher.clone(),
+        server_cipher.clone(),
+    );
+    // 路由空闲检测逻辑
+    let idle = Idle::new(Duration::from_secs(10), context.clone());
+    // 定时空闲检查
+    maintain::idle_route(
+        &scheduler,
+        idle,
+        context.clone(),
+        current_device.clone(),
+        callback,
+    );
+    // 定时客户端中继检测
+    maintain::client_relay(
+        &scheduler,
+        context.clone(),
+        current_device.clone(),
+        device_list.clone(),
+        client_cipher.clone(),
+    );
+    // 定时地址探测
+    maintain::addr_request(
+        &scheduler,
+        context.clone(),
+        current_device.clone(),
+        server_cipher.clone(),
+        config_info.clone(),
+    );
+    if !relay {
+        // 定时打洞
+        maintain::punch(
+            &scheduler,
+            context.clone(),
+            nat_test.clone(),
+            device_list.clone(),
+            current_device.clone(),
+            client_cipher.clone(),
+            punch_receiver,
+            punch,
+        );
+    }
+}
+
+impl Vnt {
+    pub fn name(&self) -> &str {
+        &self.config.name
+    }
+    pub fn server_encrypt(&self) -> bool {
+        self.config.server_encrypt
+    }
+    pub fn client_encrypt(&self) -> bool {
+        self.config.password.is_some()
+    }
+    pub fn current_device(&self) -> CurrentDeviceInfo {
+        self.current_device.load()
+    }
+    pub fn peer_nat_info(&self, ip: &Ipv4Addr) -> Option<NatInfo> {
+        self.peer_nat_info_map.read().get(ip).cloned()
+    }
+    pub fn connection_status(&self) -> ConnectStatus {
+        self.current_device.load().status
+    }
+    pub fn nat_info(&self) -> NatInfo {
+        self.nat_test.nat_info()
+    }
+    pub fn device_list(&self) -> Vec<PeerDeviceInfo> {
+        let device_list_lock = self.device_list.lock();
+        let (_epoch, device_list) = device_list_lock.clone();
+        drop(device_list_lock);
+        device_list
+    }
+    pub fn route(&self, ip: &Ipv4Addr) -> Option<Route> {
+        self.context.route_table.route_one(ip)
+    }
+    pub fn is_gateway(&self, ip: &Ipv4Addr) -> bool {
+        self.current_device.load().is_gateway(ip)
+    }
+    pub fn route_key(&self, route_key: &RouteKey) -> Option<Ipv4Addr> {
+        self.context.route_table.route_to_id(route_key)
+    }
+    pub fn route_table(&self) -> Vec<(Ipv4Addr, Vec<Route>)> {
+        self.context.route_table.route_table()
+    }
+    pub fn up_stream(&self) -> u64 {
+        self.up_count_watcher.load(Ordering::Relaxed)
+    }
+    pub fn down_stream(&self) -> u64 {
+        self.down_count_watcher.get()
+    }
+    pub fn stop(&self) {
+        self.stop_manager.stop()
+    }
+    pub fn wait(&self) {
+        self.stop_manager.wait()
+    }
+}

vnt/src/core/mod.rs

@@ -1,569 +1,16 @@
-use std::collections::HashMap;
 use std::io;
-use std::net::UdpSocket;
+use std::net::{Ipv4Addr, SocketAddr};
-use std::net::{Ipv4Addr, SocketAddr, SocketAddrV4};
-use std::net::{TcpListener, TcpStream};
-use std::sync::Arc;
-use std::time::Duration;
 
-use crossbeam_utils::atomic::AtomicCell;
+pub use conn::Vnt;
-use parking_lot::{Mutex, RwLock};
-use rand::Rng;
-use tokio::sync::mpsc::channel;
 
-use crate::channel::channel::{Channel, Context};
+use crate::channel::punch::PunchModel;
-use crate::channel::idle::Idle;
+use crate::cipher::CipherModel;
-use crate::channel::punch::{NatInfo, Punch, PunchModel};
-use crate::channel::sender::ChannelSender;
-use crate::channel::{Route, RouteKey};
-use crate::cipher::{Cipher, CipherModel, RsaCipher};
-use crate::core::status::VntStatusManger;
-use crate::error::Error;
-use crate::external_route::{AllowExternalRoute, ExternalRoute};
-use crate::handle::handshake_handler::HandshakeEnum;
-use crate::handle::recv_handler::ChannelDataHandler;
-use crate::handle::registration_handler::{RegResponse, ReqEnum};
-#[cfg(any(target_os = "linux", target_os = "macos", target_os = "windows"))]
-use crate::handle::tun_tap::tap_handler;
-use crate::handle::tun_tap::tun_handler;
-use crate::handle::{
-    handshake_handler, heartbeat_handler, punch_handler, registration_handler, ConnectStatus,
-    CurrentDeviceInfo, PeerDeviceInfo,
-};
-use crate::igmp_server::IgmpServer;
-use crate::nat::NatTest;
-use crate::tun_tap_device;
-use crate::tun_tap_device::{DeviceReader, DeviceWriter};
 
-pub mod status;
+mod conn;
-pub mod sync;
-
-#[derive(Clone)]
-pub struct Vnt {
-    config: Config,
-    current_device: Arc<AtomicCell<CurrentDeviceInfo>>,
-    context: Context,
-    vnt_status_manager: VntStatusManger,
-    device_writer: DeviceWriter,
-    /// 0. 机器纪元，每一次上线或者下线都会增1，用于感知网络中机器变化
-    /// 服务端和客户端的不一致，则服务端会推送新的设备列表
-    /// 1. 网络中的虚拟ip列表
-    device_list: Arc<Mutex<(u16, Vec<PeerDeviceInfo>)>>,
-    nat_test: NatTest,
-    connect_status: Arc<AtomicCell<ConnectStatus>>,
-    peer_nat_info_map: Arc<RwLock<HashMap<Ipv4Addr, NatInfo>>>,
-}
-
-pub struct VntUtil {
-    config: Config,
-    main_channel: UdpSocket,
-    main_tcp_channel: Option<TcpStream>,
-    response: Option<RegResponse>,
-    iface: Option<(DeviceWriter, DeviceReader)>,
-    server_cipher: Cipher,
-    rsa_cipher: Option<RsaCipher>,
-}
-
-impl VntUtil {
-    pub fn new(config: Config) -> io::Result<VntUtil> {
-        let address: SocketAddr = format!("[::]:{}", config.port).parse().unwrap();
-        //单个udp用同步的性能更好，但是代理和多端口监听用异步更方便，这里将两者结合起来
-        let socket = socket2::Socket::new(socket2::Domain::IPV6, socket2::Type::DGRAM, None)?;
-        socket.set_only_v6(false)?;
-        socket.bind(&address.into())?;
-        let main_channel: UdpSocket = socket.into();
-        main_channel.set_write_timeout(Some(Duration::from_secs(5)))?;
-        main_channel.set_read_timeout(Some(Duration::from_secs(2)))?;
-        let server_cipher = if config.server_encrypt {
-            let mut key = [0u8; 32];
-            rand::thread_rng().fill(&mut key);
-            Cipher::new_key(key, config.token.clone())?
-        } else {
-            Cipher::None
-        };
-        Ok(VntUtil {
-            config,
-            main_channel,
-            main_tcp_channel: None,
-            response: None,
-            iface: None,
-            server_cipher,
-            rsa_cipher: None,
-        })
-    }
-    ///链接
-    pub fn connect(&mut self) -> io::Result<()> {
-        if self.config.tcp {
-            let tcp = TcpStream::connect(self.config.server_address)?;
-            tcp.set_read_timeout(Some(Duration::from_secs(10)))?;
-            let _ = self.main_tcp_channel.insert(tcp);
-        }
-        Ok(())
-    }
-
-    ///握手 用于获取公钥
-    pub fn handshake(&mut self) -> Result<Option<RsaCipher>, HandshakeEnum> {
-        let rsa_cipher = handshake_handler::handshake(
-            &self.main_channel,
-            self.main_tcp_channel.as_mut(),
-            self.config.server_address,
-            self.config.server_encrypt,
-        )?;
-        self.rsa_cipher = rsa_cipher.clone();
-        Ok(rsa_cipher)
-    }
-    /// 加密握手 用于同步密钥
-    pub fn secret_handshake(&mut self) -> Result<(), HandshakeEnum> {
-        handshake_handler::secret_handshake(
-            &self.main_channel,
-            self.main_tcp_channel.as_mut(),
-            self.config.server_address,
-            self.rsa_cipher.as_ref().unwrap(),
-            &self.server_cipher,
-            self.config.token.clone(),
-        )
-    }
-    /// 注册
-    pub fn register(&mut self) -> Result<RegResponse, ReqEnum> {
-        match registration_handler::registration(
-            &self.main_channel,
-            self.main_tcp_channel.as_mut(),
-            &self.server_cipher,
-            self.config.server_address,
-            self.config.token.clone(),
-            self.config.device_id.clone(),
-            self.config.name.clone(),
-            self.config.ip.unwrap_or(Ipv4Addr::UNSPECIFIED),
-            self.config.password.is_some(),
-        ) {
-            Ok(res) => {
-                let _ = self.response.insert(res.clone());
-                Ok(res)
-            }
-            Err(e) => Err(e),
-        }
-    }
-    #[cfg(any(target_os = "android"))]
-    pub fn create_iface(&mut self, vpn_fd: i32) {
-        let (device_writer, device_reader) = tun_tap_device::create(vpn_fd);
-        let _ = self.iface.insert((device_writer, device_reader));
-    }
-    #[cfg(any(target_os = "linux", target_os = "macos", target_os = "windows"))]
-    pub fn create_iface(&mut self) -> io::Result<tun_tap_device::DriverInfo> {
-        if self.iface.is_some() {
-            return Err(io::Error::from(io::ErrorKind::AlreadyExists));
-        }
-        let response = match &self.response {
-            None => {
-                return Err(io::Error::from(io::ErrorKind::AlreadyExists));
-            }
-            Some(res) => res,
-        };
-        let device_type = if self.config.tap {
-            {
-                //删除tun网卡避免ip冲突，因为非正常退出会保留网卡
-                tun_tap_device::delete_device(tun_tap_device::DeviceType::Tun);
-            }
-            tun_tap_device::DeviceType::Tap
-        } else {
-            {
-                //删除tap网卡避免ip冲突，非正常退出会保留网卡
-                tun_tap_device::delete_device(tun_tap_device::DeviceType::Tap);
-            }
-            tun_tap_device::DeviceType::Tun
-        };
-        let mtu = match self.config.mtu {
-            None => {
-                if self.config.password.is_none() {
-                    1450
-                } else {
-                    1410
-                }
-            }
-            Some(mtu) => mtu,
-        };
-        let in_ips = self
-            .config
-            .in_ips
-            .iter()
-            .map(|(dest, mask, _)| (Ipv4Addr::from(*dest & *mask), Ipv4Addr::from(*mask)))
-            .collect::<Vec<(Ipv4Addr, Ipv4Addr)>>();
-
-        let (device_writer, device_reader, driver_info) = tun_tap_device::create_device(
-            device_type,
-            response.virtual_ip,
-            response.virtual_netmask,
-            response.virtual_gateway,
-            in_ips,
-            mtu,
-        )?;
-        let _ = self.iface.insert((device_writer, device_reader));
-        Ok(driver_info)
-    }
-    pub async fn build(self) -> crate::Result<Vnt> {
-        //将读的超时时间清空
-        self.main_channel.set_read_timeout(None)?;
-        let response = match self.response {
-            None => {
-                return Err(Error::Stop("response None".to_string()));
-            }
-            Some(res) => res,
-        };
-        let (device_writer, device_reader) = match self.iface {
-            None => {
-                return Err(Error::Stop("iface None".to_string()));
-            }
-            Some(res) => res,
-        };
-        let config = self.config.clone();
-        let vnt_status_manager = VntStatusManger::new();
-        let finger = if config.finger {
-            Some(config.token.clone())
-        } else {
-            None
-        };
-        let client_cipher =
-            Cipher::new_password(config.cipher_model, config.password.clone(), finger);
-        let virtual_ip = response.virtual_ip;
-        let virtual_gateway = response.virtual_gateway;
-        let virtual_netmask = response.virtual_netmask;
-        let current_device = Arc::new(AtomicCell::new(CurrentDeviceInfo::new(
-            virtual_ip,
-            virtual_gateway,
-            virtual_netmask,
-            config.server_address,
-        )));
-
-        let (cone_sender, cone_receiver) = channel(3);
-        let (symmetric_sender, symmetric_receiver) = channel(2);
-        let (tcp_sender, tcp_receiver) = if let Some(main_tcp_channel) = self.main_tcp_channel {
-            (Some(main_tcp_channel.try_clone()?), Some(main_tcp_channel))
-        } else {
-            (None, None)
-        };
-        let tcp_listener = TcpListener::bind(format!("[::]:{}", config.port))?;
-        let local_tcp_port = tcp_listener.local_addr()?.port();
-        let context = Context::new(
-            self.main_channel,
-            tcp_sender,
-            current_device.clone(),
-            1,
-            config.first_latency,
-            local_tcp_port,
-        );
-        let idle = Idle::new(Duration::from_secs(16), context.clone());
-        let channel_sender = ChannelSender::new(context.clone());
-
-        let register = Arc::new(registration_handler::Register::new(
-            self.server_cipher.clone(),
-            channel_sender.clone(),
-            config.server_address,
-            config.token.clone(),
-            config.device_id.clone(),
-            config.name.clone(),
-            config.password.is_some(),
-        ));
-        let device_list: Arc<Mutex<(u16, Vec<PeerDeviceInfo>)>> =
-            Arc::new(Mutex::new((response.epoch, response.device_info_list)));
-        let peer_nat_info_map: Arc<RwLock<HashMap<Ipv4Addr, NatInfo>>> =
-            Arc::new(RwLock::new(HashMap::with_capacity(16)));
-        let connect_status = Arc::new(AtomicCell::new(ConnectStatus::Connected));
-        let public_ip = response.public_ip;
-        let public_port = response.public_port;
-        let local_udp_port = context.main_local_udp_port().unwrap_or(0);
-        let local_ipv4 = crate::nat::local_ipv4();
-        let ipv6 = crate::nat::local_ipv6();
-
-        // NAT检测
-        let nat_test = NatTest::new(
-            config.stun_server.clone(),
-            public_ip,
-            public_port,
-            local_ipv4,
-            ipv6,
-            local_udp_port,
-            local_tcp_port,
-        );
-        let in_external_route = if config.in_ips.is_empty() {
-            None
-        } else {
-            Some(ExternalRoute::new(config.in_ips))
-        };
-        #[cfg(feature = "ip_proxy")]
-        let (tcp_proxy, udp_proxy, ip_proxy_map) = if config.out_ips.is_empty() || config.no_proxy {
-            (None, None, None)
-        } else {
-            let (tcp_proxy, udp_proxy, ip_proxy_map) = crate::ip_proxy::init_proxy(
-                #[cfg(not(target_os = "android"))]
-                channel_sender.clone(),
-                #[cfg(not(target_os = "android"))]
-                current_device.clone(),
-                #[cfg(not(target_os = "android"))]
-                client_cipher.clone(),
-            )
-            .await?;
-            (Some(tcp_proxy), Some(udp_proxy), Some(ip_proxy_map))
-        };
-        let out_external_route = AllowExternalRoute::new(config.out_ips);
-
-        let igmp_server = if config.simulate_multicast {
-            Some(IgmpServer::new(device_writer.clone()))
-        } else {
-            None
-        };
-        #[cfg(any(target_os = "linux", target_os = "macos", target_os = "windows"))]
-        if config.tap {
-            tap_handler::start(
-                vnt_status_manager.worker("tap_handler"),
-                channel_sender.clone(),
-                device_reader,
-                device_writer.clone(),
-                igmp_server.clone(),
-                current_device.clone(),
-                in_external_route,
-                #[cfg(feature = "ip_proxy")]
-                ip_proxy_map.clone(),
-                client_cipher.clone(),
-                self.server_cipher.clone(),
-                config.parallel,
-            );
-        } else {
-            tun_handler::start(
-                vnt_status_manager.worker("tun_handler"),
-                channel_sender.clone(),
-                device_reader,
-                device_writer.clone(),
-                igmp_server.clone(),
-                current_device.clone(),
-                in_external_route,
-                #[cfg(feature = "ip_proxy")]
-                ip_proxy_map.clone(),
-                client_cipher.clone(),
-                self.server_cipher.clone(),
-                config.parallel,
-            );
-        }
-        #[cfg(any(target_os = "android"))]
-        tun_handler::start(
-            vnt_status_manager.worker("android tun_handler"),
-            channel_sender.clone(),
-            device_reader,
-            device_writer.clone(),
-            igmp_server.clone(),
-            current_device.clone(),
-            in_external_route,
-            #[cfg(feature = "ip_proxy")]
-            ip_proxy_map.clone(),
-            client_cipher.clone(),
-            self.server_cipher.clone(),
-            config.parallel,
-        );
-
-        //外部数据接收处理
-        let channel_recv_handler = ChannelDataHandler::new(
-            current_device.clone(),
-            device_list.clone(),
-            register.clone(),
-            nat_test.clone(),
-            igmp_server,
-            device_writer.clone(),
-            connect_status.clone(),
-            peer_nat_info_map.clone(),
-            #[cfg(feature = "ip_proxy")]
-            ip_proxy_map,
-            out_external_route,
-            cone_sender,
-            symmetric_sender,
-            client_cipher.clone(),
-            self.server_cipher.clone(),
-            self.rsa_cipher.clone(),
-            config.relay,
-            config.token.clone(),
-            14,
-        );
-        let punch = Punch::new(
-            context.clone(),
-            config.punch_model,
-            config.tcp,
-            channel_recv_handler.clone(),
-        );
-        {
-            let channel = Channel::new(context.clone(), channel_recv_handler, tcp_listener);
-            let channel_worker = vnt_status_manager.worker("channel_worker");
-            let relay = config.relay;
-            tokio::spawn(
-                async move { channel.start(channel_worker, tcp_receiver, 65, relay).await },
-            );
-        }
-        {
-            let nat_test = nat_test.clone();
-            let device_list = device_list.clone();
-            let current_device = current_device.clone();
-            // 定时心跳
-            heartbeat_handler::start_heartbeat_main(
-                vnt_status_manager.worker("main-heartbeat"),
-                channel_sender.clone(),
-                device_list.clone(),
-                current_device.clone(),
-                config.server_address_str,
-                client_cipher.clone(),
-                self.server_cipher.clone(),
-            );
-            heartbeat_handler::start_heartbeat(
-                vnt_status_manager.worker("heartbeat"),
-                channel_sender.clone(),
-                device_list.clone(),
-                current_device.clone(),
-                client_cipher.clone(),
-                self.server_cipher.clone(),
-            );
-            // 空闲检查
-            heartbeat_handler::start_idle(
-                vnt_status_manager.worker("idle"),
-                idle,
-                channel_sender.clone(),
-            );
-            if !config.relay {
-                // 打洞处理
-                punch_handler::start(
-                    vnt_status_manager.worker("cone_receiver"),
-                    cone_receiver,
-                    punch.clone(),
-                    current_device.clone(),
-                    client_cipher.clone(),
-                );
-                punch_handler::start(
-                    vnt_status_manager.worker("symmetric_receiver"),
-                    symmetric_receiver,
-                    punch,
-                    current_device.clone(),
-                    client_cipher.clone(),
-                );
-                tokio::spawn(punch_handler::start_punch(
-                    vnt_status_manager.worker("punch_handler"),
-                    nat_test,
-                    device_list,
-                    channel_sender,
-                    current_device,
-                    client_cipher.clone(),
-                ));
-            }
-        }
-        #[cfg(feature = "ip_proxy")]
-        {
-            //代理
-            if let Some(tcp_proxy) = tcp_proxy {
-                tokio::spawn(tcp_proxy.start());
-            }
-            if let Some(udp_proxy) = udp_proxy {
-                tokio::spawn(udp_proxy.start());
-            }
-        }
-        {
-            let context = context.clone();
-            let nat_test = nat_test.clone();
-            tokio::spawn(async move {
-                let info = nat_test
-                    .re_test(
-                        public_ip,
-                        public_port,
-                        local_ipv4,
-                        ipv6,
-                        local_udp_port,
-                        local_tcp_port,
-                    )
-                    .await;
-                context.switch(info.nat_type);
-            });
-        }
-        Ok(Vnt {
-            config: self.config,
-            current_device,
-            context,
-            vnt_status_manager,
-            device_writer,
-            nat_test,
-            device_list,
-            connect_status,
-            peer_nat_info_map,
-        })
-    }
-}
-
-impl Vnt {
-    pub fn name(&self) -> &str {
-        &self.config.name
-    }
-    pub fn server_encrypt(&self) -> bool {
-        self.config.server_encrypt
-    }
-    pub fn client_encrypt(&self) -> bool {
-        self.config.password.is_some()
-    }
-    pub fn current_device(&self) -> CurrentDeviceInfo {
-        self.current_device.load()
-    }
-    pub fn peer_nat_info(&self, ip: &Ipv4Addr) -> Option<NatInfo> {
-        self.peer_nat_info_map.read().get(ip).cloned()
-    }
-    pub fn connection_status(&self) -> ConnectStatus {
-        self.connect_status.load()
-    }
-    pub fn nat_info(&self) -> NatInfo {
-        self.nat_test.nat_info()
-    }
-    pub fn device_list(&self) -> Vec<PeerDeviceInfo> {
-        let device_list_lock = self.device_list.lock();
-        let (_epoch, device_list) = device_list_lock.clone();
-        drop(device_list_lock);
-        device_list
-    }
-    pub fn route(&self, ip: &Ipv4Addr) -> Option<Route> {
-        self.context.route_one(ip)
-    }
-    pub fn route_key(&self, route_key: &RouteKey) -> Option<Ipv4Addr> {
-        self.context.route_to_id(route_key)
-    }
-    pub fn route_table(&self) -> Vec<(Ipv4Addr, Route)> {
-        self.context.route_table_one()
-    }
-    pub fn stop(&self) -> io::Result<()> {
-        let _ = self.context.close();
-        self.vnt_status_manager.stop_all();
-        let _ = self.device_writer.close();
-        let virtual_gateway = self.current_device.load().virtual_gateway;
-        let _ = UdpSocket::bind("0.0.0.0:0")?.send_to(
-            b"stop",
-            SocketAddr::V4(SocketAddrV4::new(virtual_gateway, 10000)),
-        );
-
-        Ok(())
-    }
-    pub async fn wait_stop(&mut self) {
-        self.vnt_status_manager.wait().await;
-        let _ = self.stop();
-    }
-    pub async fn wait_stop_ms(&mut self, ms: Duration) -> bool {
-        tokio::select! {
-            _=self.vnt_status_manager.wait()=>{
-                let _ = self.stop();
-                return true;
-            }
-            _=tokio::time::sleep(ms)=>{
-                return false;
-            }
-        }
-    }
-}
-
-impl Drop for Vnt {
-    fn drop(&mut self) {
-        let _ = self.stop();
-    }
-}
 
 #[derive(Clone, Debug)]
 pub struct Config {
+    #[cfg(any(target_os = "windows", target_os = "linux"))]
     pub tap: bool,
     pub token: String,
     pub device_id: String,
@@ -574,8 +21,7 @@ pub struct Config {
     pub in_ips: Vec<(u32, u32, Ipv4Addr)>,
     pub out_ips: Vec<(u32, u32)>,
     pub password: Option<String>,
-    pub simulate_multicast: bool,
+    pub mtu: Option<u32>,
-    pub mtu: Option<u16>,
     pub tcp: bool,
     pub ip: Option<Ipv4Addr>,
     pub relay: bool,
@@ -586,13 +32,17 @@ pub struct Config {
     pub cipher_model: CipherModel,
     pub finger: bool,
     pub punch_model: PunchModel,
-    pub port: u16,
+    pub ports: Option<Vec<u16>>,
     pub first_latency: bool,
+    #[cfg(not(target_os = "android"))]
+    pub device_name: Option<String>,
+    #[cfg(target_os = "android")]
+    pub device_fd: i32,
 }
 
 impl Config {
     pub fn new(
-        tap: bool,
+        #[cfg(any(target_os = "windows", target_os = "linux"))] tap: bool,
         token: String,
         device_id: String,
         name: String,
@@ -602,8 +52,7 @@ impl Config {
         in_ips: Vec<(u32, u32, Ipv4Addr)>,
         out_ips: Vec<(u32, u32)>,
         password: Option<String>,
-        simulate_multicast: bool,
+        mtu: Option<u32>,
-        mtu: Option<u16>,
         tcp: bool,
         ip: Option<Ipv4Addr>,
         relay: bool,
@@ -613,24 +62,27 @@ impl Config {
         cipher_model: CipherModel,
         finger: bool,
         punch_model: PunchModel,
-        port: u16,
+        ports: Option<Vec<u16>>,
         first_latency: bool,
-    ) -> Result<Self, Error> {
+        #[cfg(not(target_os = "android"))] device_name: Option<String>,
+        #[cfg(target_os = "android")] device_fd: i32,
+    ) -> io::Result<Self> {
         for x in stun_server.iter_mut() {
             if !x.contains(":") {
                 x.push_str(":3478");
             }
         }
         if token.is_empty() || token.len() > 128 {
-            return Err(Error::Stop(String::from("token too long")));
+            return Err(io::Error::new(io::ErrorKind::Other, "token too long"));
         }
         if device_id.is_empty() || device_id.len() > 128 {
-            return Err(Error::Stop(String::from("device_id too long")));
+            return Err(io::Error::new(io::ErrorKind::Other, "device_id too long"));
         }
         if name.is_empty() || name.len() > 128 {
-            return Err(Error::Stop(String::from("name too long")));
+            return Err(io::Error::new(io::ErrorKind::Other, "name too long"));
         }
         Ok(Self {
+            #[cfg(any(target_os = "windows", target_os = "linux"))]
             tap,
             token,
             device_id,
@@ -641,7 +93,6 @@ impl Config {
             in_ips,
             out_ips,
             password,
-            simulate_multicast,
             mtu,
             tcp,
             ip,
@@ -653,8 +104,12 @@ impl Config {
             cipher_model,
             finger,
             punch_model,
-            port,
+            ports,
             first_latency,
+            #[cfg(not(target_os = "android"))]
+            device_name,
+            #[cfg(target_os = "android")]
+            device_fd,
         })
     }
 }

vnt/src/core/status.rs

@@ -1,92 +0,0 @@
-use crate::util::wait::WaitGroup;
-use std::sync::Arc;
-use tokio::sync::watch;
-use tokio::sync::watch::{Receiver, Sender};
-
-#[derive(Copy, Clone, Eq, PartialEq)]
-pub enum VntStatus {
-    Starting,
-    Stopping,
-}
-
-pub struct VntWorker {
-    name: String,
-    wg: WaitGroup,
-    status_s: Arc<Sender<VntStatus>>,
-    status_r: Receiver<VntStatus>,
-}
-
-impl VntWorker {
-    pub fn worker(&self, name: &str) -> Self {
-        self.wg.add();
-        VntWorker {
-            name: name.to_string(),
-            wg: self.wg.clone(),
-            status_s: self.status_s.clone(),
-            status_r: self.status_r.clone(),
-        }
-    }
-}
-
-impl Drop for VntWorker {
-    fn drop(&mut self) {
-        log::info!("任务停止:{}", self.name);
-        self.wg.done();
-    }
-}
-
-impl VntWorker {
-    pub fn stop_all(&self) {
-        let _ = self.status_s.send(VntStatus::Stopping);
-    }
-    pub async fn stop_wait(&mut self) {
-        loop {
-            if *self.status_r.borrow() == VntStatus::Stopping {
-                return;
-            }
-            match self.status_r.changed().await {
-                Ok(_) => {
-                    if *self.status_r.borrow() == VntStatus::Stopping {
-                        return;
-                    }
-                }
-                Err(_) => {
-                    return;
-                }
-            }
-        }
-    }
-}
-
-#[derive(Clone)]
-pub struct VntStatusManger {
-    wg: WaitGroup,
-    status_s: Arc<Sender<VntStatus>>,
-    status_r: Receiver<VntStatus>,
-}
-
-impl VntStatusManger {
-    pub fn new() -> Self {
-        let (status_s, status_r) = watch::channel(VntStatus::Starting);
-        Self {
-            wg: WaitGroup::new(),
-            status_s: Arc::new(status_s),
-            status_r,
-        }
-    }
-    pub fn stop_all(&self) {
-        let _ = self.status_s.send(VntStatus::Stopping);
-    }
-    pub async fn wait(&mut self) {
-        self.wg.wait().await
-    }
-    pub fn worker(&self, name: &str) -> VntWorker {
-        self.wg.add();
-        VntWorker {
-            name: name.to_string(),
-            wg: self.wg.clone(),
-            status_s: self.status_s.clone(),
-            status_r: self.status_r.clone(),
-        }
-    }
-}

vnt/src/core/sync.rs

@@ -1,81 +0,0 @@
-use crate::cipher::RsaCipher;
-use crate::core::{Config, Vnt, VntUtil};
-use crate::handle::handshake_handler::HandshakeEnum;
-use crate::handle::registration_handler::{RegResponse, ReqEnum};
-use std::io;
-use std::ops::Deref;
-use std::time::Duration;
-use tokio::runtime::Runtime;
-
-pub struct VntUtilSync {
-    vnt_util: VntUtil,
-    runtime: Runtime,
-}
-
-pub struct VntSync {
-    vnt: Vnt,
-    runtime: Runtime,
-}
-
-impl VntUtilSync {
-    pub fn new(config: Config) -> io::Result<VntUtilSync> {
-        let runtime = tokio::runtime::Builder::new_multi_thread()
-            .enable_all()
-            .build()?;
-        let vnt_util = VntUtil::new(config)?;
-        Ok(VntUtilSync { vnt_util, runtime })
-    }
-    pub fn connect(&mut self) -> io::Result<()> {
-        self.vnt_util.connect()
-    }
-    pub fn handshake(&mut self) -> Result<Option<RsaCipher>, HandshakeEnum> {
-        self.vnt_util.handshake()
-    }
-    pub fn secret_handshake(&mut self) -> Result<(), HandshakeEnum> {
-        self.vnt_util.secret_handshake()
-    }
-    pub fn register(&mut self) -> Result<RegResponse, ReqEnum> {
-        self.vnt_util.register()
-    }
-    #[cfg(any(target_os = "android"))]
-    pub fn create_iface(&mut self, vpn_fd: i32) {
-        self.vnt_util.create_iface(vpn_fd)
-    }
-    #[cfg(any(target_os = "linux", target_os = "macos", target_os = "windows"))]
-    pub fn create_iface(&mut self) -> io::Result<crate::tun_tap_device::DriverInfo> {
-        self.vnt_util.create_iface()
-    }
-    pub fn build(self) -> crate::Result<VntSync> {
-        let runtime = self.runtime;
-        let vnt = runtime.block_on(self.vnt_util.build())?;
-        {
-            let mut vnt = vnt.clone();
-            std::thread::spawn(move || runtime.block_on(vnt.wait_stop()));
-        }
-        Ok(VntSync {
-            vnt,
-            runtime: tokio::runtime::Builder::new_current_thread()
-                .enable_all()
-                .build()
-                .unwrap(),
-        })
-    }
-}
-
-impl VntSync {
-    pub fn wait_stop(&mut self) {
-        self.runtime.block_on(self.vnt.wait_stop())
-    }
-    pub fn wait_stop_ms(&mut self, ms: u64) -> bool {
-        self.runtime
-            .block_on(self.vnt.wait_stop_ms(Duration::from_millis(ms)))
-    }
-}
-
-impl Deref for VntSync {
-    type Target = Vnt;
-
-    fn deref(&self) -> &Self::Target {
-        &self.vnt
-    }
-}