yeyinlaoying/vnt
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

1.增加设备名称和状态

Browse Source 
2.测试windows服务
This commit is contained in:
lubeilin
2023-01-15 19:15:41 +08:00
parent 2232965a1d
commit 9820c56be6
15 changed files with 767 additions and 186 deletions
Download Patch File Download Diff File Expand all files Collapse all files
switch/src/handle/heartbeat_handler.rs
+1 -1
View File
@@ -23,7 +23,7 @@ pub async fn start<F>(
match handle_loop(status_watch, udp, cur_info.connect_server).await {
Ok(_) => {}
Err(e) => {
log::error!("{:?}", e)
log::warn!("{:?}", e)
}
}
stop_fn();
switch/src/handle/mod.rs
+44 -1
View File
@@ -19,7 +19,7 @@ lazy_static! {
/// 0. 机器纪元，每一次上线或者下线都会增1，由服务端维护，用于感知网络中机器变化
/// 服务端和客户端的不一致，则服务端会推送新的设备列表
/// 1. 网络中的虚拟ip列表
pub static ref DEVICE_LIST:Mutex<(u32,Vec<Ipv4Addr>)> = const_mutex((0,Vec::new()));
pub static ref DEVICE_LIST:Mutex<(u32,Vec<PeerDeviceInfo>)> = const_mutex((0,Vec::new()));
/// 服务器延迟
pub static ref SERVER_RT:AtomicI64 = AtomicI64::new(-1);
/// id
@@ -32,6 +32,49 @@ lazy_static! {
/// 当前设备的nat信息
pub static ref NAT_INFO:Mutex<Option<NatInfo>> = const_mutex(None);
}
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct PeerDeviceInfo {
pub virtual_ip: Ipv4Addr,
pub name: String,
pub status: PeerDeviceStatus,
}
impl PeerDeviceInfo {
pub fn new(virtual_ip: Ipv4Addr,
name: String,
status: u8, ) -> Self {
Self {
virtual_ip,
name,
status: PeerDeviceStatus::from(status),
}
}
}
#[derive(Copy, Clone, Debug, Eq, PartialEq)]
pub enum PeerDeviceStatus {
Online,
Offline,
}
impl Into<u8> for PeerDeviceStatus {
fn into(self) -> u8 {
match self {
PeerDeviceStatus::Online => 0,
PeerDeviceStatus::Offline => 1,
}
}
}
impl From<u8> for PeerDeviceStatus {
fn from(value: u8) -> Self {
match value {
0 => PeerDeviceStatus::Online,
_ => PeerDeviceStatus::Offline
}
}
}
#[derive(Copy, Clone, Debug, Eq, PartialEq)]
pub enum ApplicationStatus {
Starting,
switch/src/handle/punch_handler.rs
+11 -10
View File
@@ -5,17 +5,17 @@ use std::time::Duration;
use dashmap::DashMap;
use lazy_static::lazy_static;
use protobuf::Message;
use tokio::sync::mpsc::error::TrySendError;
use tokio::sync::mpsc::{Receiver, Sender};
use tokio::sync::mpsc::error::TrySendError;
use tokio::sync::watch;
use crate::{CurrentDeviceInfo, DEVICE_LIST, handle::NAT_INFO, handle::NatInfo};
use crate::error::*;
use crate::handle::{ApplicationStatus, DIRECT_ROUTE_TABLE};
use crate::proto::message::{NatType, Punch, Step};
use crate::protocol::{control_packet, NetPacket, Protocol, turn_packet, Version};
use crate::protocol::control_packet::PunchRequestPacket;
use crate::protocol::turn_packet::TurnPacket;
use crate::protocol::{control_packet, turn_packet, NetPacket, Protocol, Version};
use crate::{handle::NatInfo, handle::NAT_INFO, CurrentDeviceInfo, DEVICE_LIST};
lazy_static! {
pub static ref STEP_MAP: DashMap<Ipv4Addr, Step> = DashMap::new();
@@ -193,7 +193,7 @@ pub async fn req_symmetric_handler_start<F>(
match handle_loop(status_watch, receiver, udp, cur_info).await {
Ok(_) => {}
Err(e) => {
log::error!("{:?}", e)
log::warn!("{:?}", e)
}
}
stop_fn()
@@ -224,7 +224,7 @@ pub async fn res_symmetric_handler_start<F>(
match res_symmetric_handle_loop(status_watch, receiver, udp, cur_info).await {
Ok(_) => {}
Err(e) => {
log::error!("{:?}", e)
log::warn!("{:?}", e)
}
}
stop_fn()
@@ -287,7 +287,7 @@ async fn res_symmetric_handle_loop(
}
}
if let Err(e) = handle(&status_watch,&udp, list, packet.buffer()) {
log::error!("{:?}",e)
log::warn!("{:?}",e)
}
}else {
return Err(Error::Stop("打洞线程通道关闭".to_string()));
@@ -323,7 +323,7 @@ pub async fn cone_handler_start<F>(
match handle_loop(status_watch, receiver, udp, cur_info).await {
Ok(_) => {}
Err(e) => {
log::error!("{:?}", e)
log::warn!("{:?}", e)
}
}
stop_fn();
@@ -363,7 +363,7 @@ async fn handle_loop(
}
}
if let Err(e) = handle(&status_watch,&udp, list, packet.buffer()) {
log::error!("{:?}",e)
log::warn!("{:?}",e)
}
}else {
return Err(Error::Stop("打洞线程通道关闭".to_string()));
@@ -390,7 +390,7 @@ fn punch_request_handle(udp: &UdpSocket, cur_info: &CurrentDeviceInfo) -> Result
drop(nat_info_lock);
if let Some(nat_info) = nat_info {
if let Err(e) = send_punch(&udp, &cur_info, nat_info) {
log::error!("发送打洞数据失败 {:?}", e)
log::warn!("发送打洞数据失败 {:?}", e)
}
Ok(())
} else {
@@ -402,7 +402,8 @@ fn send_punch(udp: &UdpSocket, cur_info: &CurrentDeviceInfo, nat_info: NatInfo)
let lock = DEVICE_LIST.lock();
let list = lock.1.clone();
drop(lock);
for ip in list {
for peer_info in list {
let ip = peer_info.virtual_ip;
//只向ip比自己大的发起打洞，避免双方同时发起打洞浪费流量
if ip > cur_info.virtual_ip && !DIRECT_ROUTE_TABLE.contains_key(&ip) {
let step = if let Some(step) = STEP_MAP.get(&ip) {
switch/src/handle/registration_handler.rs
+40 -15
View File
@@ -11,10 +11,11 @@ use protobuf::Message;
use crate::error::*;
use crate::handle::ConnectStatus;
use crate::proto::message::{RegistrationRequest, RegistrationResponse};
use crate::protocol::{error_packet, service_packet, NetPacket, Protocol, Version};
use crate::protocol::{error_packet, NetPacket, Protocol, service_packet, Version};
use crate::protocol::error_packet::InErrorPacket;
lazy_static::lazy_static! {
static ref REQUEST:RwLock<Option<(String,String)>> = parking_lot::const_rwlock(None);
static ref REQUEST:RwLock<Option<(String,String,String)>> = parking_lot::const_rwlock(None);
static ref REGISTRATION_TIME:AtomicI64=AtomicI64::new(0);
pub(crate) static ref CONNECTION_STATUS:AtomicCell<ConnectStatus> = AtomicCell::new(ConnectStatus::Connecting);
}
@@ -25,9 +26,10 @@ pub fn registration(
server_address: SocketAddr,
token: String,
mac_address: String,
name: String,
) -> Result<RegistrationResponse> {
// todo 和服务器通信加密
let request_packet = registration_request_packet(token.clone(), mac_address.clone())?;
let request_packet = registration_request_packet(token.clone(), mac_address.clone(), name.clone(), false)?;
let buf = request_packet.buffer();
let mut counter = 0;
let mut recv_buf = [0u8; 10240];
@@ -57,7 +59,7 @@ pub fn registration(
service_packet::Protocol::RegistrationResponse => {
let response =
RegistrationResponse::parse_from_bytes(net_packet.payload())?;
let _ = REQUEST.write().replace((token, mac_address));
let _ = REQUEST.write().replace((token, mac_address, name));
udp.set_read_timeout(None)?;
CONNECTION_STATUS.store(ConnectStatus::Connected);
return Ok(response);
@@ -66,22 +68,45 @@ pub fn registration(
}
}
Protocol::Error => {
match error_packet::Protocol::from(net_packet.transport_protocol()) {
error_packet::Protocol::TokenError => {
return Err(Error::Stop("token错误".to_string()));
return match InErrorPacket::new(net_packet.transport_protocol(), net_packet.payload()) {
Ok(e) => {
match e {
InErrorPacket::TokenError => {
Err(Error::Stop("token错误".to_string()))
}
InErrorPacket::Disconnect => {
Err(Error::Stop("断开连接".to_string()))
}
InErrorPacket::OtherError(e) => {
match e.message() {
Ok(str) => {
Err(Error::Stop(str))
}
Err(e) => {
Err(Error::Stop(format!("{:?}", e)))
}
}
}
}
}
_ => {}
}
Err(e) => {
Err(Error::Stop(format!("{:?}", e)))
}
};
}
_ => {
return Err(Error::Stop(format!("数据错误：{:?}", net_packet)));
}
_ => {}
}
}
}
fn registration_request_packet(token: String, mac_address: String) -> Result<NetPacket<Vec<u8>>> {
fn registration_request_packet(token: String, mac_address: String, name: String, is_fast: bool) -> Result<NetPacket<Vec<u8>>> {
let mut request = RegistrationRequest::new();
request.token = token;
request.mac_address = mac_address;
request.name = name;
request.is_fast = is_fast;
let bytes = request.write_to_bytes()?;
let buf = vec![0u8; 4 + bytes.len()];
let mut net_packet = NetPacket::new(buf)?;
@@ -98,8 +123,8 @@ pub fn fast_registration(udp: &UdpSocket, server_address: SocketAddr) -> Result<
let new = Local::now().timestamp_millis();
if new - last < 2000
|| REGISTRATION_TIME
.compare_exchange(last, new, Ordering::Relaxed, Ordering::Relaxed)
.is_err()
.compare_exchange(last, new, Ordering::Relaxed, Ordering::Relaxed)
.is_err()
{
//短时间不重复注册
return Ok(());
@@ -108,8 +133,8 @@ pub fn fast_registration(udp: &UdpSocket, server_address: SocketAddr) -> Result<
let lock = REQUEST.read();
let option = lock.clone();
drop(lock);
if let Some((token, mac_address)) = option {
let request_packet = registration_request_packet(token, mac_address)?;
if let Some((token, mac_address, name)) = option {
let request_packet = registration_request_packet(token, mac_address, name, true)?;
udp.send_to(request_packet.buffer(), server_address)?;
REGISTRATION_TIME.store(Local::now().timestamp_millis(), Ordering::Relaxed);
return Ok(());
switch/src/handle/tun_handler.rs
+6 -6
View File
@@ -10,12 +10,12 @@ use packet::icmp::Kind;
use packet::ip::ipv4;
use packet::ip::ipv4::packet::IpV4Packet;
use crate::ApplicationStatus;
use crate::error::*;
use crate::handle::{CurrentDeviceInfo, DIRECT_ROUTE_TABLE};
use crate::protocol::turn_packet::TurnPacket;
use crate::protocol::{NetPacket, Protocol, Version};
use crate::protocol::turn_packet::TurnPacket;
use crate::tun_device::TunReader;
use crate::ApplicationStatus;
/// 是否在一个网段
fn check_dest(dest: Ipv4Addr, cur_info: &CurrentDeviceInfo) -> bool {
@@ -114,7 +114,7 @@ pub async fn handler_start<F>(
});
thread::spawn(move || {
if let Err(e) = handle_loop(udp, tun_reader, cur_info) {
log::error!("tun数据处理线程停止 {:?}", e);
log::warn!("tun数据处理线程停止 {:?}", e);
}
stop_fn();
});
@@ -132,7 +132,7 @@ fn handle_loop(udp: UdpSocket, tun_reader: TunReader, cur_info: CurrentDeviceInf
match handle(&udp, data.bytes_mut(), &cur_info, &mut net_packet) {
Ok(_) => {}
Err(e) => {
println!("{:?}", e)
log::warn!("{:?}", e)
}
}
}
@@ -164,7 +164,7 @@ pub async fn handler_start<F>(
});
thread::spawn(move || {
if let Err(e) = handle_loop(udp, tun_reader, cur_info) {
log::error!(" tun数据处理线程停止 {:?}", e);
log::warn!(" tun数据处理线程停止 {:?}", e);
}
stop_fn();
});
@@ -187,7 +187,7 @@ pub fn handle_loop(
match handle(&udp, data, &cur_info, &mut net_packet) {
Ok(_) => {}
Err(e) => {
log::error!("{:?}", e)
log::warn!("{:?}", e)
}
}
}
switch/src/handle/udp_recv_handler.rs
+13 -15
View File
@@ -23,7 +23,7 @@ use crate::protocol::error_packet::InErrorPacket;
use crate::protocol::turn_packet::TurnPacket;
use crate::protocol::{control_packet, service_packet, turn_packet, NetPacket, Protocol, Version};
use crate::tun_device::TunWriter;
use crate::{ApplicationStatus, CurrentDeviceInfo};
use crate::{ApplicationStatus, CurrentDeviceInfo, PeerDeviceInfo};
const UDP_STOP_BUF: [u8; 1] = [0u8];
@@ -50,7 +50,7 @@ pub async fn udp_recv_start<F>(
thread::spawn(move || {
if let Err(e) = recv_loop(udp, server_addr, other_sender, tun_writer, current_device) {
log::error!("udp数据处理线程停止 {:?}", e);
log::warn!("udp数据处理线程停止 {:?}", e);
}
stop_fn();
});
@@ -95,12 +95,12 @@ fn recv_loop(
return Err(Error::Stop(str));
}
Err(e) => {
log::error!("{:?}", e);
log::warn!("{:?}", e);
}
}
}
Err(e) => {
log::error!("{:?}", e);
log::warn!("{:?}", e);
}
};
}
@@ -156,7 +156,7 @@ fn recv_handle(
return Err(Error::Stop("子处理线程停止".to_string()));
}
Err(e) => {
log::error!("子线程处理 {:?}", e);
log::warn!("子线程处理 {:?}", e);
}
}
}
@@ -178,7 +178,7 @@ pub async fn udp_other_recv_start<F>(
match other_loop(status_watch, udp, receiver, current_device, sender).await {
Ok(_) => {}
Err(e) => {
log::error!("{:?}", e);
log::warn!("{:?}", e);
}
}
stop_fn();
@@ -202,7 +202,7 @@ async fn other_loop(
return Err(Error::Stop(str));
}
Err(e) => {
log::error!("other_loop {:?}",e);
log::warn!("other_loop {:?}",e);
}
}
}
@@ -241,10 +241,10 @@ fn other_handle(
}
service_packet::Protocol::UpdateDeviceList => {
let device_list = DeviceList::parse_from_bytes(net_packet.payload())?;
let ip_list: Vec<Ipv4Addr> = device_list
.virtual_ip_list
.iter()
.map(|ip| Ipv4Addr::from(*ip))
let ip_list = device_list
.device_info_list
.into_iter()
.map(|info| PeerDeviceInfo::new(Ipv4Addr::from(info.virtual_ip), info.name, info.device_status as u8))
.collect();
let mut dev = DEVICE_LIST.lock();
if dev.0 < device_list.epoch || device_list.epoch - dev.0 > u32::MAX >> 2 {
@@ -269,7 +269,7 @@ fn other_handle(
}
}
InErrorPacket::OtherError(e) => {
log::error!("OtherError {:?}", e.message());
log::warn!("OtherError {:?}", e.message());
}
}
}
@@ -369,12 +369,10 @@ fn other_handle(
}
turn_packet::Protocol::UnKnow(_) => {}
}
} else {
panic!("ip")
}
}
Protocol::UnKnow(p) => {
log::error!("未知协议 {}", p);
log::warn!("未知协议 {}", p);
}
}
Ok(())