使用tokio改写处理逻辑

switch/Cargo.toml

@@ -1,13 +1,12 @@
 [package]
 name = "switch"
-version = "1.0.2"
+version = "1.0.6"
 edition = "2021"
 
 # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
 
 [dependencies]
 packet = { path = "./packet" }
-p2p_channel = { path = "./p2p_channel" }
 bytes = "1.3.0"
 log = "0.4.17"
 libc = "0.2.137"
@@ -24,7 +23,7 @@ sha2 = { version = "0.10.6", features = ["oid"] }
 thiserror = "1.0.37"
 chrono = "0.4.23"
 #lazy_static = "1.4.0"
-#moka = "0.9.6"
+moka = "0.9.6"
 protobuf = "3.2.0"
 #local-ip-address = "0.4.9"
 socket2 ={ version = "0.5.2", features = ["all"] }

switch/proto/message.proto

@@ -4,6 +4,7 @@ message RegistrationRequest{
   string device_id = 2;
   string name = 3;
   bool is_fast = 4;
+  string version = 5;
 }
 
 message RegistrationResponse{

switch/src/core/mod.rs

@@ -1,28 +1,34 @@
-use std::io;
+use std::{io, thread};
 use std::net::{Ipv4Addr, SocketAddr};
 use std::sync::Arc;
 
 use crossbeam::atomic::AtomicCell;
 use crossbeam_skiplist::SkipMap;
 use parking_lot::Mutex;
+use tokio::net::UdpSocket;
+use tokio::sync::mpsc::channel;
 
-use p2p_channel::boot::Boot;
-use p2p_channel::channel::{Channel, Route, RouteKey};
-use p2p_channel::punch::NatInfo;
 
-use crate::{tap_device, tun_device};
+use crate::channel::channel::{Channel, Context};
+use crate::channel::idle::Idle;
+use crate::channel::punch::{NatInfo, Punch};
+use crate::channel::{Route, RouteKey};
+use crate::channel::sender::ChannelSender;
+
 use crate::external_route::ExternalRoute;
-use crate::handle::{ConnectStatus, CurrentDeviceInfo, CurrentDeviceInfoExt, heartbeat_handler, PeerDeviceInfo, punch_handler, recv_handler, registration_handler, tap_handler, tun_handler};
+use crate::handle::{ConnectStatus, CurrentDeviceInfo, heartbeat_handler, PeerDeviceInfo, punch_handler, registration_handler};
+use crate::handle::recv_handler::ChannelDataHandler;
+use crate::handle::tun_tap::{tap_handler, tun_handler};
+use crate::igmp_server::IgmpServer;
 use crate::nat::NatTest;
-use crate::tap_device::TapWriter;
-use crate::tun_device::TunWriter;
+use crate::tun_tap_device;
+use crate::tun_tap_device::DeviceWriter;
 
 pub struct Switch {
     name: String,
     current_device: Arc<AtomicCell<CurrentDeviceInfo>>,
-    tun_writer: Option<TunWriter>,
-    tap_writer: Option<TapWriter>,
-    nat_channel: Channel<Ipv4Addr>,
+    context: Context,
+    device_writer: DeviceWriter,
     /// 0. 机器纪元，每一次上线或者下线都会增1，用于感知网络中机器变化
     /// 服务端和客户端的不一致，则服务端会推送新的设备列表
     /// 1. 网络中的虚拟ip列表
@@ -35,10 +41,16 @@ pub struct Switch {
 impl Switch {
     pub async fn start(config: Config) -> crate::Result<Switch> {
         log::info!("config:{:?}",config);
+        let main_channel = Arc::new(UdpSocket::bind("0.0.0.0:0").await?);
+        let response = registration_handler::registration(&main_channel, config.server_address, config.token.clone(), config.device_id.clone(), config.name.clone()).await?;
+        let (cone_sender, cone_receiver) = channel(3);
+        let (symmetric_sender, symmetric_receiver) = channel(2);
+        let context = Context::new(main_channel, 1);
+        let punch = Punch::new(context.clone());
+        let idle = Idle::new(16000, context.clone());
+        let channel_sender = ChannelSender::new(context.clone());
 
-        let (mut channel, punch, idle) = Boot::new::<Ipv4Addr>(80, 15000, 0)?;
-        let response = registration_handler::registration(&mut channel, config.server_address, config.token.clone(), config.device_id.clone(), config.name.clone())?;
-        let register = Arc::new(registration_handler::Register::new(channel.sender()?, config.server_address, config.token.clone(), config.device_id.clone(), config.name.clone()));
+        let register = Arc::new(registration_handler::Register::new(channel_sender.clone(), config.server_address, config.token.clone(), config.device_id.clone(), config.name.clone()));
         let device_list: Arc<Mutex<(u16, Vec<PeerDeviceInfo>)>> = Arc::new(Mutex::new((0, Vec::new())));
         let peer_nat_info_map: Arc<SkipMap<Ipv4Addr, NatInfo>> = Arc::new(SkipMap::new());
         let connect_status = Arc::new(AtomicCell::new(ConnectStatus::Connected));
@@ -47,7 +59,7 @@ impl Switch {
         let virtual_netmask = Ipv4Addr::from(response.virtual_netmask);
 
         let local_ip = crate::nat::local_ip()?;
-        let local_port = channel.local_addr()?.port();
+        let local_port = context.main_local_port()?;
         // NAT检测
         let nat_test = NatTest::new(config.nat_test_server.clone(), Ipv4Addr::from(response.public_ip), response.public_port as u16, local_ip, local_port);
         let in_ips = config.in_ips.iter().map(|(dest, mask, _)| { (Ipv4Addr::from(*dest), Ipv4Addr::from(*mask)) }).collect::<Vec<(Ipv4Addr, Ipv4Addr)>>();
@@ -56,51 +68,60 @@ impl Switch {
         let out_external_route = ExternalRoute::new(config.out_ips);
         let in_external_route = ExternalRoute::new(config.in_ips);
         let current_device = Arc::new(AtomicCell::new(CurrentDeviceInfo::new(virtual_ip, virtual_gateway, virtual_netmask, config.server_address)));
-        let ip_proxy_map = crate::ip_proxy::init_proxy(channel.sender()?,out_ips,current_device.clone()).await?;
-        let ( current_device_ext,tun_writer, tap_writer) = if config.tap {
+        let ip_proxy_map = crate::ip_proxy::init_proxy(channel_sender.clone(), out_ips, current_device.clone()).await?;
+        let (device_writer, igmp_server) = if config.tap {
             #[cfg(windows)]
             {
                 //删除switch的tun网卡避免ip冲突，因为非正常退出会保留网卡
-                tun_device::delete_tun();
+                tun_tap_device::delete_device(tun_tap_device::DeviceType::Tap);
             }
-            let (tap_writer, tap_reader, mac) = tap_device::create_tap(virtual_ip, virtual_netmask, virtual_gateway)?;
-            let current_device_ext = Arc::new(AtomicCell::new(CurrentDeviceInfoExt::new( mac)));
+            let (tap_writer, tap_reader) = tun_tap_device::create_device(tun_tap_device::DeviceType::Tap, virtual_ip, virtual_netmask, virtual_gateway, in_ips)?;
+            let igmp_server = IgmpServer::new(tap_writer.clone());
             //tap数据处理
-            tap_handler::start(channel.sender()?, tap_reader.clone(), tap_writer.clone(), current_device.clone(), in_external_route, ip_proxy_map.clone());
-            (current_device_ext, None, Some(tap_writer))
+            tap_handler::start(channel_sender.clone(), tap_reader.clone(), tap_writer.clone(), igmp_server.clone(), current_device.clone(), in_external_route, ip_proxy_map.clone());
+            (tap_writer, igmp_server)
         } else {
             #[cfg(windows)]
             {
                 //删除switch的tap网卡避免ip冲突，非正常退出会保留网卡
-                tap_device::delete_tap();
+                tun_tap_device::delete_device(tun_tap_device::DeviceType::Tap);
             }
             // tun通道
-            let (tun_writer, tun_reader) = tun_device::create_tun(virtual_ip, virtual_netmask, virtual_gateway,in_ips)?;
-            let current_device_ext = Arc::new(AtomicCell::new(CurrentDeviceInfoExt::new( [0, 0, 0, 0, 0, 0])));
+            let (tun_writer, tun_reader) = tun_tap_device::create_device(tun_tap_device::DeviceType::Tun, virtual_ip, virtual_netmask, virtual_gateway, in_ips)?;
+            let igmp_server = IgmpServer::new(tun_writer.clone());
             //tun数据接收处理
-            tun_handler::start(channel.sender()?, tun_reader.clone(), tun_writer.clone(), current_device.clone(), in_external_route, ip_proxy_map.clone());
-            (current_device_ext,Some(tun_writer), None)
+            tun_handler::start(channel_sender.clone(), tun_reader.clone(), tun_writer.clone(), igmp_server.clone(), current_device.clone(), in_external_route, ip_proxy_map.clone());
+            (tun_writer, igmp_server)
         };
         //外部数据接收处理
-        let channel_recv_handler = recv_handler::RecvHandler::new(channel.try_clone()?, current_device.clone(), current_device_ext,device_list.clone(), register.clone(),
-                                                                  nat_test.clone(), tun_writer.clone(), tap_writer.clone(),
-                                                                  connect_status.clone(), peer_nat_info_map.clone(), ip_proxy_map, out_external_route);
-        recv_handler::start(channel_recv_handler);
+        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(), ip_proxy_map, out_external_route,
+                                                           cone_sender, symmetric_sender);
+        let channel = Channel::new(context.clone(), channel_recv_handler);
+        thread::spawn(move || {
+            tokio::runtime::Builder::new_multi_thread()
+                .enable_all()
+                .build().unwrap()
+                .block_on(channel.start(14, 60));
+        });
+        context.switch(nat_test.nat_info().nat_type);
         // 定时心跳
-        heartbeat_handler::start_heartbeat(channel.sender()?, device_list.clone(), current_device.clone());
+        heartbeat_handler::start_heartbeat(channel_sender.clone(), device_list.clone(), current_device.clone()).await;
         // 空闲检查
-        heartbeat_handler::start_idle(idle, channel.sender()?);
+        heartbeat_handler::start_idle(idle, channel_sender.clone()).await;
         // 打洞处理
-        punch_handler::start_cone(punch.try_clone()?, current_device.clone());
-        punch_handler::start_symmetric(punch, current_device.clone());
-        punch_handler::start_punch(nat_test.clone(), device_list.clone(), channel.sender()?, current_device.clone());
+        punch_handler::start(cone_receiver, punch.clone(), current_device.clone()).await;
+        punch_handler::start(symmetric_receiver, punch, current_device.clone()).await;
+        punch_handler::start_punch(nat_test.clone(), device_list.clone(), channel_sender.clone(), current_device.clone()).await;
+
         log::info!("switch启动成功");
         Ok(Switch {
             name: config.name,
             current_device,
-            tun_writer,
-            tap_writer,
-            nat_channel: channel,
+            context,
+            device_writer,
             nat_test,
             device_list,
             connect_status,
@@ -132,22 +153,17 @@ impl Switch {
         device_list
     }
     pub fn route(&self, ip: &Ipv4Addr) -> Option<Route> {
-        self.nat_channel.route(ip)
+        self.context.route_one(ip)
     }
     pub fn route_key(&self, route_key: &RouteKey) -> Option<Ipv4Addr> {
-        self.nat_channel.route_to_id(route_key)
+        self.context.route_to_id(route_key)
     }
     pub fn route_table(&self) -> Vec<(Ipv4Addr, Route)> {
-        self.nat_channel.route_table()
+        self.context.route_table_one()
     }
     pub fn stop(&self) -> io::Result<()> {
-        if let Some(tap) = &self.tap_writer {
-            tap.close()?;
-        }
-        if let Some(tun) = &self.tun_writer {
-            tun.close()?;
-        }
-        self.nat_channel.close()?;
+        self.context.close();
+        self.device_writer.close()?;
         Ok(())
     }
 }

switch/src/error/mod.rs

@@ -5,8 +5,6 @@ use thiserror::Error;
 
 #[derive(Error, Debug)]
 pub enum Error {
-    #[error("packet error")]
-    PacketError(#[from] packet::error::Error),
     #[error("Io error")]
     Io(#[from] io::Error),
     #[error("Channel error")]

switch/src/external_route/mod.rs

@@ -2,6 +2,7 @@ use std::net::Ipv4Addr;
 
 // 目标ip，子网掩码，网关
 
+#[derive(Clone)]
 pub struct ExternalRoute {
     route_table: Vec<(u32, u32, Ipv4Addr)>,
 }

switch/src/handle/heartbeat_handler.rs

@@ -1,60 +1,54 @@
 use std::net::Ipv4Addr;
 use std::sync::Arc;
 use std::time::Duration;
-use std::{io, thread};
+use std::io;
 
 use chrono::Local;
 use crossbeam::atomic::AtomicCell;
 use parking_lot::Mutex;
 use rand::prelude::SliceRandom;
+use crate::channel::idle::Idle;
+use crate::channel::Route;
+use crate::channel::sender::ChannelSender;
 
-use p2p_channel::channel::sender::Sender;
-use p2p_channel::channel::Route;
-use p2p_channel::idle::Idle;
 
 use crate::handle::{CurrentDeviceInfo, PeerDeviceInfo};
 use crate::protocol::control_packet::PingPacket;
-use crate::protocol::{control_packet, NetPacket, Protocol, Version, MAX_TTL};
+use crate::protocol::{control_packet, NetPacket, Protocol, Version};
 
-pub fn start_idle(idle: Idle<Ipv4Addr>, sender: Sender<Ipv4Addr>) {
-    thread::Builder::new()
-        .name("idle".into())
-        .spawn(move || {
-            if let Err(e) = start_idle_(idle, sender) {
-                log::info!("空闲检测线程停止:{:?}", e);
+pub async fn start_idle(idle: Idle, sender: ChannelSender) {
+    tokio::spawn(async move {
+        match start_idle_(idle, sender).await {
+            Ok(_) => {}
+            Err(e) => {
+                log::warn!("空闲检测任务停止:{:?}", e);
             }
-        })
-        .unwrap();
+        }
+    });
 }
 
-fn start_idle_(idle: Idle<Ipv4Addr>, sender: Sender<Ipv4Addr>) -> io::Result<()> {
+async fn start_idle_(idle: Idle, sender: ChannelSender) -> io::Result<()> {
     loop {
-        let (idle_status, peer_ips, route) = idle.next_idle()?;
-        log::warn!(
-            "peer_ip:{:?},route:{:?},idle_status:{:?}",
-            peer_ips,
-            route,
-            idle_status
+        let (peer_ip, route) = idle.next_idle().await?;
+        log::info!(
+            "peer_ip:{:?},route:{:?}",
+            peer_ip,
+            route
         );
-        for peer_ip in peer_ips {
-            sender.remove_route(&peer_ip);
-        }
+        sender.remove_route(&peer_ip, route);
     }
 }
 
-pub fn start_heartbeat(
-    sender: Sender<Ipv4Addr>,
+pub async fn start_heartbeat(
+    sender: ChannelSender,
     device_list: Arc<Mutex<(u16, Vec<PeerDeviceInfo>)>>,
     current_device: Arc<AtomicCell<CurrentDeviceInfo>>,
 ) {
-    thread::Builder::new()
-        .name("heartbeat".into())
-        .spawn(move || {
-            if let Err(e) = start_heartbeat_(sender, device_list, current_device) {
-                log::info!("空闲检测线程停止:{:?}", e);
-            }
-        })
-        .unwrap();
+    tokio::spawn(async move {
+        if let Err(e) = start_heartbeat_(sender, device_list, current_device).await {
+            log::warn!("心跳任务停止:{:?}", e);
+        }
+    });
 }
 
 fn set_now_time(packet: &mut NetPacket<[u8; 16]>) -> io::Result<()> {
@@ -64,8 +58,8 @@ fn set_now_time(packet: &mut NetPacket<[u8; 16]>) -> io::Result<()> {
     Ok(())
 }
 
-fn start_heartbeat_(
-    sender: Sender<Ipv4Addr>,
+async fn start_heartbeat_(
+    sender: ChannelSender,
     device_list: Arc<Mutex<(u16, Vec<PeerDeviceInfo>)>>,
     current_device: Arc<AtomicCell<CurrentDeviceInfo>>,
 ) -> io::Result<()> {
@@ -73,7 +67,8 @@ fn start_heartbeat_(
     net_packet.set_version(Version::V1);
     net_packet.set_protocol(Protocol::Control);
     net_packet.set_transport_protocol(control_packet::Protocol::Ping.into());
-    net_packet.first_set_ttl(MAX_TTL);
+    //只寻找两跳以内能到的目标
+    net_packet.first_set_ttl(2);
     let mut count = 0;
     loop {
         let current_device = current_device.load();
@@ -83,19 +78,28 @@ fn start_heartbeat_(
             let epoch = { device_list.lock().0 };
             ping.set_epoch(epoch);
         }
+        set_now_time(&mut net_packet)?;
+        net_packet.set_destination(current_device.virtual_gateway());
+        if let Err(e) = sender.send_main(net_packet.buffer(), current_device.connect_server).await
+        {
+            log::warn!(
+                    "connect_server:{:?},e:{:?}",
+                    current_device.connect_server,
+                    e
+                );
+        }
         if count < 7 || count % 7 == 0 {
-            let mut route_list: Option<Vec<(Ipv4Addr, Route)>> = None;
-            let peer_list = device_list.lock().1.clone();
+            let mut route_list: Option<Vec<(Ipv4Addr, Vec<Route>)>> = None;
+            let peer_list = {device_list.lock().1.clone()};
             for peer in peer_list {
                 set_now_time(&mut net_packet)?;
-                net_packet.first_set_ttl(2);
                 net_packet.set_destination(peer.virtual_ip);
                 if sender
-                    .send_to_id(net_packet.buffer(), &peer.virtual_ip)
+                    .send_by_id(net_packet.buffer(), &peer.virtual_ip).await
                     .is_err()
                 {
                     //没有路由则发送到网关
-                    let _ = sender.send_to_addr(net_packet.buffer(), current_device.connect_server);
+                    let _ = sender.try_send_main(net_packet.buffer(), current_device.connect_server);
                     //再随机发送到其他地址，看有没有客户端符合转发条件
                     let route_list = route_list.get_or_insert_with(|| {
                         let mut l = sender.route_table();
@@ -103,43 +107,37 @@ fn start_heartbeat_(
                         l
                     });
                     let mut num = 0;
-                    //只寻找两跳以内能到的目标
-                    net_packet.first_set_ttl(2);
-                    for (peer_ip, route) in route_list.iter() {
-                        if peer_ip != &peer.virtual_ip && route.metric == 1 {
-                            set_now_time(&mut net_packet)?;
-                            let _ = sender.send_to_route(net_packet.buffer(), &route.route_key());
-                            num += 1;
-                        }
-                        if num >= 3 {
-                            break;
+                    'a: for (peer_ip, route_list) in route_list.iter() {
+                        for route in route_list {
+                            if peer_ip != &peer.virtual_ip && route.metric == 1 {
+                                set_now_time(&mut net_packet)?;
+                                let _ = sender.try_send_by_key(net_packet.buffer(), &route.route_key());
+                                num += 1;
+                                break;
+                            }
+                            if num >= 3 {
+                                break 'a;
+                            }
                         }
                     }
                 }
-                thread::sleep(Duration::from_millis(1));
-            }
-            set_now_time(&mut net_packet)?;
-            net_packet.set_destination(current_device.virtual_gateway());
-            if let Err(e) = sender.send_to_addr(net_packet.buffer(), current_device.connect_server)
-            {
-                log::warn!(
-                    "connect_server:{:?},e:{:?}",
-                    current_device.connect_server,
-                    e
-                );
+                tokio::time::sleep(Duration::from_millis(1)).await;
             }
+
         } else {
-            for (peer_ip, route) in sender.route_table().iter() {
+            for (peer_ip, route_list) in sender.route_table().iter() {
                 set_now_time(&mut net_packet)?;
                 net_packet.set_destination(*peer_ip);
-                if let Err(e) = sender.send_to_route(net_packet.buffer(), &route.route_key()) {
-                    log::warn!("peer_ip:{:?},route:{:?},e:{:?}", peer_ip, route, e);
+                for route in route_list {
+                    if let Err(e) = sender.send_by_key(net_packet.buffer(), &route.route_key()).await {
+                        log::warn!("peer_ip:{:?},route:{:?},e:{:?}", peer_ip, route, e);
+                    }
+                    tokio::time::sleep(Duration::from_millis(2)).await;
                 }
-                thread::sleep(Duration::from_millis(1));
             }
         }
 
         count += 1;
-        thread::sleep(Duration::from_millis(5000));
+        tokio::time::sleep(Duration::from_millis(5000)).await;
     }
 }

switch/src/handle/mod.rs

@@ -1,11 +1,10 @@
 use std::net::{Ipv4Addr, SocketAddr};
 
 pub mod heartbeat_handler;
-pub mod tun_handler;
-pub mod tap_handler;
 pub mod punch_handler;
 pub mod recv_handler;
 pub mod registration_handler;
+pub mod tun_tap;
 
 /// 是否在一个网段
 fn check_dest(dest: Ipv4Addr, virtual_netmask: Ipv4Addr, virtual_network: Ipv4Addr) -> bool {
@@ -60,19 +59,6 @@ pub enum ConnectStatus {
     Connected,
 }
 
-#[derive(Copy, Clone, Debug, Eq, PartialEq)]
-pub struct CurrentDeviceInfoExt {
-    pub mac: [u8; 6],
-}
-
-impl CurrentDeviceInfoExt {
-    pub fn new(mac: [u8; 6]) -> Self {
-        Self {
-            mac,
-        }
-    }
-}
-
 #[derive(Copy, Clone, Debug, Eq, PartialEq)]
 pub struct CurrentDeviceInfo {
     virtual_ip: Ipv4Addr,

switch/src/handle/punch_handler.rs

@@ -1,93 +1,72 @@
 use crate::handle::{CurrentDeviceInfo, PeerDeviceInfo};
 use crate::nat::NatTest;
 use crate::proto::message::{PunchInfo, PunchNatType};
-use crate::protocol::{control_packet, turn_packet, NetPacket, Protocol, Version, MAX_TTL};
+use crate::protocol::{control_packet, other_turn_packet, NetPacket, Protocol, Version, MAX_TTL};
 use crossbeam::atomic::AtomicCell;
-use p2p_channel::channel::sender::Sender;
-use p2p_channel::punch::{NatInfo, NatType, Punch};
 use parking_lot::Mutex;
 use protobuf::Message;
 use rand::prelude::SliceRandom;
-use std::net::{IpAddr, Ipv4Addr};
+use std::net::Ipv4Addr;
 use std::sync::Arc;
 use std::time::Duration;
-use std::{io, thread};
+use std::io;
+use tokio::sync::mpsc::Receiver;
+use crate::channel::punch::{NatInfo, Punch};
+use crate::channel::sender::ChannelSender;
 
-pub fn start_cone(punch: Punch<Ipv4Addr>, current_device: Arc<AtomicCell<CurrentDeviceInfo>>) {
-    thread::Builder::new()
-        .name("punch-cone".into())
-        .spawn(move || {
-            if let Err(e) = start_(true, punch, current_device) {
-                log::warn!("锥形网络打洞处理线程停止 {:?}", e);
-            }
-        })
-        .unwrap();
+pub async fn start(receiver: Receiver<(Ipv4Addr, NatInfo)>, punch: Punch, current_device: Arc<AtomicCell<CurrentDeviceInfo>>) {
+    tokio::spawn(async move {
+        start0(receiver, punch, current_device).await;
+    });
 }
 
-pub fn start_symmetric(punch: Punch<Ipv4Addr>, current_device: Arc<AtomicCell<CurrentDeviceInfo>>) {
-    thread::Builder::new()
-        .name("punch-symmetric".into())
-        .spawn(move || {
-            if let Err(e) = start_(false, punch, current_device) {
-                log::warn!("对称网络打洞处理线程停止 {:?}", e);
-            }
-        })
-        .unwrap();
+pub async fn start0(mut receiver: Receiver<(Ipv4Addr, NatInfo)>, mut punch: Punch, current_device: Arc<AtomicCell<CurrentDeviceInfo>>) {
+    while let Some((peer_ip, nat_info)) = receiver.recv().await {
+        if let Err(e) = start_(&mut punch, &current_device, peer_ip, nat_info).await {
+            log::warn!("网络打洞异常 {:?}", e);
+        }
+    }
 }
 
-fn start_(
-    is_cone: bool,
-    mut punch: Punch<Ipv4Addr>,
-    current_device: Arc<AtomicCell<CurrentDeviceInfo>>,
+async fn start_(
+    punch: &mut Punch,
+    current_device: &Arc<AtomicCell<CurrentDeviceInfo>>,
+    peer_ip: Ipv4Addr,
+    nat_info: NatInfo,
 ) -> io::Result<()> {
     let mut packet = NetPacket::new([0u8; 12])?;
     packet.set_version(Version::V1);
     packet.first_set_ttl(1);
     packet.set_protocol(Protocol::Control);
     packet.set_transport_protocol(control_packet::Protocol::PunchRequest.into());
-    loop {
-        let (peer_ip, nat_info) = if is_cone {
-            punch.next_cone(None)?
-        } else {
-            punch.next_symmetric(None)?
-        };
-        if let Some(route) = punch.sender().route(&peer_ip) {
-            if route.metric == 1 {
-                //直连地址不需要打洞
-                continue;
-            }
-        }
-        packet.set_source(current_device.load().virtual_ip());
-        packet.set_destination(peer_ip);
-        log::info!("发起打洞，目标:{:?},{:?}", peer_ip, nat_info);
-        if let Err(e) = punch.punch(packet.buffer(), peer_ip, nat_info) {
-            log::warn!("peer_ip:{:?},e:{:?}", peer_ip, e);
-        }
-    }
+    packet.set_source(current_device.load().virtual_ip());
+    packet.set_destination(peer_ip);
+    log::info!("发起打洞，目标:{:?},{:?}", peer_ip, nat_info);
+
+    punch.punch(packet.buffer(), peer_ip, nat_info).await
 }
 
-pub fn start_punch(
+pub async fn start_punch(
     nat_test: NatTest,
     device_list: Arc<Mutex<(u16, Vec<PeerDeviceInfo>)>>,
-    sender: Sender<Ipv4Addr>,
+    sender: ChannelSender,
     current_device: Arc<AtomicCell<CurrentDeviceInfo>>,
 ) {
-    thread::Builder::new()
-        .name("punch-send-request".into())
-        .spawn(move || {
-            if let Err(e) = start_punch_(nat_test, device_list, sender, current_device) {
-                log::warn!("对称网络打洞处理线程停止 {:?}", e);
-            }
-        })
-        .unwrap();
+    tokio::spawn(async move {
+        if let Err(e) = start_punch_(nat_test, device_list, sender, current_device).await {
+            log::warn!("打洞处理任务停止 {:?}", e);
+        }
+    });
 }
 
-fn start_punch_(
+async fn start_punch_(
     nat_test: NatTest,
     device_list: Arc<Mutex<(u16, Vec<PeerDeviceInfo>)>>,
-    sender: Sender<Ipv4Addr>,
+    sender: ChannelSender,
     current_device: Arc<AtomicCell<CurrentDeviceInfo>>,
 ) -> crate::Result<()> {
+    let mut num = 0;
+    let sleep_time = [3, 5, 7, 11, 13, 17, 19, 23, 29];
     loop {
         if sender.is_close() {
             return Ok(());
@@ -102,28 +81,19 @@ fn start_punch_(
                 if info.virtual_ip <= current_device.virtual_ip {
                     continue;
                 }
-                if let Some(route) = sender.route(&info.virtual_ip) {
-                    if route.metric == 1 {
-                        //直连地址不需要打洞
-                        continue;
-                    }
+                if !sender.need_punch(&info.virtual_ip) {
+                    continue;
                 }
                 count += 1;
-                if count > 3 {
+                if count > 2 {
                     break;
                 }
                 let buf = punch_packet(current_device.virtual_ip(), &nat_info, info.virtual_ip)?;
-                sender.send_to_addr(&buf, current_device.connect_server)?;
-            }
-        }
-        match nat_info.nat_type {
-            NatType::Symmetric => {
-                thread::sleep(Duration::from_secs(28));
-            }
-            NatType::Cone => {
-                thread::sleep(Duration::from_secs(20));
+                sender.send_main(&buf, current_device.connect_server).await?;
             }
         }
+        num += 1;
+        tokio::time::sleep(Duration::from_secs(sleep_time[num % sleep_time.len()])).await;
     }
 }
 
@@ -137,28 +107,18 @@ pub fn punch_packet(
     punch_reply.public_ip_list = nat_info
         .public_ips
         .iter()
-        .map(|i| match i {
-            IpAddr::V4(ip) => u32::from_be_bytes(ip.octets()),
-            IpAddr::V6(_) => {
-                panic!()
-            }
-        })
+        .map(|ip| u32::from_be_bytes(ip.octets()))
         .collect();
     punch_reply.public_port = nat_info.public_port as u32;
     punch_reply.public_port_range = nat_info.public_port_range as u32;
-    punch_reply.local_ip = match nat_info.local_ip {
-        IpAddr::V4(ip) => u32::from_be_bytes(ip.octets()),
-        IpAddr::V6(_) => {
-            panic!()
-        }
-    };
+    punch_reply.local_ip = u32::from_be_bytes(nat_info.local_ip.octets());
     punch_reply.local_port = nat_info.local_port as u32;
     punch_reply.nat_type = protobuf::EnumOrUnknown::new(PunchNatType::from(nat_info.nat_type));
     let bytes = punch_reply.write_to_bytes()?;
     let mut net_packet = NetPacket::new(vec![0u8; 12 + bytes.len()])?;
     net_packet.set_version(Version::V1);
     net_packet.set_protocol(Protocol::OtherTurn);
-    net_packet.set_transport_protocol(turn_packet::Protocol::Punch.into());
+    net_packet.set_transport_protocol(other_turn_packet::Protocol::Punch.into());
     net_packet.first_set_ttl(MAX_TTL);
     net_packet.set_source(virtual_ip);
     net_packet.set_destination(dest);

switch/src/handle/recv_handler.rs

@@ -1,5 +1,4 @@
-use std::thread;
-use std::net::{IpAddr, Ipv4Addr, SocketAddr, SocketAddrV4};
+use std::net::{Ipv4Addr, SocketAddr, SocketAddrV4};
 use std::sync::Arc;
 
 use chrono::Local;
@@ -7,128 +6,100 @@ use crossbeam::atomic::AtomicCell;
 use crossbeam_skiplist::SkipMap;
 use parking_lot::Mutex;
 use protobuf::Message;
+use tokio::net::UdpSocket;
+use tokio::sync::mpsc::Sender;
 
-use p2p_channel::channel::{Channel, Route, RouteKey};
-use p2p_channel::punch::NatInfo;
-use packet::ethernet;
 use packet::icmp::{icmp, Kind};
 use packet::icmp::icmp::HeaderOther;
 use packet::ip::ipv4;
 use packet::ip::ipv4::packet::IpV4Packet;
+use crate::channel::channel::Context;
+use crate::channel::punch::{NatInfo, NatType};
+use crate::channel::{Route, RouteKey};
 
 use crate::error::Error;
 use crate::external_route::ExternalRoute;
-use crate::handle::{check_dest, ConnectStatus, CurrentDeviceInfo, CurrentDeviceInfoExt, PeerDeviceInfo};
+use crate::handle::{check_dest, ConnectStatus, CurrentDeviceInfo, PeerDeviceInfo};
 use crate::handle::registration_handler::Register;
+use crate::igmp_server::IgmpServer;
 use crate::ip_proxy::IpProxyMap;
 use crate::nat;
 use crate::nat::NatTest;
 use crate::proto::message::{DeviceList, PunchInfo, PunchNatType, RegistrationResponse};
-use crate::protocol::{control_packet, MAX_TTL, NetPacket, Protocol, service_packet, turn_packet, Version};
+use crate::protocol::{control_packet, MAX_TTL, NetPacket, Protocol, service_packet, other_turn_packet, Version, ip_turn_packet};
 use crate::protocol::control_packet::ControlPacket;
 use crate::protocol::error_packet::InErrorPacket;
-use crate::tap_device::TapWriter;
-use crate::tun_device::TunWriter;
+use crate::tun_tap_device::DeviceWriter;
 
-pub fn start(mut handler: RecvHandler) {
-    thread::Builder::new().name("udp-recv-handler".into()).spawn(move || {
-        let mut buf = [0; 4096];
-        loop {
-            match handler.channel.recv_from(&mut buf, None) {
-                Ok((len, route)) => {
-                    if let Err(e) = handler.handle(&mut buf[..len], &route) {
-                        log::warn!("数据处理失败:{:?},e:{:?}",route,e);
-                        if let Error::Stop(_) = e {
-                            let _ = handler.channel.close();
-                            break;
-                        }
-                    }
-                }
-                Err(e) => {
-                    log::warn!("{:?}",e);
-                    // 检查关闭状态
-                    if handler.channel.is_close() {
-                        break;
-                    }
-                }
-            }
-        }
-    }).unwrap();
-}
-
-pub struct RecvHandler {
-    channel: Channel<Ipv4Addr>,
+#[derive(Clone)]
+pub struct ChannelDataHandler {
     current_device: Arc<AtomicCell<CurrentDeviceInfo>>,
-    current_device_ext: Arc<AtomicCell<CurrentDeviceInfoExt>>,
     device_list: Arc<Mutex<(u16, Vec<PeerDeviceInfo>)>>,
     register: Arc<Register>,
     nat_test: NatTest,
-    tun_writer: Option<TunWriter>,
-    tap_writer: Option<TapWriter>,
+    igmp_server: IgmpServer,
+    device_writer: DeviceWriter,
     connect_status: Arc<AtomicCell<ConnectStatus>>,
     peer_nat_info_map: Arc<SkipMap<Ipv4Addr, NatInfo>>,
     ip_proxy_map: IpProxyMap,
     out_external_route: ExternalRoute,
+    cone_sender: Sender<(Ipv4Addr, NatInfo)>,
+    symmetric_sender: Sender<(Ipv4Addr, NatInfo)>,
 }
 
-impl RecvHandler {
-    pub fn new(channel: Channel<Ipv4Addr>,
-               current_device: Arc<AtomicCell<CurrentDeviceInfo>>,
-               current_device_ext: Arc<AtomicCell<CurrentDeviceInfoExt>>,
+impl ChannelDataHandler {
+    pub fn new(current_device: Arc<AtomicCell<CurrentDeviceInfo>>,
                device_list: Arc<Mutex<(u16, Vec<PeerDeviceInfo>)>>,
                register: Arc<Register>,
                nat_test: NatTest,
-               tun_writer: Option<TunWriter>,
-               tap_writer: Option<TapWriter>,
+               igmp_server: IgmpServer,
+               device_writer: DeviceWriter,
                connect_status: Arc<AtomicCell<ConnectStatus>>,
                peer_nat_info_map: Arc<SkipMap<Ipv4Addr, NatInfo>>,
                ip_proxy_map: IpProxyMap,
                out_external_route: ExternalRoute,
-    ) -> Self {
+               cone_sender: Sender<(Ipv4Addr, NatInfo)>,
+               symmetric_sender: Sender<(Ipv4Addr, NatInfo)>, ) -> Self {
         Self {
-            channel,
             current_device,
-            current_device_ext,
             device_list,
             register,
             nat_test,
-            tun_writer,
-            tap_writer,
+            igmp_server,
+            device_writer,
             connect_status,
             peer_nat_info_map,
             ip_proxy_map,
             out_external_route,
+            cone_sender,
+            symmetric_sender,
         }
     }
-    // pub fn try_clone(&self) -> io::Result<Self> {
-    //     Ok(Self {
-    //         channel: self.channel.try_clone()?,
-    //         current_device: self.current_device.clone(),
-    //         device_list: self.device_list.clone(),
-    //         register: self.register.clone(),
-    //         nat_test: self.nat_test.clone(),
-    //         tun_writer: self.tun_writer.clone(),
-    //         tap_writer: self.tap_writer.clone(),
-    //         connect_status: self.connect_status.clone(),
-    //         peer_nat_info_map: self.peer_nat_info_map.clone(),
-    //     })
-    // }
 }
 
-impl RecvHandler {
-    fn handle(&self, buf: &mut [u8], route_key: &RouteKey) -> crate::Result<()> {
-        let mut net_packet = NetPacket::new(buf)?;
+
+impl ChannelDataHandler {
+    pub async fn handle(&mut self, buf: &mut [u8], start: usize, end: usize, route_key: RouteKey, _udp: &Arc<UdpSocket>,
+                        context: &Context, ) {
+        assert_eq!(start, 14);
+        match self.handle0(&mut buf[..end], &route_key, context).await {
+            Ok(_) => {}
+            Err(e) => {
+                log::error!("{:?}",e);
+            }
+        }
+    }
+    async fn handle0(&self, buf: &mut [u8], route_key: &RouteKey, context: &Context) -> crate::Result<()> {
+        let mut net_packet = NetPacket::new(&mut buf[14..])?;
         if net_packet.ttl() == 0 {
             return Ok(());
         }
         let source = net_packet.source();
         let current_device = self.current_device.load();
-        if source == current_device.virtual_ip() {
-            return Ok(());
-        }
         let destination = net_packet.destination();
-        if !destination.is_broadcast() && destination != current_device.broadcast_address
-            && current_device.virtual_ip() != destination && self.connect_status.load() == ConnectStatus::Connected {
+        if current_device.virtual_ip() != destination
+            && !destination.is_broadcast() && !destination.is_multicast() && destination != current_device.broadcast_address
+            && self.connect_status.load() == ConnectStatus::Connected {
             if !check_dest(source, current_device.virtual_netmask, current_device.virtual_network) {
                 log::warn!("转发数据，源地址错误:{:?},当前网络:{:?},route_key:{:?}",source,current_device.virtual_network,route_key);
                 return Ok(());
@@ -141,111 +112,118 @@ impl RecvHandler {
             let ttl = net_packet.ttl();
             if ttl > 0 {
                 // 转发
-                if let Some(route) = self.channel.route(&destination) {
+                if let Some(route) = context.route_one(&destination) {
                     if route.metric <= net_packet.ttl() {
-                        self.channel.send_to_route(net_packet.buffer(), &route.route_key())?;
+                        context.send_by_key(net_packet.buffer(), &route.route_key()).await?;
                     }
                 } else if (ttl > 1 || destination == current_device.virtual_gateway())
                     && source != current_device.virtual_gateway() {
                     //网关默认要转发一次，生存时间不够的发到网关也会被丢弃
-                    self.channel.send_to_addr(net_packet.buffer(), current_device.connect_server)?;
+                    context.send_main(net_packet.buffer(), current_device.connect_server).await?;
                 }
             }
             return Ok(());
         }
         match net_packet.protocol() {
-            Protocol::Ipv4Turn => {
-                let mut ipv4 = IpV4Packet::new(net_packet.payload_mut())?;
-                if ipv4.destination_ip() == destination && ipv4.protocol() == ipv4::protocol::Protocol::Icmp {
-                    let mut icmp_packet = icmp::IcmpPacket::new(ipv4.payload_mut())?;
-                    if icmp_packet.kind() == Kind::EchoRequest {
-                        //开启ping
-                        icmp_packet.set_kind(Kind::EchoReply);
-                        icmp_packet.update_checksum();
-                        ipv4.set_source_ip(destination);
-                        ipv4.set_destination_ip(source);
-                        ipv4.update_checksum();
-                        net_packet.set_source(destination);
-                        net_packet.set_destination(source);
-                        self.channel.send_to_route(net_packet.buffer(), route_key)?;
-                        return Ok(());
-                    }
-                }
-                if ipv4.destination_ip() != destination {
-                    if let Some(gate_way) = self.out_external_route.route(&ipv4.destination_ip()) {
+            Protocol::IpTurn => {
+                match ip_turn_packet::Protocol::from(net_packet.transport_protocol()) {
+                    ip_turn_packet::Protocol::Ipv4 => {
+                        let mut ipv4 = IpV4Packet::new(net_packet.payload_mut())?;
                         match ipv4.protocol() {
-                            ipv4::protocol::Protocol::Tcp => {
-                                let dest_ip = ipv4.destination_ip();
-                                //转发到代理目标地址
-                                let mut tcp_packet = packet::tcp::tcp::TcpPacket::new(source, destination, ipv4.payload_mut())?;
-                                let source_port = tcp_packet.source_port();
-                                let dest_port = tcp_packet.destination_port();
-                                tcp_packet.set_destination_port(self.ip_proxy_map.tcp_proxy_port);
-                                tcp_packet.update_checksum();
-                                ipv4.set_destination_ip(destination);
-                                ipv4.update_checksum();
-                                self.ip_proxy_map.tcp_proxy_map.insert(SocketAddrV4::new(source, source_port),
-                                                                       (SocketAddrV4::new(gate_way, 0), SocketAddrV4::new(dest_ip, dest_port)));
-                            }
-                            ipv4::protocol::Protocol::Udp => {
-                                let dest_ip = ipv4.destination_ip();
-                                //转发到代理目标地址
-                                let mut udp_packet = packet::udp::udp::UdpPacket::new(source, destination, ipv4.payload_mut())?;
-                                let source_port = udp_packet.source_port();
-                                let dest_port = udp_packet.destination_port();
-                                udp_packet.set_destination_port(self.ip_proxy_map.udp_proxy_port);
-                                udp_packet.update_checksum();
-                                ipv4.set_destination_ip(destination);
-                                ipv4.update_checksum();
-                                println!("{:?}",ipv4);
-                                self.ip_proxy_map.udp_proxy_map.insert(SocketAddrV4::new(source, source_port),
-                                                                       (SocketAddrV4::new(gate_way, 0), SocketAddrV4::new(dest_ip, dest_port)));
+                            ipv4::protocol::Protocol::Igmp => {
+                                self.igmp_server.handle(ipv4.payload(), source)?;
+                                return Ok(());
                             }
                             ipv4::protocol::Protocol::Icmp => {
-                                let dest_ip = ipv4.destination_ip();
-                                //转发到代理目标地址
-                                let icmp_packet = icmp::IcmpPacket::new(ipv4.payload())?;
-                                match icmp_packet.header_other() {
-                                    HeaderOther::Identifier(id, seq) => {
-                                        self.ip_proxy_map.icmp_proxy_map.insert((dest_ip, id, seq), source);
-                                        self.ip_proxy_map.send_icmp(ipv4.payload(), &gate_way, &dest_ip)?;
-                                    }
-                                    _ => {
+                                if ipv4.destination_ip() == destination {
+                                    let mut icmp_packet = icmp::IcmpPacket::new(ipv4.payload_mut())?;
+                                    if icmp_packet.kind() == Kind::EchoRequest {
+                                        //开启ping
+                                        icmp_packet.set_kind(Kind::EchoReply);
+                                        icmp_packet.update_checksum();
+                                        ipv4.set_source_ip(destination);
+                                        ipv4.set_destination_ip(source);
+                                        ipv4.update_checksum();
+                                        net_packet.set_source(destination);
+                                        net_packet.set_destination(source);
+                                        context.send_by_key(net_packet.buffer(), route_key).await?;
                                         return Ok(());
                                     }
                                 }
                             }
                             _ => {
-                                return Ok(());
+                                if ipv4.destination_ip() != destination {
+                                    if let Some(gate_way) = self.out_external_route.route(&ipv4.destination_ip()) {
+                                        match ipv4.protocol() {
+                                            ipv4::protocol::Protocol::Tcp => {
+                                                let dest_ip = ipv4.destination_ip();
+                                                //转发到代理目标地址
+                                                let mut tcp_packet = packet::tcp::tcp::TcpPacket::new(source, destination, ipv4.payload_mut())?;
+                                                let source_port = tcp_packet.source_port();
+                                                let dest_port = tcp_packet.destination_port();
+                                                tcp_packet.set_destination_port(self.ip_proxy_map.tcp_proxy_port);
+                                                tcp_packet.update_checksum();
+                                                ipv4.set_destination_ip(destination);
+                                                ipv4.update_checksum();
+                                                self.ip_proxy_map.tcp_proxy_map.insert(SocketAddrV4::new(source, source_port),
+                                                                                       (SocketAddrV4::new(gate_way, 0), SocketAddrV4::new(dest_ip, dest_port)));
+                                            }
+                                            ipv4::protocol::Protocol::Udp => {
+                                                let dest_ip = ipv4.destination_ip();
+                                                //转发到代理目标地址
+                                                let mut udp_packet = packet::udp::udp::UdpPacket::new(source, destination, ipv4.payload_mut())?;
+                                                let source_port = udp_packet.source_port();
+                                                let dest_port = udp_packet.destination_port();
+                                                udp_packet.set_destination_port(self.ip_proxy_map.udp_proxy_port);
+                                                udp_packet.update_checksum();
+                                                ipv4.set_destination_ip(destination);
+                                                ipv4.update_checksum();
+                                                self.ip_proxy_map.udp_proxy_map.insert(SocketAddrV4::new(source, source_port),
+                                                                                       (SocketAddrV4::new(gate_way, 0), SocketAddrV4::new(dest_ip, dest_port)));
+                                            }
+                                            ipv4::protocol::Protocol::Icmp => {
+                                                let dest_ip = ipv4.destination_ip();
+                                                //转发到代理目标地址
+                                                let icmp_packet = icmp::IcmpPacket::new(ipv4.payload())?;
+                                                match icmp_packet.header_other() {
+                                                    HeaderOther::Identifier(id, seq) => {
+                                                        self.ip_proxy_map.icmp_proxy_map.insert((dest_ip, id, seq), source);
+                                                        self.ip_proxy_map.send_icmp(ipv4.payload(), &gate_way, &dest_ip)?;
+                                                    }
+                                                    _ => {
+                                                        return Ok(());
+                                                    }
+                                                }
+                                            }
+                                            _ => {
+                                                return Ok(());
+                                            }
+                                        }
+                                    }
+                                }
                             }
                         }
+                        //传输协议12字节
+                        self.device_writer.write_ipv4(&mut buf[12..])?;
+                        return Ok(());
                     }
-                }
-                if let Some(tun_writer) = &self.tun_writer {
-                    tun_writer.write(net_packet.payload())?;
-                } else {
-                    if let Some(tap_writer) = &self.tap_writer {
-                        let mut ethernet_packet = ethernet::packet::EthernetPacket::unchecked(vec![0; 14 + ipv4.buffer.len()]);
-                        let source = source.octets();
-                        ethernet_packet.set_source(&[source[0], source[1], source[2], source[3], 123, 234]);
-                        ethernet_packet.set_destination(&self.current_device_ext.load().mac);
-                        ethernet_packet.set_protocol(ethernet::protocol::Protocol::Ipv4);
-                        ethernet_packet.payload_mut().copy_from_slice(ipv4.buffer);
-                        tap_writer.write(&ethernet_packet.buffer)?;
+                    ip_turn_packet::Protocol::Ipv4Broadcast => {
+                        //客户端不帮忙转发广播包，所以不会出现这种类型的数据
                     }
+                    ip_turn_packet::Protocol::Unknown(_) => {}
                 }
             }
             Protocol::Service => {
-                self.service(current_device, source, net_packet, route_key)?;
+                self.service(context, current_device, source, net_packet, route_key).await?;
             }
             Protocol::Error => {
-                self.error(current_device, source, net_packet, route_key)?;
+                self.error(context, current_device, source, net_packet, route_key).await?;
             }
             Protocol::Control => {
-                self.control(current_device, source, net_packet, route_key)?;
+                self.control(context, current_device, source, net_packet, route_key).await?;
             }
             Protocol::OtherTurn => {
-                self.other_turn(current_device, source, net_packet, route_key)?;
+                self.other_turn(context, current_device, source, net_packet, route_key).await?;
             }
             Protocol::UnKnow(e) => {
                 log::info!("不支持的协议:{}",e);
@@ -253,18 +231,20 @@ impl RecvHandler {
         }
         Ok(())
     }
-    fn service(&self, current_device: CurrentDeviceInfo, source: Ipv4Addr, net_packet: NetPacket<&mut [u8]>, route_key: &RouteKey) -> crate::Result<()> {
-        if route_key.addr != current_device.connect_server || source != current_device.virtual_gateway() {
+    async fn service(&self, context: &Context, current_device: CurrentDeviceInfo, _source: Ipv4Addr, net_packet: NetPacket<&mut [u8]>, route_key: &RouteKey) -> crate::Result<()> {
+        //todo 校验来源
+        if route_key.addr != current_device.connect_server {
             return Ok(());
         }
         match service_packet::Protocol::from(net_packet.transport_protocol()) {
             service_packet::Protocol::RegistrationRequest => {}
             service_packet::Protocol::RegistrationResponse => {
                 let response = RegistrationResponse::parse_from_bytes(net_packet.payload())?;
-                let local_port = self.channel.local_addr()?.port();
+                let local_port = context.main_local_port()?;
                 let local_ip = nat::local_ip()?;
-                let nat_info = self.nat_test.re_test(Ipv4Addr::from(response.public_ip), response.public_port as u16, local_ip, local_port);
-                self.channel.set_nat_type(nat_info.nat_type)?;
+                let nat_info = self.nat_test.re_test(Ipv4Addr::from(response.public_ip),
+                                                     response.public_port as u16, local_ip, local_port);
+                context.switch(nat_info.nat_type);
                 let new_ip = Ipv4Addr::from(response.virtual_ip);
                 let current_ip = current_device.virtual_ip();
                 if current_ip != new_ip {
@@ -275,13 +255,7 @@ impl RecvHandler {
                     let virtual_ip = Ipv4Addr::from(response.virtual_ip);
                     let virtual_gateway = Ipv4Addr::from(response.virtual_gateway);
                     let virtual_netmask = Ipv4Addr::from(response.virtual_netmask);
-                    if let Some(tun_writer) = &self.tun_writer {
-                        tun_writer.change_ip(virtual_ip, virtual_netmask, virtual_gateway, old_netmask, old_gateway)?;
-                    } else {
-                        if let Some(tap_writer) = &self.tap_writer {
-                            tap_writer.change_ip(virtual_ip, virtual_netmask, virtual_gateway, old_netmask, old_gateway)?;
-                        }
-                    }
+                    self.device_writer.change_ip(virtual_ip, virtual_netmask, virtual_gateway, old_netmask, old_gateway)?;
                     let new_current_device = CurrentDeviceInfo::new(virtual_ip, virtual_gateway,
                                                                     virtual_netmask, current_device.connect_server);
                     if let Err(e) = self.current_device.compare_exchange(current_device, new_current_device) {
@@ -310,14 +284,14 @@ impl RecvHandler {
                     dev.1 = ip_list;
                 }
             }
-            service_packet::Protocol::UnKnow(u) => {
+            service_packet::Protocol::Unknown(u) => {
                 log::warn!("未知服务协议:{}",u);
             }
         }
         Ok(())
     }
-    fn error(&self, current_device: CurrentDeviceInfo, source: Ipv4Addr, net_packet: NetPacket<&mut [u8]>, route_key: &RouteKey) -> crate::Result<()> {
-        if route_key.addr != current_device.connect_server || source != current_device.virtual_gateway() {
+    async fn error(&self, _context: &Context, current_device: CurrentDeviceInfo, _source: Ipv4Addr, net_packet: NetPacket<&mut [u8]>, route_key: &RouteKey) -> crate::Result<()> {
+        if route_key.addr != current_device.connect_server {
             return Ok(());
         }
         match InErrorPacket::new(net_packet.transport_protocol(), net_packet.payload())? {
@@ -325,8 +299,14 @@ impl RecvHandler {
                 return Err(Error::Stop("Token error".to_string()));
             }
             InErrorPacket::Disconnect => {
+                {
+                    //掉线epoch要归零
+                    let mut dev = self.device_list.lock();
+                    dev.0 = 0;
+                }
+
                 self.connect_status.store(ConnectStatus::Connecting);
-                self.register.fast_register()?;
+                self.register.fast_register().await?;
             }
             InErrorPacket::AddressExhausted => {
                 //地址用尽
@@ -338,42 +318,28 @@ impl RecvHandler {
         }
         Ok(())
     }
-    fn control(&self, current_device: CurrentDeviceInfo, source: Ipv4Addr, mut net_packet: NetPacket<&mut [u8]>, route_key: &RouteKey) -> crate::Result<()> {
+    async fn control(&self, context: &Context, current_device: CurrentDeviceInfo, source: Ipv4Addr, mut net_packet: NetPacket<&mut [u8]>, route_key: &RouteKey) -> crate::Result<()> {
+        let metric = net_packet.source_ttl() - net_packet.ttl() + 1;
         match ControlPacket::new(net_packet.transport_protocol(), net_packet.payload())? {
             ControlPacket::PingPacket(_) => {
-                let metric = net_packet.source_ttl() - net_packet.ttl() + 1;
+                context.update_read_time(&source, route_key);
                 net_packet.set_transport_protocol(control_packet::Protocol::Pong.into());
                 net_packet.set_source(current_device.virtual_ip());
                 net_packet.set_destination(source);
                 net_packet.first_set_ttl(MAX_TTL);
-                self.channel.send_to_route(net_packet.buffer(), route_key)?;
-                if metric == 1 {
-                    if let Some(current_route) = self.channel.route(&source) {
-                        if current_route.metric > 1 {
-                            let route = Route::from(*route_key, 1, -1);
-                            self.channel.add_route(source, route);
-                        }
-                    }
-                }
+                context.send_by_key(net_packet.buffer(), route_key).await?;
+                let route = Route::from(*route_key, metric, 99);
+                context.add_route_if_absent(source, route);
             }
             ControlPacket::PongPacket(pong_packet) => {
+                context.update_read_time(&source, route_key);
                 let current_time = Local::now().timestamp_millis() as u16;
                 if current_time < pong_packet.time() {
                     return Ok(());
                 }
                 let rt = (current_time - pong_packet.time()) as i64;
-                let metric = net_packet.source_ttl() - net_packet.ttl() + 1;
-                if let Some(current_route) = self.channel.route(&source) {
-                    if &current_route.route_key() == route_key {
-                        self.channel.update_route(&source, metric, rt);
-                    } else if current_route.metric >= metric && current_route.rt > rt {
-                        let route = Route::from(*route_key, metric, rt);
-                        self.channel.add_route(source, route);
-                    }
-                } else {
-                    let route = Route::from(*route_key, metric, rt);
-                    self.channel.add_route(source, route);
-                }
+                let route = Route::from(*route_key, metric, rt);
+                context.add_route(source, route);
                 if route_key.addr == current_device.connect_server && source == current_device.virtual_gateway() {
                     let epoch = self.device_list.lock().0;
                     if pong_packet.epoch() != epoch {
@@ -384,7 +350,7 @@ impl RecvHandler {
                         poll_device.first_set_ttl(MAX_TTL);
                         poll_device.set_protocol(Protocol::Service);
                         poll_device.set_transport_protocol(service_packet::Protocol::PollDeviceList.into());
-                        self.channel.send_to_route(poll_device.buffer(), route_key)?;
+                        context.send_by_key(poll_device.buffer(), route_key).await?;
                     }
                 }
             }
@@ -395,28 +361,28 @@ impl RecvHandler {
                 net_packet.set_source(current_device.virtual_ip());
                 net_packet.set_destination(source);
                 net_packet.first_set_ttl(1);
-                self.channel.send_to_route(net_packet.buffer(), route_key)?;
-                let route = Route::from(*route_key, 1, -1);
-                self.channel.add_route(source, route);
+                context.send_by_key(net_packet.buffer(), route_key).await?;
+                let route = Route::from(*route_key, metric, 99);
+                context.add_route_if_absent(source, route);
             }
             ControlPacket::PunchResponse => {
                 // log::info!("PunchResponse route_key:{:?}",route_key);
-                let route = Route::from(*route_key, 1, -1);
-                self.channel.add_route(net_packet.source(), route);
+                let route = Route::from(*route_key, metric, 99);
+                context.add_route_if_absent(source, route);
             }
         }
         Ok(())
     }
-    fn other_turn(&self, current_device: CurrentDeviceInfo, source: Ipv4Addr, net_packet: NetPacket<&mut [u8]>, route_key: &RouteKey) -> crate::Result<()> {
-        match turn_packet::Protocol::from(net_packet.transport_protocol()) {
-            turn_packet::Protocol::Punch => {
+    async fn other_turn(&self, context: &Context, current_device: CurrentDeviceInfo, source: Ipv4Addr, net_packet: NetPacket<&mut [u8]>, route_key: &RouteKey) -> crate::Result<()> {
+        match other_turn_packet::Protocol::from(net_packet.transport_protocol()) {
+            other_turn_packet::Protocol::Punch => {
                 let punch_info = PunchInfo::parse_from_bytes(net_packet.payload())?;
                 let public_ips = punch_info.public_ip_list.
-                    iter().map(|v| { IpAddr::from(v.to_be_bytes()) }).collect();
+                    iter().map(|v| { Ipv4Addr::from(v.to_be_bytes()) }).collect();
                 let peer_nat_info = NatInfo::new(public_ips,
                                                  punch_info.public_port as u16,
                                                  punch_info.public_port_range as u16,
-                                                 IpAddr::from(punch_info.local_ip.to_be_bytes()),
+                                                 Ipv4Addr::from(punch_info.local_ip.to_be_bytes()),
                                                  punch_info.local_port as u16,
                                                  punch_info.nat_type.enum_value_or_default().into());
                 self.peer_nat_info_map.insert(source, peer_nat_info.clone());
@@ -424,20 +390,12 @@ impl RecvHandler {
                     let mut punch_reply = PunchInfo::new();
                     punch_reply.reply = true;
                     let nat_info = self.nat_test.nat_info();
-                    punch_reply.public_ip_list = nat_info.public_ips.iter().map(|i| {
-                        match i {
-                            IpAddr::V4(ip) => u32::from_be_bytes(ip.octets()),
-                            IpAddr::V6(_) => 0
-                        }
-                    }).collect();
+                    punch_reply.public_ip_list = nat_info.public_ips.iter().map(|ip| u32::from_be_bytes(ip.octets())).collect();
                     punch_reply.public_port = nat_info.public_port as u32;
                     punch_reply.public_port_range = nat_info.public_port_range as u32;
                     punch_reply.nat_type =
                         protobuf::EnumOrUnknown::new(PunchNatType::from(nat_info.nat_type));
-                    punch_reply.local_ip = match nat_info.local_ip {
-                        IpAddr::V4(ip) => u32::from_be_bytes(ip.octets()),
-                        IpAddr::V6(_) => 0
-                    };
+                    punch_reply.local_ip = u32::from_be_bytes(nat_info.local_ip.octets());
                     punch_reply.local_port = nat_info.local_port as u32;
                     let bytes = punch_reply.write_to_bytes()?;
                     let mut net_packet =
@@ -445,7 +403,7 @@ impl RecvHandler {
                     net_packet.set_version(Version::V1);
                     net_packet.set_protocol(Protocol::OtherTurn);
                     net_packet.set_transport_protocol(
-                        turn_packet::Protocol::Punch.into(),
+                        other_turn_packet::Protocol::Punch.into(),
                     );
                     net_packet.first_set_ttl(MAX_TTL);
                     net_packet.set_source(current_device.virtual_ip());
@@ -459,21 +417,29 @@ impl RecvHandler {
                         packet.set_transport_protocol(control_packet::Protocol::PunchRequest.into());
                         packet.set_source(current_device.virtual_ip());
                         packet.set_destination(source);
-                        let _ = self.channel.send_to_addr(packet.buffer(), SocketAddr::new(peer_nat_info.local_ip, peer_nat_info.local_port));
+                        let _ = context.send_main(packet.buffer(), SocketAddr::V4(SocketAddrV4::new(peer_nat_info.local_ip, peer_nat_info.local_port))).await;
                     }
-                    if let Err(e) = self.channel.punch(source, peer_nat_info) {
-                        log::warn!("发送到打洞通道失败 {:?}",e);
-                        return Ok(());
+                    if self.punch(source, peer_nat_info).await {
+                        context.send_by_key(net_packet.buffer(), route_key).await?;
                     }
-                    self.channel.send_to_route(net_packet.buffer(), route_key)?;
                 } else {
-                    let _ = self.channel.punch(source, peer_nat_info);
+                    self.punch(source, peer_nat_info).await;
                 }
             }
-            turn_packet::Protocol::UnKnow(e) => {
+            other_turn_packet::Protocol::Unknown(e) => {
                 log::warn!("不支持的转发协议 {:?},source:{:?}",e,source);
             }
         }
         Ok(())
     }
+    async fn punch(&self, peer_ip: Ipv4Addr, peer_nat_info: NatInfo) -> bool {
+        match peer_nat_info.nat_type {
+            NatType::Symmetric => {
+                self.symmetric_sender.try_send((peer_ip, peer_nat_info)).is_ok()
+            }
+            NatType::Cone => {
+                self.cone_sender.try_send((peer_ip, peer_nat_info)).is_ok()
+            }
+        }
+    }
 }

switch/src/handle/registration_handler.rs

@@ -1,12 +1,12 @@
 use std::io;
-use std::net::{Ipv4Addr, SocketAddr};
+use std::net::SocketAddr;
 use std::sync::atomic::{AtomicI64, Ordering};
 use std::time::Duration;
 
 use chrono::Local;
-use p2p_channel::channel::sender::Sender;
-use p2p_channel::channel::Channel;
 use protobuf::Message;
+use tokio::net::UdpSocket;
+use crate::channel::sender::ChannelSender;
 
 use crate::error::*;
 use crate::proto::message::{RegistrationRequest, RegistrationResponse};
@@ -14,8 +14,8 @@ use crate::protocol::error_packet::InErrorPacket;
 use crate::protocol::{service_packet, NetPacket, Protocol, Version, MAX_TTL};
 
 ///向中继服务器注册，token标识一个虚拟网关，device_id防止多次注册时得到的ip不一致
-pub fn registration(
-    channel: &mut Channel<Ipv4Addr>,
+pub async fn registration(
+    main_channel: &UdpSocket,
     server_address: SocketAddr,
     token: String,
     device_id: String,
@@ -26,52 +26,60 @@ pub fn registration(
     let buf = request_packet.buffer();
     let mut recv_buf = [0u8; 10240];
     let mut count = 0;
-    let len = loop {
-        match channel.send_to_addr(buf, server_address) {
+    loop {
+        match main_channel.send_to(buf, server_address).await {
             Ok(_) => {
-                match channel.recv_from(&mut recv_buf, Some(Duration::from_millis(300))) {
-                    Ok((len, route)) => {
-                        if server_address == route.addr {
-                            let net_packet = NetPacket::new(&recv_buf[..len])?;
-                            match net_packet.protocol() {
-                                Protocol::Service => {
-                                    match service_packet::Protocol::from(net_packet.transport_protocol()) {
-                                        service_packet::Protocol::RegistrationResponse => {
-                                            let response = RegistrationResponse::parse_from_bytes(net_packet.payload())?;
-                                            return Ok(response);
+                match tokio::time::timeout(Duration::from_millis(300), main_channel.recv_from(&mut recv_buf)).await {
+                    Ok(rs) => {
+                        match rs {
+                            Ok((len, addr)) => {
+                                if server_address == addr {
+                                    let net_packet = NetPacket::new(&recv_buf[..len])?;
+                                    match net_packet.protocol() {
+                                        Protocol::Service => {
+                                            match service_packet::Protocol::from(net_packet.transport_protocol()) {
+                                                service_packet::Protocol::RegistrationResponse => {
+                                                    let response = RegistrationResponse::parse_from_bytes(net_packet.payload())?;
+                                                    return Ok(response);
+                                                }
+                                                _ => println!("响应数据错误"),
+                                            }
+                                        }
+                                        Protocol::Error => {
+                                            match InErrorPacket::new(net_packet.transport_protocol(), net_packet.payload()) {
+                                                Ok(e) => match e {
+                                                    InErrorPacket::TokenError => return Err(Error::Stop("token错误".to_string())),
+                                                    InErrorPacket::Disconnect => {
+                                                        println!("断开连接");
+                                                    }
+                                                    InErrorPacket::AddressExhausted => {
+                                                        println!("地址用尽");
+                                                        log::warn!("地址用尽");
+                                                    }
+                                                    InErrorPacket::OtherError(e) => match e.message() {
+                                                        Ok(str) => {
+                                                            println!("其他异常:{:?}", str);
+                                                            log::warn!("其他异常{:?}",str);
+                                                        }
+                                                        Err(e) => println!("其他异常:{:?}", e),
+                                                    },
+                                                },
+                                                Err(e) => println!("数据解析异常:{:?}", e),
+                                            }
                                         }
                                         _ => println!("响应数据错误"),
-                                    }
+                                    };
                                 }
-                                Protocol::Error => {
-                                    match InErrorPacket::new(net_packet.transport_protocol(), net_packet.payload()) {
-                                        Ok(e) => match e {
-                                            InErrorPacket::TokenError => return Err(Error::Stop("token错误".to_string())),
-                                            InErrorPacket::Disconnect => {
-                                                println!("断开连接");
-                                            }
-                                            InErrorPacket::AddressExhausted => {
-                                                println!("地址用尽");
-                                                log::warn!("地址用尽");
-                                            }
-                                            InErrorPacket::OtherError(e) => match e.message() {
-                                                Ok(str) => {
-                                                    println!("其他异常:{:?}", str);
-                                                    log::warn!("其他异常{:?}",str);
-                                                }
-                                                Err(e) => println!("其他异常:{:?}", e),
-                                            },
-                                        },
-                                        Err(e) => println!("数据解析异常:{:?}", e),
-                                    }
-                                }
-                                _ => println!("响应数据错误"),
-                            };
+                            }
+                            Err(e) => {
+                                println!("接收服务器数据失败:{:?}", e);
+                                log::warn!("接收服务器数据失败:{:?}",e);
+                            }
                         }
                     }
-                    Err(e) => {
-                        println!("接收服务器数据失败:{:?}", e);
-                        log::warn!("接收服务器数据失败:{:?}",e);
+                    Err(_) => {
+                        println!("接收超时");
+                        log::warn!("接收超时");
                     }
                 }
             }
@@ -97,6 +105,7 @@ fn registration_request_packet(
     request.device_id = device_id;
     request.name = name;
     request.is_fast = is_fast;
+    request.version = "1.0.6".to_string();
     let bytes = request.write_to_bytes()?;
     let buf = vec![0u8; 12 + bytes.len()];
     let mut net_packet = NetPacket::new(buf)?;
@@ -109,7 +118,7 @@ fn registration_request_packet(
 }
 
 pub struct Register {
-    sender: Sender<Ipv4Addr>,
+    sender: ChannelSender,
     server_address: SocketAddr,
     token: String,
     device_id: String,
@@ -119,7 +128,7 @@ pub struct Register {
 
 impl Register {
     pub fn new(
-        sender: Sender<Ipv4Addr>,
+        sender: ChannelSender,
         server_address: SocketAddr,
         token: String,
         device_id: String,
@@ -134,7 +143,7 @@ impl Register {
             time: AtomicI64::new(0),
         }
     }
-    pub fn fast_register(&self) -> io::Result<()> {
+    pub async fn fast_register(&self) -> io::Result<()> {
         let last = self.time.load(Ordering::Relaxed);
         let new = Local::now().timestamp_millis();
         if new - last < 1000
@@ -155,7 +164,7 @@ impl Register {
         )
             .unwrap();
         let buf = request_packet.buffer();
-        self.sender.send_to_addr(buf, self.server_address)?;
+        self.sender.send_main(buf, self.server_address).await?;
         Ok(())
     }
 }

switch/src/handle/tap_handler.rs

@@ -1,163 +0,0 @@
-use std::net::{Ipv4Addr, SocketAddrV4};
-use std::sync::Arc;
-use std::{io, thread};
-use crossbeam::atomic::AtomicCell;
-use p2p_channel::channel::sender::Sender;
-use packet::arp::arp::ArpPacket;
-use packet::ethernet;
-use packet::ethernet::packet::EthernetPacket;
-use packet::icmp::icmp::IcmpPacket;
-use packet::icmp::Kind;
-use packet::ip::ipv4;
-use packet::ip::ipv4::packet::IpV4Packet;
-use crate::external_route::ExternalRoute;
-use crate::handle::{check_dest, CurrentDeviceInfo};
-use crate::ip_proxy::IpProxyMap;
-use crate::protocol::{MAX_TTL, NetPacket, Protocol, Version};
-use crate::tap_device::{TapReader, TapWriter};
-
-pub fn start(sender: Sender<Ipv4Addr>,
-             tap_reader: TapReader,
-             tap_writer: TapWriter,
-             current_device: Arc<AtomicCell<CurrentDeviceInfo>>,
-             ip_route: ExternalRoute,
-             ip_proxy_map: IpProxyMap) {
-    thread::Builder::new().name("tap-handler".into()).spawn(move || {
-        if let Err(e) = start_(sender, tap_reader, tap_writer, current_device,ip_route,ip_proxy_map) {
-            log::warn!("{:?}",e);
-        }
-    }).unwrap();
-}
-
-fn start_(sender: Sender<Ipv4Addr>,
-          tap_reader: TapReader,
-          tap_writer: TapWriter,
-          current_device: Arc<AtomicCell<CurrentDeviceInfo>>,
-          ip_route: ExternalRoute,
-          ip_proxy_map: IpProxyMap) -> io::Result<()> {
-    let mut net_packet = NetPacket::new(vec![0u8; 4 + 8 + 1500]).unwrap();
-    net_packet.set_version(Version::V1);
-    net_packet.set_protocol(Protocol::Ipv4Turn);
-    net_packet.set_transport_protocol(ipv4::protocol::Protocol::Ipv4.into());
-    net_packet.set_ttl(MAX_TTL);
-    let mut buf = [0; 2048];
-    loop {
-        let len = tap_reader.read(&mut buf)?;
-        if len == 0 {
-            continue;
-        }
-        let mut ethernet_packet = EthernetPacket::unchecked(&mut buf[..len]);
-        if let Err(e) = handle(&mut net_packet, &current_device, &tap_writer, &mut ethernet_packet, &sender,&ip_route,&ip_proxy_map) {
-            log::error!("tap handle{:?}",e);
-        }
-    }
-}
-
-fn handle(net_packet: &mut NetPacket<Vec<u8>>, current_device: &AtomicCell<CurrentDeviceInfo>, tap_writer: &TapWriter, ethernet_packet: &mut EthernetPacket<&mut [u8]>, sender: &Sender<Ipv4Addr>, ip_route: &ExternalRoute, proxy_map: &IpProxyMap) -> crate::Result<()> {
-    let current_device = current_device.load();
-    match ethernet_packet.protocol() {
-        ethernet::protocol::Protocol::Arp => {
-            let mut out_ethernet_packet = EthernetPacket::unchecked(ethernet_packet.buffer.to_vec());
-            let arp_packet = ArpPacket::unchecked(ethernet_packet.payload());
-            let mut out_arp_packet = ArpPacket::unchecked(out_ethernet_packet.payload_mut());
-            let sender_h = arp_packet.sender_hardware_addr();
-            let sender_p = arp_packet.sender_protocol_addr();
-            let target_p = arp_packet.target_protocol_addr();
-            if target_p == &[0, 0, 0, 0] || sender_p == &[0, 0, 0, 0] || target_p == sender_p {
-                return Ok(());
-            }
-            //回复一个虚假的MAC地址
-            out_arp_packet.set_sender_hardware_addr(&[target_p[0], target_p[1], target_p[2], target_p[3], 123, 234]);
-            out_arp_packet.set_sender_protocol_addr(target_p);
-            out_arp_packet.set_target_hardware_addr(sender_h);
-            out_arp_packet.set_target_protocol_addr(sender_p);
-            out_arp_packet.set_op_code(2);
-            out_ethernet_packet.set_source(&[target_p[0], target_p[1], target_p[2], target_p[3], 123, 234]);
-            out_ethernet_packet.set_destination(sender_h);
-
-            tap_writer.write(&out_ethernet_packet.buffer)?;
-        }
-        ethernet::protocol::Protocol::Ipv4 => {
-            // println!("in ethernet_packet {:?}", ethernet_packet);
-            let mut ipv4_packet = IpV4Packet::unchecked(ethernet_packet.payload_mut());
-            let src_ip = ipv4_packet.source_ip();
-            let mut dest_ip = ipv4_packet.destination_ip();
-            if src_ip != current_device.virtual_ip() {
-                return Ok(());
-            }
-            if !check_dest(dest_ip, current_device.virtual_netmask, current_device.virtual_network) && !dest_ip.is_broadcast() {
-                if let Some(r_dest_ip) = ip_route.route(&dest_ip) {
-                    //路由的目标不能是自己
-                    if r_dest_ip == src_ip {
-                        return Ok(());
-                    }
-                    dest_ip = r_dest_ip;
-                } else {
-                    return Ok(());
-                }
-            }else{
-                match ipv4_packet.protocol() {
-                    ipv4::protocol::Protocol::Tcp => {
-                        let dest_addr = {
-                            let tcp_packet = packet::tcp::tcp::TcpPacket::new(src_ip, dest_ip, ipv4_packet.payload())?;
-                            SocketAddrV4::new(dest_ip, tcp_packet.destination_port())
-                        };
-                        if let Some(entry) = proxy_map.tcp_proxy_map.get(&dest_addr) {
-                            let source_addr = entry.value().1;
-                            let source_ip = *source_addr.ip();
-                            let mut tcp_packet = packet::tcp::tcp::TcpPacket::new(source_ip, dest_ip, ipv4_packet.payload_mut())?;
-                            tcp_packet.set_source_port(source_addr.port());
-                            tcp_packet.update_checksum();
-                            ipv4_packet.set_source_ip(source_ip);
-                            ipv4_packet.update_checksum();
-                        }
-                    }
-                    ipv4::protocol::Protocol::Udp => {
-                        let dest_addr = {
-                            let udp_packet = packet::udp::udp::UdpPacket::new(src_ip, dest_ip, ipv4_packet.payload())?;
-                            SocketAddrV4::new(dest_ip, udp_packet.destination_port())
-                        };
-                        if let Some(entry) = proxy_map.udp_proxy_map.get(&dest_addr) {
-                            let source_addr = entry.value().1;
-                            let source_ip = *source_addr.ip();
-                            let mut udp_packet = packet::udp::udp::UdpPacket::new(src_ip, dest_ip, ipv4_packet.payload_mut())?;
-                            udp_packet.set_source_port(source_addr.port());
-                            udp_packet.update_checksum();
-                            ipv4_packet.set_source_ip(source_ip);
-                            ipv4_packet.update_checksum();
-                        }
-                    }
-                    _ => {}
-                }
-            }
-            if src_ip == dest_ip {
-                if ipv4_packet.protocol() == ipv4::protocol::Protocol::Icmp {
-                    let mut icmp = IcmpPacket::unchecked(ipv4_packet.payload_mut());
-                    if icmp.kind() == Kind::EchoRequest {
-                        icmp.set_kind(Kind::EchoReply);
-                        icmp.update_checksum();
-                        let src = ipv4_packet.source_ip();
-                        ipv4_packet.set_source_ip(ipv4_packet.destination_ip());
-                        ipv4_packet.set_destination_ip(src);
-                        ipv4_packet.update_checksum();
-                        tap_writer.write(ethernet_packet.buffer)?;
-                        return Ok(());
-                    }
-                }
-            }
-            net_packet.set_source(src_ip);
-            net_packet.set_destination(dest_ip);
-            let data_len = ipv4_packet.buffer.len();
-            net_packet.set_payload(ipv4_packet.buffer);
-            //优先发到直连到地址
-            if sender.send_to_id(&net_packet.buffer()[..(12 + data_len)], &dest_ip).is_err() {
-                sender.send_to_addr(&net_packet.buffer()[..(12 + data_len)], current_device.connect_server)?;
-            }
-        }
-        _ => {
-            // log::warn!("不支持的二层协议：{:?}",p)
-        }
-    }
-    Ok(())
-}
-

switch/src/handle/tun_handler.rs

@@ -1,175 +0,0 @@
-use std::{io, thread};
-use std::net::{Ipv4Addr, SocketAddrV4};
-use std::sync::Arc;
-
-use crossbeam::atomic::AtomicCell;
-
-use p2p_channel::channel::sender::Sender;
-use packet::icmp:: Kind;
-use packet::icmp::icmp:: IcmpPacket;
-use packet::ip::ipv4;
-use packet::ip::ipv4::packet::IpV4Packet;
-
-use crate::error::*;
-use crate::external_route::ExternalRoute;
-use crate::handle::{check_dest, CurrentDeviceInfo};
-use crate::ip_proxy::IpProxyMap;
-use crate::protocol::{MAX_TTL, NetPacket, Protocol, Version};
-use crate::tun_device::{TunReader, TunWriter};
-
-fn icmp(tun_writer: &TunWriter, mut ipv4_packet: IpV4Packet<&mut [u8]>) -> Result<()> {
-    if ipv4_packet.protocol() == ipv4::protocol::Protocol::Icmp {
-        let mut icmp = IcmpPacket::new(ipv4_packet.payload_mut())?;
-        if icmp.kind() == Kind::EchoRequest {
-            icmp.set_kind(Kind::EchoReply);
-            icmp.update_checksum();
-            let src = ipv4_packet.source_ip();
-            ipv4_packet.set_source_ip(ipv4_packet.destination_ip());
-            ipv4_packet.set_destination_ip(src);
-            ipv4_packet.update_checksum();
-            tun_writer.write(ipv4_packet.buffer)?;
-        }
-    }
-    Ok(())
-}
-
-/// 接收tun数据，并且转发到udp上
-#[inline]
-fn handle(sender: &Sender<Ipv4Addr>, data: &mut [u8], tun_writer: &TunWriter, current_device: CurrentDeviceInfo, net_packet: &mut NetPacket<[u8; 1512]>, ip_route: &ExternalRoute, proxy_map: &IpProxyMap) -> Result<()> {
-    let data_len = data.len();
-    let mut ipv4_packet = match IpV4Packet::new(data) {
-        Ok(ipv4_packet) => ipv4_packet,
-        Err(packet::error::Error::Unimplemented) => {
-            return Ok(());
-        }
-        Err(e) => Err(e)?,
-    };
-    let src_ip = ipv4_packet.source_ip();
-    let mut dest_ip = ipv4_packet.destination_ip();
-    // if dest_ip == cur_info.broadcast_address {
-    //     // 启动服务后会收到对137端口的广播
-    //     // 137端口是在局域网中提供计算机的名字或IP地址查询服务
-    //     return Ok(());
-    // }
-    if src_ip != current_device.virtual_ip() {
-        return Ok(());
-    }
-    if src_ip == dest_ip {
-        return icmp(&tun_writer, ipv4_packet);
-    }
-    if !check_dest(dest_ip, current_device.virtual_netmask, current_device.virtual_network) && !dest_ip.is_broadcast() {
-        // println!("非目标 {:?}",ipv4_packet);
-        if let Some(r_dest_ip) = ip_route.route(&dest_ip) {
-            //路由的目标不能是自己
-            if r_dest_ip == src_ip {
-                return Ok(());
-            }
-            dest_ip = r_dest_ip;
-        } else {
-            return Ok(());
-        }
-    } else {
-        match ipv4_packet.protocol() {
-            ipv4::protocol::Protocol::Tcp => {
-                let dest_addr = {
-                    let tcp_packet = packet::tcp::tcp::TcpPacket::new(src_ip, dest_ip, ipv4_packet.payload())?;
-                    SocketAddrV4::new(dest_ip, tcp_packet.destination_port())
-                };
-                if let Some(entry) = proxy_map.tcp_proxy_map.get(&dest_addr) {
-                    let source_addr = entry.value().1;
-                    let source_ip = *source_addr.ip();
-                    let mut tcp_packet = packet::tcp::tcp::TcpPacket::new(source_ip, dest_ip, ipv4_packet.payload_mut())?;
-                    tcp_packet.set_source_port(source_addr.port());
-                    tcp_packet.update_checksum();
-                    ipv4_packet.set_source_ip(source_ip);
-                    ipv4_packet.update_checksum();
-                }
-            }
-            ipv4::protocol::Protocol::Udp => {
-                let dest_addr = {
-                    let udp_packet = packet::udp::udp::UdpPacket::new(src_ip, dest_ip, ipv4_packet.payload())?;
-                    SocketAddrV4::new(dest_ip, udp_packet.destination_port())
-                };
-                if let Some(entry) = proxy_map.udp_proxy_map.get(&dest_addr) {
-                    let source_addr = entry.value().1;
-                    let source_ip = *source_addr.ip();
-                    let mut udp_packet = packet::udp::udp::UdpPacket::new(source_ip, dest_ip, ipv4_packet.payload_mut())?;
-                    udp_packet.set_source_port(source_addr.port());
-                    udp_packet.update_checksum();
-                    ipv4_packet.set_source_ip(source_ip);
-                    ipv4_packet.update_checksum();
-                }
-            }
-            _ => {}
-        }
-    }
-
-    net_packet.set_source(src_ip);
-    net_packet.set_destination(dest_ip);
-    net_packet.set_payload(ipv4_packet.buffer);
-    //优先发到直连到地址
-    if sender.send_to_id(&net_packet.buffer()[..(12 + data_len)], &dest_ip).is_err() {
-        sender.send_to_addr(&net_packet.buffer()[..(12 + data_len)], current_device.connect_server)?;
-    }
-    return Ok(());
-}
-
-pub fn start(sender: Sender<Ipv4Addr>,
-             tun_reader: TunReader,
-             tun_writer: TunWriter,
-             current_device: Arc<AtomicCell<CurrentDeviceInfo>>,
-             ip_route: ExternalRoute,
-             ip_proxy_map: IpProxyMap) {
-    thread::Builder::new().name("tun-handler".into()).spawn(move || {
-        if let Err(e) = start_(sender, tun_reader, tun_writer, current_device, ip_route, ip_proxy_map) {
-            log::warn!("{:?}",e);
-        }
-    }).unwrap();
-}
-
-#[cfg(target_os = "windows")]
-fn start_(sender: Sender<Ipv4Addr>,
-          tun_reader: TunReader,
-          tun_writer: TunWriter,
-          current_device: Arc<AtomicCell<CurrentDeviceInfo>>,
-          ip_route: ExternalRoute,
-          ip_proxy_map: IpProxyMap) -> io::Result<()> {
-    let mut net_packet = NetPacket::new([0u8; 4 + 8 + 1500])?;
-    net_packet.set_version(Version::V1);
-    net_packet.set_protocol(Protocol::Ipv4Turn);
-    net_packet.set_transport_protocol(ipv4::protocol::Protocol::Ipv4.into());
-    net_packet.set_ttl(MAX_TTL);
-    loop {
-        let mut data = tun_reader.next()?;
-        match handle(&sender, data.bytes_mut(), &tun_writer, current_device.load(), &mut net_packet, &ip_route, &ip_proxy_map) {
-            Ok(_) => {}
-            Err(e) => {
-                log::warn!("{:?}", e)
-            }
-        }
-    }
-}
-
-#[cfg(any(target_os = "linux", target_os = "macos"))]
-fn start_(sender: Sender<Ipv4Addr>,
-          tun_reader: TunReader,
-          tun_writer: TunWriter,
-          current_device: Arc<AtomicCell<CurrentDeviceInfo>>,
-          ip_route: ExternalRoute,
-          ip_proxy_map: IpProxyMap) -> io::Result<()> {
-    let mut net_packet = NetPacket::new([0u8; 4 + 8 + 1500])?;
-    net_packet.set_version(Version::V1);
-    net_packet.set_protocol(Protocol::Ipv4Turn);
-    net_packet.set_transport_protocol(ipv4::protocol::Protocol::Ipv4.into());
-    net_packet.set_ttl(MAX_TTL);
-    let mut buf = [0; 4096];
-    loop {
-        let len = tun_reader.read(&mut buf)?;
-        match handle(&sender, &mut buf[..len], &tun_writer, current_device.load(), &mut net_packet, &ip_route, &ip_proxy_map) {
-            Ok(_) => {}
-            Err(e) => {
-                log::warn!("{:?}", e)
-            }
-        }
-    }
-}

switch/src/ip_proxy/icmp_proxy.rs

@@ -6,11 +6,11 @@ use crossbeam::atomic::AtomicCell;
 
 use crossbeam_skiplist::SkipMap;
 use socket2::{Domain, SockAddr, Socket, Type};
-use p2p_channel::channel::sender::Sender;
 
 use packet::icmp::icmp;
 use packet::icmp::icmp::HeaderOther;
 use packet::ip::ipv4;
+use crate::channel::sender::ChannelSender;
 use crate::handle::CurrentDeviceInfo;
 use crate::protocol::{MAX_TTL, NetPacket, Protocol, Version};
 
@@ -18,12 +18,12 @@ pub struct IcmpProxy {
     icmp_socket: Arc<Socket>,
     // 对端-> 真实来源
     icmp_proxy_map: Arc<SkipMap<(Ipv4Addr, u16, u16), Ipv4Addr>>,
-    sender: Sender<Ipv4Addr>,
+    sender: ChannelSender,
     current_device: Arc<AtomicCell<CurrentDeviceInfo>>,
 }
 
 impl IcmpProxy {
-    pub fn new(addr: SocketAddrV4, icmp_proxy_map: Arc<SkipMap<(Ipv4Addr, u16, u16), Ipv4Addr>>, sender: Sender<Ipv4Addr>, current_device: Arc<AtomicCell<CurrentDeviceInfo>>) -> io::Result<IcmpProxy> {
+    pub fn new(addr: SocketAddrV4, icmp_proxy_map: Arc<SkipMap<(Ipv4Addr, u16, u16), Ipv4Addr>>, sender: ChannelSender, current_device: Arc<AtomicCell<CurrentDeviceInfo>>) -> io::Result<IcmpProxy> {
         let icmp_socket = Arc::new(Socket::new(Domain::IPV4, Type::RAW, Some(socket2::Protocol::ICMPV4))?);
         icmp_socket.bind(&SockAddr::from(addr))?;
         // // 设置 SIO_RCVALL 参数
@@ -66,7 +66,7 @@ impl IcmpProxy {
             unsafe { std::mem::transmute(&mut buf[..]) };
         let mut net_packet = NetPacket::new([0u8; 4 + 8 + 1500]).unwrap();
         net_packet.set_version(Version::V1);
-        net_packet.set_protocol(Protocol::Ipv4Turn);
+        net_packet.set_protocol(Protocol::IpTurn);
         net_packet.set_transport_protocol(ipv4::protocol::Protocol::Ipv4.into());
         net_packet.set_ttl(MAX_TTL);
         loop {
@@ -90,7 +90,7 @@ impl IcmpProxy {
                                                         net_packet.set_destination(dest_ip);
                                                         let data_len = ipv4_packet.buffer.len();
                                                         net_packet.set_payload(ipv4_packet.buffer);
-                                                        let _ = self.sender.send_to_id(&net_packet.buffer()[..(12 + data_len)], &dest_ip);
+                                                        let _ = self.sender.try_send_by_id(&net_packet.buffer()[..(12 + data_len)], &dest_ip);
                                                     }
                                                 }
                                                 _ => {

switch/src/ip_proxy/mod.rs

@@ -6,7 +6,7 @@ use crossbeam::atomic::AtomicCell;
 use crossbeam_skiplist::SkipMap;
 use socket2::{SockAddr, Socket};
 use tokio::net::{TcpListener, UdpSocket};
-use p2p_channel::channel::sender::Sender;
+use crate::channel::sender::ChannelSender;
 use crate::handle::CurrentDeviceInfo;
 use crate::ip_proxy::icmp_proxy::IcmpProxy;
 use crate::ip_proxy::tcp_proxy::TcpProxy;
@@ -45,34 +45,34 @@ impl IpProxyMap {
     }
 }
 
-pub async fn init_proxy(sender: Sender<Ipv4Addr>, bind_ips: Vec<Ipv4Addr>, current_device: Arc<AtomicCell<CurrentDeviceInfo>>) -> io::Result<IpProxyMap> {
+pub async fn init_proxy(sender: ChannelSender, bind_ips: Vec<Ipv4Addr>, current_device: Arc<AtomicCell<CurrentDeviceInfo>>) -> io::Result<IpProxyMap> {
     let mut icmp_sockets = HashMap::new();
     let tcp_proxy_map: Arc<SkipMap<SocketAddrV4, (SocketAddrV4, SocketAddrV4)>> = Arc::new(SkipMap::new());
     let udp_proxy_map: Arc<SkipMap<SocketAddrV4, (SocketAddrV4, SocketAddrV4)>> = Arc::new(SkipMap::new());
     let icmp_proxy_map: Arc<SkipMap<(Ipv4Addr, u16, u16), Ipv4Addr>> = Arc::new(SkipMap::new());
-    let tcp_listener = TcpListener::bind("0.0.0.0:0").await?;
-    let udp_socket = UdpSocket::bind("0.0.0.0:0").await?;
-    let tcp_proxy_port = tcp_listener.local_addr()?.port();
-    let udp_proxy_port = udp_socket.local_addr()?.port();
-
-    {
+    let (tcp_proxy_port, udp_proxy_port) = if !bind_ips.is_empty() {
+        let tcp_listener = TcpListener::bind("0.0.0.0:0").await?;
+        let udp_socket = UdpSocket::bind("0.0.0.0:0").await?;
+        let tcp_proxy_port = tcp_listener.local_addr()?.port();
+        let udp_proxy_port = udp_socket.local_addr()?.port();
         let tcp_proxy_map = tcp_proxy_map.clone();
         tokio::spawn(async {
             let tcp_proxy = TcpProxy::new(tcp_listener, tcp_proxy_map);
             tcp_proxy.start().await
         });
-    }
-    {
         let udp_proxy_map = udp_proxy_map.clone();
         tokio::spawn(async {
             let udp_proxy = UdpProxy::new(udp_socket, udp_proxy_map);
             udp_proxy.start().await
         });
-    }
+        (tcp_proxy_port, udp_proxy_port)
+    } else {
+        (0, 0)
+    };
     for ip in bind_ips {
         let addr = SocketAddrV4::new(ip, 0);
         let icmp_proxy_map = icmp_proxy_map.clone();
-        let icmp_proxy = IcmpProxy::new(addr, icmp_proxy_map, sender.try_clone()?, current_device.clone())?;
+        let icmp_proxy = IcmpProxy::new(addr, icmp_proxy_map, sender.clone(), current_device.clone())?;
         icmp_sockets.insert(ip, icmp_proxy.icmp_socket());
         thread::spawn(move || {
             icmp_proxy.start();

switch/src/lib.rs

@@ -1,7 +1,5 @@
 use crate::error::Error;
 
-pub use p2p_channel::channel::{Route, RouteKey};
-
 pub type Result<T> = std::result::Result<T, Error>;
 
 pub mod error;
@@ -9,8 +7,9 @@ pub mod handle;
 pub mod nat;
 pub mod proto;
 pub mod protocol;
-pub mod tun_device;
-pub mod tap_device;
 pub mod ip_proxy;
 pub mod external_route;
+pub mod igmp_server;
+pub mod tun_tap_device;
 pub mod core;
+pub mod channel;

switch/src/nat/check.rs

@@ -1,8 +1,8 @@
-use p2p_channel::punch::NatType;
 use std::collections::HashSet;
 use std::net::{IpAddr, Ipv4Addr, SocketAddr, UdpSocket};
 use std::time::Duration;
 use std::{io, thread};
+use crate::channel::punch::NatType;
 
 // #[derive(Debug, Copy, Clone, PartialEq)]
 // pub enum NatType {

switch/src/nat/mod.rs

@@ -1,5 +1,4 @@
 use crate::proto::message::PunchNatType;
-use p2p_channel::punch::{NatInfo, NatType};
 use parking_lot::Mutex;
 use std::io;
 use std::net::{IpAddr, Ipv4Addr, SocketAddr};
@@ -8,12 +7,20 @@ use std::sync::Arc;
 pub mod check;
 
 use std::net::UdpSocket;
+use crate::channel::punch::{NatInfo, NatType};
 
-pub fn local_ip() -> io::Result<IpAddr> {
+pub fn local_ip() -> io::Result<Ipv4Addr> {
     let socket = UdpSocket::bind("0.0.0.0:0")?;
     socket.connect("8.8.8.8:80")?;
     let addr = socket.local_addr()?;
-    Ok(addr.ip())
+    match addr.ip() {
+        IpAddr::V4(ip) => {
+            Ok(ip)
+        }
+        IpAddr::V6(_) => {
+            Ok(Ipv4Addr::UNSPECIFIED)
+        }
+    }
 }
 
 #[derive(Clone)]
@@ -45,7 +52,7 @@ impl NatTest {
         nat_test_server: Vec<SocketAddr>,
         public_ip: Ipv4Addr,
         public_port: u16,
-        local_ip: IpAddr,
+        local_ip: Ipv4Addr,
         local_port: u16,
     ) -> NatTest {
         let info = NatTest::re_test_(
@@ -67,7 +74,7 @@ impl NatTest {
         &self,
         public_ip: Ipv4Addr,
         public_port: u16,
-        local_ip: IpAddr,
+        local_ip: Ipv4Addr,
         local_port: u16,
     ) -> NatInfo {
         let info = NatTest::re_test_(
@@ -84,16 +91,16 @@ impl NatTest {
         nat_test_server: &Vec<SocketAddr>,
         public_ip: Ipv4Addr,
         public_port: u16,
-        local_ip: IpAddr,
+        local_ip: Ipv4Addr,
         local_port: u16,
     ) -> NatInfo {
         return match check::public_ip_list(nat_test_server) {
             Ok((nat_type, ips, port_range)) => {
                 let mut public_ips = Vec::new();
-                public_ips.push(IpAddr::from(public_ip));
+                public_ips.push(Ipv4Addr::from(public_ip));
                 for ip in ips {
                     if ip != public_ip {
-                        public_ips.push(IpAddr::from(ip));
+                        public_ips.push(ip);
                     }
                 }
                 NatInfo::new(
@@ -108,7 +115,7 @@ impl NatTest {
             Err(e) => {
                 log::warn!("{:?}", e);
                 NatInfo::new(
-                    vec![IpAddr::from(public_ip)],
+                    vec![public_ip],
                     public_port,
                     0,
                     local_ip,

switch/src/proto/message.rs

@@ -37,6 +37,8 @@ pub struct RegistrationRequest {
     pub name: ::std::string::String,
     // @@protoc_insertion_point(field:RegistrationRequest.is_fast)
     pub is_fast: bool,
+    // @@protoc_insertion_point(field:RegistrationRequest.version)
+    pub version: ::std::string::String,
     // special fields
     // @@protoc_insertion_point(special_field:RegistrationRequest.special_fields)
     pub special_fields: ::protobuf::SpecialFields,
@@ -54,7 +56,7 @@ impl RegistrationRequest {
     }
 
     fn generated_message_descriptor_data() -> ::protobuf::reflect::GeneratedMessageDescriptorData {
-        let mut fields = ::std::vec::Vec::with_capacity(4);
+        let mut fields = ::std::vec::Vec::with_capacity(5);
         let mut oneofs = ::std::vec::Vec::with_capacity(0);
         fields.push(::protobuf::reflect::rt::v2::make_simpler_field_accessor::<_, _>(
             "token",
@@ -76,6 +78,11 @@ impl RegistrationRequest {
             |m: &RegistrationRequest| { &m.is_fast },
             |m: &mut RegistrationRequest| { &mut m.is_fast },
         ));
+        fields.push(::protobuf::reflect::rt::v2::make_simpler_field_accessor::<_, _>(
+            "version",
+            |m: &RegistrationRequest| { &m.version },
+            |m: &mut RegistrationRequest| { &mut m.version },
+        ));
         ::protobuf::reflect::GeneratedMessageDescriptorData::new_2::<RegistrationRequest>(
             "RegistrationRequest",
             fields,
@@ -106,6 +113,9 @@ impl ::protobuf::Message for RegistrationRequest {
                 32 => {
                     self.is_fast = is.read_bool()?;
                 },
+                42 => {
+                    self.version = is.read_string()?;
+                },
                 tag => {
                     ::protobuf::rt::read_unknown_or_skip_group(tag, is, self.special_fields.mut_unknown_fields())?;
                 },
@@ -130,6 +140,9 @@ impl ::protobuf::Message for RegistrationRequest {
         if self.is_fast != false {
             my_size += 1 + 1;
         }
+        if !self.version.is_empty() {
+            my_size += ::protobuf::rt::string_size(5, &self.version);
+        }
         my_size += ::protobuf::rt::unknown_fields_size(self.special_fields.unknown_fields());
         self.special_fields.cached_size().set(my_size as u32);
         my_size
@@ -148,6 +161,9 @@ impl ::protobuf::Message for RegistrationRequest {
         if self.is_fast != false {
             os.write_bool(4, self.is_fast)?;
         }
+        if !self.version.is_empty() {
+            os.write_string(5, &self.version)?;
+        }
         os.write_unknown_fields(self.special_fields.unknown_fields())?;
         ::std::result::Result::Ok(())
     }
@@ -169,6 +185,7 @@ impl ::protobuf::Message for RegistrationRequest {
         self.device_id.clear();
         self.name.clear();
         self.is_fast = false;
+        self.version.clear();
         self.special_fields.clear();
     }
 
@@ -178,6 +195,7 @@ impl ::protobuf::Message for RegistrationRequest {
             device_id: ::std::string::String::new(),
             name: ::std::string::String::new(),
             is_fast: false,
+            version: ::std::string::String::new(),
             special_fields: ::protobuf::SpecialFields::new(),
         };
         &instance
@@ -1017,28 +1035,29 @@ impl PunchNatType {
 }
 
 static file_descriptor_proto_data: &'static [u8] = b"\
-    \n\rmessage.proto\"u\n\x13RegistrationRequest\x12\x14\n\x05token\x18\x01\
-    \x20\x01(\tR\x05token\x12\x1b\n\tdevice_id\x18\x02\x20\x01(\tR\x08device\
-    Id\x12\x12\n\x04name\x18\x03\x20\x01(\tR\x04name\x12\x17\n\x07is_fast\
-    \x18\x04\x20\x01(\x08R\x06isFast\"\x92\x02\n\x14RegistrationResponse\x12\
-    \x1d\n\nvirtual_ip\x18\x01\x20\x01(\x07R\tvirtualIp\x12'\n\x0fvirtual_ga\
-    teway\x18\x02\x20\x01(\x07R\x0evirtualGateway\x12'\n\x0fvirtual_netmask\
-    \x18\x03\x20\x01(\x07R\x0evirtualNetmask\x12\x14\n\x05epoch\x18\x04\x20\
-    \x01(\rR\x05epoch\x125\n\x10device_info_list\x18\x05\x20\x03(\x0b2\x0b.D\
-    eviceInfoR\x0edeviceInfoList\x12\x1b\n\tpublic_ip\x18\x06\x20\x01(\x07R\
-    \x08publicIp\x12\x1f\n\x0bpublic_port\x18\x07\x20\x01(\rR\npublicPort\"d\
-    \n\nDeviceInfo\x12\x12\n\x04name\x18\x01\x20\x01(\tR\x04name\x12\x1d\n\n\
-    virtual_ip\x18\x02\x20\x01(\x07R\tvirtualIp\x12#\n\rdevice_status\x18\
-    \x03\x20\x01(\rR\x0cdeviceStatus\"Y\n\nDeviceList\x12\x14\n\x05epoch\x18\
-    \x01\x20\x01(\rR\x05epoch\x125\n\x10device_info_list\x18\x02\x20\x03(\
-    \x0b2\x0b.DeviceInfoR\x0edeviceInfoList\"\xf8\x01\n\tPunchInfo\x12$\n\
-    \x0epublic_ip_list\x18\x02\x20\x03(\x07R\x0cpublicIpList\x12\x1f\n\x0bpu\
-    blic_port\x18\x03\x20\x01(\rR\npublicPort\x12*\n\x11public_port_range\
-    \x18\x04\x20\x01(\rR\x0fpublicPortRange\x12(\n\x08nat_type\x18\x05\x20\
-    \x01(\x0e2\r.PunchNatTypeR\x07natType\x12\x14\n\x05reply\x18\x06\x20\x01\
-    (\x08R\x05reply\x12\x19\n\x08local_ip\x18\x07\x20\x01(\x07R\x07localIp\
-    \x12\x1d\n\nlocal_port\x18\x08\x20\x01(\rR\tlocalPort*'\n\x0cPunchNatTyp\
-    e\x12\r\n\tSymmetric\x10\0\x12\x08\n\x04Cone\x10\x01b\x06proto3\
+    \n\rmessage.proto\"\x8f\x01\n\x13RegistrationRequest\x12\x14\n\x05token\
+    \x18\x01\x20\x01(\tR\x05token\x12\x1b\n\tdevice_id\x18\x02\x20\x01(\tR\
+    \x08deviceId\x12\x12\n\x04name\x18\x03\x20\x01(\tR\x04name\x12\x17\n\x07\
+    is_fast\x18\x04\x20\x01(\x08R\x06isFast\x12\x18\n\x07version\x18\x05\x20\
+    \x01(\tR\x07version\"\x92\x02\n\x14RegistrationResponse\x12\x1d\n\nvirtu\
+    al_ip\x18\x01\x20\x01(\x07R\tvirtualIp\x12'\n\x0fvirtual_gateway\x18\x02\
+    \x20\x01(\x07R\x0evirtualGateway\x12'\n\x0fvirtual_netmask\x18\x03\x20\
+    \x01(\x07R\x0evirtualNetmask\x12\x14\n\x05epoch\x18\x04\x20\x01(\rR\x05e\
+    poch\x125\n\x10device_info_list\x18\x05\x20\x03(\x0b2\x0b.DeviceInfoR\
+    \x0edeviceInfoList\x12\x1b\n\tpublic_ip\x18\x06\x20\x01(\x07R\x08publicI\
+    p\x12\x1f\n\x0bpublic_port\x18\x07\x20\x01(\rR\npublicPort\"d\n\nDeviceI\
+    nfo\x12\x12\n\x04name\x18\x01\x20\x01(\tR\x04name\x12\x1d\n\nvirtual_ip\
+    \x18\x02\x20\x01(\x07R\tvirtualIp\x12#\n\rdevice_status\x18\x03\x20\x01(\
+    \rR\x0cdeviceStatus\"Y\n\nDeviceList\x12\x14\n\x05epoch\x18\x01\x20\x01(\
+    \rR\x05epoch\x125\n\x10device_info_list\x18\x02\x20\x03(\x0b2\x0b.Device\
+    InfoR\x0edeviceInfoList\"\xf8\x01\n\tPunchInfo\x12$\n\x0epublic_ip_list\
+    \x18\x02\x20\x03(\x07R\x0cpublicIpList\x12\x1f\n\x0bpublic_port\x18\x03\
+    \x20\x01(\rR\npublicPort\x12*\n\x11public_port_range\x18\x04\x20\x01(\rR\
+    \x0fpublicPortRange\x12(\n\x08nat_type\x18\x05\x20\x01(\x0e2\r.PunchNatT\
+    ypeR\x07natType\x12\x14\n\x05reply\x18\x06\x20\x01(\x08R\x05reply\x12\
+    \x19\n\x08local_ip\x18\x07\x20\x01(\x07R\x07localIp\x12\x1d\n\nlocal_por\
+    t\x18\x08\x20\x01(\rR\tlocalPort*'\n\x0cPunchNatType\x12\r\n\tSymmetric\
+    \x10\0\x12\x08\n\x04Cone\x10\x01b\x06proto3\
 ";
 
 /// `FileDescriptorProto` object which was a source for this generated file

switch/src/protocol/control_packet.rs

@@ -17,7 +17,7 @@ pub enum Protocol {
     PunchRequest,
     /// 打洞响应
     PunchResponse,
-    UnKnow(u8),
+    Unknown(u8),
 }
 
 impl From<u8> for Protocol {
@@ -27,7 +27,7 @@ impl From<u8> for Protocol {
             2 => Protocol::Pong,
             3 => Protocol::PunchRequest,
             4 => Protocol::PunchResponse,
-            val => Protocol::UnKnow(val),
+            val => Protocol::Unknown(val),
         }
     }
 }
@@ -39,7 +39,7 @@ impl Into<u8> for Protocol {
             Protocol::Pong => 2,
             Protocol::PunchRequest => 3,
             Protocol::PunchResponse => 4,
-            Protocol::UnKnow(val) => val,
+            Protocol::Unknown(val) => val,
         }
     }
 }
@@ -58,18 +58,17 @@ impl<B: AsRef<[u8]>> ControlPacket<B> {
             Protocol::Pong => Ok(ControlPacket::PongPacket(PongPacket::new(buffer)?)),
             Protocol::PunchRequest => Ok(ControlPacket::PunchRequest),
             Protocol::PunchResponse => Ok(ControlPacket::PunchResponse),
-            Protocol::UnKnow(_) => Err(io::Error::new(io::ErrorKind::InvalidData, "Unsupported")),
+            Protocol::Unknown(_) => Err(io::Error::new(io::ErrorKind::InvalidData, "Unsupported")),
         }
     }
 }
 
 /// 网络探针
-#[derive(Copy, Clone)]
 pub struct PingPacket<B> {
     buffer: B,
 }
 
-type PongPacket<B> = PingPacket<B>;
+pub type PongPacket<B> = PingPacket<B>;
 
 impl<B: AsRef<[u8]>> PingPacket<B> {
     pub fn new(buffer: B) -> io::Result<PingPacket<B>> {

switch/src/protocol/ip_turn_packet.rs

@@ -0,0 +1,99 @@
+use std::io;
+use std::net::Ipv4Addr;
+
+#[derive(Copy, Clone, Eq, PartialEq, Debug)]
+pub enum Protocol {
+    Ipv4,
+    Ipv4Broadcast,
+    Unknown(u8),
+}
+
+impl From<u8> for Protocol {
+    fn from(value: u8) -> Self {
+        match value {
+            4 => Protocol::Ipv4,
+            201 => Protocol::Ipv4Broadcast,
+            val => Protocol::Unknown(val),
+        }
+    }
+}
+
+impl Into<u8> for Protocol {
+    fn into(self) -> u8 {
+        match self {
+            Protocol::Ipv4 => 4,
+            Protocol::Ipv4Broadcast => 201,
+            Protocol::Unknown(val) => val,
+        }
+    }
+}
+
+pub struct BroadcastPacketEnd<B> {
+    buffer: B,
+}
+
+impl<B: AsRef<[u8]>> BroadcastPacketEnd<B> {
+    pub fn unchecked(buffer: B) -> Self {
+        Self { buffer }
+    }
+    pub fn new(buffer: B) -> io::Result<Self> {
+        let len = buffer.as_ref().len();
+        let packet = Self::unchecked(buffer);
+        if len < 1 || packet.len() != len {
+            Err(io::Error::new(
+                io::ErrorKind::InvalidData,
+                "InvalidData",
+            ))
+        } else {
+            Ok(packet)
+        }
+    }
+}
+
+impl<B: AsRef<[u8]>> BroadcastPacketEnd<B> {
+    pub fn len(&self) -> usize {
+        1 + self.num() as usize * 4
+    }
+    pub fn num(&self) -> u8 {
+        let len = self.buffer.as_ref().len();
+        self.buffer.as_ref()[len - 1]
+    }
+    /// 已经发送给了这些地址
+    /// 从尾往头拿
+    pub fn addresses(&self) -> Vec<Ipv4Addr> {
+        let num = self.num() as usize;
+        let mut list = Vec::with_capacity(num);
+        let buf = self.buffer.as_ref();
+        let mut offset = buf.len() + 4 - 2;
+        for _ in 0..num {
+            offset -= 4;
+            list.push(Ipv4Addr::new(buf[offset - 3], buf[offset - 2], buf[offset - 1], buf[offset]));
+        }
+        list
+    }
+}
+
+impl<B: AsRef<[u8]> + AsMut<[u8]>> BroadcastPacketEnd<B> {
+    /// 从头往尾放
+    pub fn set_address(&mut self, addr: &[Ipv4Addr]) -> io::Result<()> {
+        let buf = self.buffer.as_mut();
+        if buf.len() < 1 + addr.len() * 4 || addr.len() > u8::MAX as usize {
+            Err(io::Error::new(
+                io::ErrorKind::InvalidData,
+                "InvalidData",
+            ))
+        } else {
+            let mut offset = 0;
+            for ip in addr {
+                buf[offset..offset + 4].copy_from_slice(&ip.octets());
+                offset += 4;
+            }
+            self.buffer.as_mut()[offset] = addr.len() as u8;
+            Ok(())
+        }
+    }
+}
+
+
+
+

switch/src/protocol/mod.rs

@@ -18,7 +18,8 @@ use std::{fmt, io};
 pub mod control_packet;
 pub mod error_packet;
 pub mod service_packet;
-pub mod turn_packet;
+pub mod ip_turn_packet;
+pub mod other_turn_packet;
 
 #[derive(Eq, PartialEq, Copy, Clone, Debug)]
 pub enum Version {
@@ -52,8 +53,8 @@ pub enum Protocol {
     Error,
     /// 控制协议
     Control,
-    /// 转发ipv4数据
-    Ipv4Turn,
+    /// 转发ip数据
+    IpTurn,
     /// 转发其他数据
     OtherTurn,
     UnKnow(u8),
@@ -65,7 +66,7 @@ impl From<u8> for Protocol {
             1 => Protocol::Service,
             2 => Protocol::Error,
             3 => Protocol::Control,
-            4 => Protocol::Ipv4Turn,
+            4 => Protocol::IpTurn,
             5 => Protocol::OtherTurn,
             val => Protocol::UnKnow(val),
         }
@@ -78,7 +79,7 @@ impl Into<u8> for Protocol {
             Protocol::Service => 1,
             Protocol::Error => 2,
             Protocol::Control => 3,
-            Protocol::Ipv4Turn => 4,
+            Protocol::IpTurn => 4,
             Protocol::OtherTurn => 5,
             Protocol::UnKnow(val) => val,
         }
@@ -143,6 +144,9 @@ impl<B: AsRef<[u8]>> NetPacket<B> {
 }
 
 impl<B: AsRef<[u8]> + AsMut<[u8]>> NetPacket<B> {
+    pub fn buffer_mut(&mut self)->&mut [u8]{
+        self.buffer.as_mut()
+    }
     pub fn set_version(&mut self, version: Version) {
         self.buffer.as_mut()[0] = version.into();
     }

switch/src/protocol/other_turn_packet.rs

@@ -1,14 +1,14 @@
 #[derive(Copy, Clone, Eq, PartialEq, Debug)]
 pub enum Protocol {
     Punch,
-    UnKnow(u8),
+    Unknown(u8),
 }
 
 impl From<u8> for Protocol {
     fn from(value: u8) -> Self {
         match value {
             1 => Protocol::Punch,
-            val => Protocol::UnKnow(val),
+            val => Protocol::Unknown(val),
         }
     }
 }
@@ -17,7 +17,7 @@ impl Into<u8> for Protocol {
     fn into(self) -> u8 {
         match self {
             Protocol::Punch => 1,
-            Protocol::UnKnow(val) => val,
+            Protocol::Unknown(val) => val,
         }
     }
 }

switch/src/protocol/service_packet.rs

@@ -8,7 +8,7 @@ pub enum Protocol {
     PollDeviceList,
     /// 推送设备列表
     PushDeviceList,
-    UnKnow(u8),
+    Unknown(u8),
 }
 
 impl From<u8> for Protocol {
@@ -18,7 +18,7 @@ impl From<u8> for Protocol {
             2 => Self::RegistrationResponse,
             3 => Self::PollDeviceList,
             4 => Self::PushDeviceList,
-            val => Self::UnKnow(val),
+            val => Self::Unknown(val),
         }
     }
 }
@@ -30,7 +30,7 @@ impl Into<u8> for Protocol {
             Self::RegistrationResponse => 2,
             Self::PollDeviceList => 3,
             Self::PushDeviceList => 4,
-            Self::UnKnow(val) => val,
+            Self::Unknown(val) => val,
         }
     }
 }