luckyimnotanigger/tun2proxy
1
0
Fork 0
mirror of https://github.com/tun2proxy/tun2proxy.git synced 2025-04-19 13:29:09 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 5

read code

Browse source
This commit is contained in:
ssrlive 2024-10-23 19:49:20 +08:00
parent 1ec7caf844
commit 01b5951ec1
1 changed files with 27 additions and 30 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

57
src/lib.rs
View file

@ -490,10 +490,9 @@ async fn handle_udp_gateway_session(
ipv6_enabled: bool, ipv6_enabled: bool,
) -> crate::Result<()> { ) -> crate::Result<()> {
let proxy_server_addr = { proxy_handler.lock().await.get_server_addr() }; let proxy_server_addr = { proxy_handler.lock().await.get_server_addr() };
let udpinfo = SessionInfo::new(udp_stack.local_addr(), udp_dst, IpProtocol::Udp);
let udp_mtu = udpgw_client.get_udp_mtu(); let udp_mtu = udpgw_client.get_udp_mtu();
let udp_timeout = udpgw_client.get_udp_timeout(); let udp_timeout = udpgw_client.get_udp_timeout();
let mut server_stream = match udpgw_client.get_server_connection().await { let mut stream = match udpgw_client.get_server_connection().await {
Some(server) => server, Some(server) => server,
None => { None => {
if udpgw_client.is_full() { if udpgw_client.is_full() {
@ -507,91 +506,89 @@ async fn handle_udp_gateway_session(
} }
}; };
let tcp_local_addr = server_stream.local_addr().clone(); let tcp_local_addr = stream.local_addr().clone();
match domain_name { match domain_name {
Some(ref d) => log::info!("Beginning {} <- {}, domain:{}", udpinfo, &tcp_local_addr, d), Some(ref d) => log::info!("[UdpGw] Beginning {} -> {}, domain:{}", &tcp_local_addr, udp_dst, d),
None => log::info!("Beginning {} <- {}", udpinfo, &tcp_local_addr), None => log::info!("[UdpGw] Beginning {} -> {}", &tcp_local_addr, udp_dst),
} }
let Some(mut stream_reader) = server_stream.get_reader() else { let Some(mut reader) = stream.get_reader() else {
return Err("get reader failed".into()); return Err("get reader failed".into());
}; };
let Some(mut stream_writer) = server_stream.get_writer() else { let Some(mut writer) = stream.get_writer() else {
return Err("get writer failed".into()); return Err("get writer failed".into());
}; };
loop { loop {
tokio::select! { tokio::select! {
len = UdpGwClient::recv_udp_packet(&mut udp_stack, &mut stream_writer) => { len = UdpGwClient::recv_udp_packet(&mut udp_stack, &mut writer) => {
let read_len; let read_len;
match len { match len {
Ok(n) => { Ok(n) => {
if n == 0 { if n == 0 {
log::info!("Ending {} <- {}",udpinfo, &tcp_local_addr); log::info!("[UdpGw] Ending {} <> {}", &tcp_local_addr, udp_dst);
break; break;
} }
read_len = n; read_len = n;
crate::traffic_status::traffic_status_update(n, 0)?; crate::traffic_status::traffic_status_update(n, 0)?;
} }
Err(e) => { Err(e) => {
log::info!("Ending {} <- {} with recv_udp_packet {}", udpinfo, &tcp_local_addr, e); log::info!("[UdpGw] Ending {} <> {} with recv_udp_packet {}", &tcp_local_addr, udp_dst, e);
break; break;
} }
} }
let new_id = server_stream.new_id(); let new_id = stream.new_id();
if let Err(e) = UdpGwClient::send_udpgw_packet(ipv6_enabled, read_len, udp_dst, domain_name.as_ref(), new_id, &mut stream_writer).await { if let Err(e) = UdpGwClient::send_udpgw_packet(ipv6_enabled, read_len, udp_dst, domain_name.as_ref(), new_id, &mut writer).await {
log::info!("Ending {} <- {} with send_udpgw_packet {}", udpinfo, &tcp_local_addr, e); log::info!("[UdpGw] Ending {} <> {} with send_udpgw_packet {}", &tcp_local_addr, udp_dst, e);
break; break;
} }
log::debug!("{} <- {} send udpgw len {}", udpinfo, &tcp_local_addr, read_len); log::debug!("[UdpGw] {} -> {} send len {}", &tcp_local_addr, udp_dst, read_len);
server_stream.update_activity(); stream.update_activity();
} }
ret = UdpGwClient::recv_udpgw_packet(udp_mtu, udp_timeout, &mut stream_reader) => { ret = UdpGwClient::recv_udpgw_packet(udp_mtu, udp_timeout, &mut reader) => {
match ret { match ret {
Ok(packet) => match packet { Ok(packet) => match packet {
//should not received keepalive //should not received keepalive
UdpGwResponse::KeepAlive => { UdpGwResponse::KeepAlive => {
log::error!("Ending {} <- {} with recv keepalive", udpinfo, &tcp_local_addr); log::error!("[UdpGw] Ending {} <> {} with recv keepalive", &tcp_local_addr, udp_dst);
server_stream.close(); stream.close();
break; break;
} }
//server udp may be timeout,can continue to receive udp data? //server udp may be timeout,can continue to receive udp data?
UdpGwResponse::Error => { UdpGwResponse::Error => {
log::info!("Ending {} <- {} with recv udp error", udpinfo, &tcp_local_addr); log::info!("[UdpGw] Ending {} <> {} with recv udp error", &tcp_local_addr, udp_dst);
server_stream.update_activity(); stream.update_activity();
continue; continue;
} }
UdpGwResponse::TcpClose => { UdpGwResponse::TcpClose => {
log::error!("Ending {} <- {} with tcp closed", udpinfo, &tcp_local_addr); log::error!("[UdpGw] Ending {} <> {} with tcp closed", &tcp_local_addr, udp_dst);
server_stream.close(); stream.close();
break; break;
} }
UdpGwResponse::Data(data) => { UdpGwResponse::Data(data) => {
let len = data.len(); let len = data.len();
log::debug!("{} <- {} receive udpgw len {}", udpinfo, &tcp_local_addr,len); log::debug!("[UdpGw] {} <- {} receive len {}", &tcp_local_addr, udp_dst, len);
if let Err(e) = UdpGwClient::send_udp_packet(data, &mut udp_stack).await { if let Err(e) = UdpGwClient::send_udp_packet(data, &mut udp_stack).await {
log::error!("Ending {} <- {} with send_udp_packet {}", udpinfo, &tcp_local_addr, e); log::error!("[UdpGw] Ending {} <> {} with send_udp_packet {}", &tcp_local_addr, udp_dst, e);
break; break;
} }
crate::traffic_status::traffic_status_update(0, len)?; crate::traffic_status::traffic_status_update(0, len)?;
} }
}, },
Err(e) => { Err(e) => {
log::warn!("Ending {} <- {} with recv_udpgw_packet {}", udpinfo, &tcp_local_addr, e); log::warn!("[UdpGw] Ending {} <> {} with recv_udpgw_packet {}", &tcp_local_addr, udp_dst, e);
break; break;
} }
} }
server_stream.update_activity(); stream.update_activity();
} }
} }
} }
if !server_stream.is_closed() { if !stream.is_closed() {
udpgw_client udpgw_client.release_server_connection_with_stream(stream, reader, writer).await;
.release_server_connection_with_stream(server_stream, stream_reader, stream_writer)
.await;
} }
Ok(()) Ok(())