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

注册失败时重试

Browse Source
This commit is contained in:
lbl
2026-02-24 16:08:18 +08:00
parent b4301a8106
commit 4708ed2b16
4 changed files with 90 additions and 62 deletions
Download Patch File Download Diff File Expand all files Collapse all files
vnt-core/src/core/mod.rs
+29 -31
View File
@@ -9,6 +9,7 @@ use crate::enhanced_tunnel::outbound::EnhancedOutbound;
use crate::fec::{FecDecoder, FecEncoder};
use crate::nat::internal_nat::{InternalNatInbound, PortMappingManager};
use crate::nat::{AllowSubnetExternalRoute, SubnetExternalRoute};
use crate::protocol::control_message::ErrorResponseMsg;
use crate::tun::enhanced_tun::EnhancedTunInbound;
use crate::tun::{DeviceConfig, DeviceIOManager, TunDataInbound, TunReceiver, tun_channel};
use crate::tunnel_core::outbound::{BasicOutbound, HybridOutbound};
@@ -47,6 +48,10 @@ pub struct NetworkManager {
tun_receiver: Option<TunReceiver>,
registration_context: Option<Box<RegistrationContext>>,
}
pub enum RegisterResponse {
Success(NetworkAddr),
Failed(ErrorResponseMsg),
}
impl NetworkManager {
pub async fn create_network(
@@ -202,47 +207,41 @@ impl NetworkManager {
/// Register with server(s) and start data handling tasks.
/// This method can only be called once.
/// Returns the registration response on success.
pub async fn register(&mut self) -> anyhow::Result<NetworkAddr> {
pub async fn register(&mut self) -> anyhow::Result<RegisterResponse> {
let Some(mut ctx) = self.registration_context.take() else {
bail!("register can only be called once");
};
let is_multi_server = ctx.server_managers.len() > 1;
let reg_response = if is_multi_server {
let response = if is_multi_server {
// Multi-server: coordinated pre-registration
log::info!(
"Multi-server mode: performing coordinated registration for {} servers",
ctx.server_managers.len()
);
let reg_response = coordinated_registration(&mut ctx.server_managers).await?;
log::info!(
"Coordinated registration completed, IP: {}, prefix_len: {}",
reg_response.ip,
reg_response.prefix_len
);
reg_response
coordinated_registration(&mut ctx.server_managers).await?
} else {
// Single-server: normal registration
log::info!("Single-server mode: performing normal registration");
let response = ctx.server_managers[0]
ctx.server_managers[0]
.connect_and_reg(crate::protocol::control_message::RegistrationMode::Normal)
.await?;
match response {
crate::protocol::control_message::ResponseMessage::Reg(reg) => {
log::info!(
"Registration completed, IP: {}, prefix_len: {}",
reg.ip,
reg.prefix_len
);
reg
}
crate::protocol::control_message::ResponseMessage::Error(e) => {
bail!("Registration failed: {}", e.message);
}
crate::protocol::control_message::ResponseMessage::ConfirmReg(_) => {
bail!("Unexpected ConfirmReg response");
}
.await?
};
let reg_response = match response {
crate::protocol::control_message::ResponseMessage::Reg(reg) => {
log::info!(
"Registration completed, IP: {}, prefix_len: {}",
reg.ip,
reg.prefix_len
);
reg
}
crate::protocol::control_message::ResponseMessage::Error(e) => {
return Ok(RegisterResponse::Failed(e));
}
crate::protocol::control_message::ResponseMessage::ConfirmReg(_) => {
bail!("Unexpected ConfirmReg response");
}
};
let network_addr = NetworkAddr {
@@ -256,10 +255,9 @@ impl NetworkManager {
// 保存服务器版本信息
if !reg_response.server_version.is_empty() {
for (index, _) in ctx.server_managers.iter().enumerate() {
self.app_state.server_info_collection.set_server_version(
index as u32,
reg_response.server_version.clone(),
);
self.app_state
.server_info_collection
.set_server_version(index as u32, reg_response.server_version.clone());
}
}
@@ -283,7 +281,7 @@ impl NetworkManager {
turn_manager.data_handle_task_connected(&self.task_group, handler_config, network_addr);
}
Ok(network_addr)
Ok(RegisterResponse::Success(network_addr))
}
pub fn is_no_tun(&self) -> bool {
vnt-core/src/tunnel_core/server/connection_manager.rs
+12 -17
View File
@@ -6,7 +6,7 @@ use crate::crypto::PacketCrypto;
use crate::enhanced_tunnel::inbound::EnhancedInbound;
use crate::fec::FecDecoder;
use crate::protocol::control_message::{
ConfirmRegResponseMsg, RegResponseMsg, RegistrationMode, RequestMessage, ResponseMessage,
ConfirmRegResponseMsg, RegistrationMode, RequestMessage, ResponseMessage,
};
use crate::tunnel_core::p2p::transport::punch::NatPuncher;
use crate::tunnel_core::server::inbound::ServerTurnInboundHandler;
@@ -117,9 +117,7 @@ impl ServerTurnManager {
let request_msg = RequestMessage::Reg(reg_msg);
let encoded = request_msg.encode();
self.transport_client
.send(encoded.freeze())
.await?;
self.transport_client.send(encoded.freeze()).await?;
let buf = self
.transport_client
.next_timeout(Duration::from_secs(10))
@@ -164,8 +162,7 @@ impl ServerTurnManager {
config: Box<InboundHandlerConfig>,
initial_response: NetworkAddr,
) {
let data_handler =
ServerTurnInboundHandler::new(self.server_id, initial_response, config);
let data_handler = ServerTurnInboundHandler::new(self.server_id, initial_response, config);
let task_group_ = task_group.clone();
let Some(mut receiver) = self.receiver.take() else {
unreachable!()
@@ -212,10 +209,7 @@ impl ServerTurnManager {
log::info!("已连接服务器:{}", self.config.server_addr);
data_handler.handle_connected();
if let Err(e) = self
.data_handle_loop(&mut receiver, &data_handler)
.await
{
if let Err(e) = self.data_handle_loop(&mut receiver, &data_handler).await {
log::error!("Error on data_handle_loop: {:?}", e);
}
already_connected = false;
@@ -271,7 +265,7 @@ impl ServerTurnManager {
/// 4. Return the registration response
pub async fn coordinated_registration(
managers: &mut Vec<ServerTurnManager>,
) -> anyhow::Result<RegResponseMsg> {
) -> anyhow::Result<ResponseMessage> {
if managers.is_empty() {
bail!("No servers to register");
}
@@ -287,7 +281,10 @@ pub async fn coordinated_registration(
let ip = match &first_response {
ResponseMessage::Reg(reg) => reg.ip,
ResponseMessage::Error(e) => bail!("First server registration failed: {}", e.message),
ResponseMessage::Error(e) => {
log::info!("First server registration failed: {}", e.message);
return Ok(first_response);
}
_ => bail!("Unexpected response from first server"),
};
log::info!("Got IP {} from first server", ip);
@@ -313,7 +310,8 @@ pub async fn coordinated_registration(
log::info!("Server {} pre-registered successfully", i + 1);
}
Ok(ResponseMessage::Error(e)) => {
bail!("Server {} registration failed: {}", i + 1, e.message)
log::info!("Server {} registration failed: {}", i + 1, e.message);
return Ok(ResponseMessage::Error(e.clone()));
}
Err(e) => bail!("Server {} registration failed: {}", i + 1, e),
_ => bail!("Unexpected response from server {}", i + 1),
@@ -339,8 +337,5 @@ pub async fn coordinated_registration(
log::info!("Coordinated registration completed successfully");
// Return first server's response (contains IP info)
match first_response {
ResponseMessage::Reg(reg) => Ok(reg),
_ => unreachable!(),
}
Ok(first_response)
}