nineap/netrunner-proxy
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

buffers update, warning fixes, ghosts killed

Browse Source
This commit is contained in:
Kirill
2026-04-03 18:18:10 +07:00
parent e53a95692e
commit 780750171b
29 changed files with 1205 additions and 1044 deletions
Download Patch File Download Diff File Expand all files Collapse all files
core/src/net/connection/bridge.rs
+2 -1
View File
@@ -2,7 +2,7 @@ use std::sync::Arc;
use std::time::Duration;
use crate::net::connection::muxer::{MuxMessage, Muxer};
use crate::net::network::NetworkConfig;
use crate::net::NetworkConfig;
use crate::nrxp::FrameType;
use bytes::{Bytes, BytesMut};
use netrunner_logger::{debug, error, info, warn};
@@ -42,6 +42,7 @@ pub(crate) async fn run_tcp_bridge<R, W>(
let mut buf = BytesMut::with_capacity(NetworkConfig::global().tcp_buffer_size);
loop {
buf.reserve(NetworkConfig::global().tcp_buffer_size);
let select_res = timeout(IDLE_TIMEOUT, async {
tokio::select! {
core/src/net/connection/connection.rs
+169 -100
View File
@@ -2,8 +2,12 @@ use std::{net::Ipv4Addr, sync::Arc, time::Duration};
use crate::{
net::{
connection::{engine::TunnelEngine, handler::StreamHandler, muxer::Muxer},
network::NetworkConfig,
connection::{
engine::TunnelEngine,
handler::{RemoteOpener, StreamHandler},
muxer::Muxer,
},
NetworkConfig,
},
nrxp::{Codec, ErrorAction, Frame, FrameType},
rawcast::{LocalProtocol, RawCastAdapter, RawCastFrame},
@@ -52,12 +56,6 @@ impl SessionManager {
}
}
#[derive(Clone, Copy, Debug, PartialEq)]
pub enum ConnectionRole {
Client,
Server,
}
#[async_trait::async_trait]
pub trait TunnelHandler {
async fn run(self) -> Result<(), String>;
@@ -91,7 +89,9 @@ impl Connection {
}
}
type StreamContext = (Ipv4Addr, u16, LocalProtocol);
// Измененный контекст: теперь храним оригинальный (локальный) socket_id
type StreamContext = (u32, Ipv4Addr, u16, LocalProtocol);
pub struct ClientHandler;
impl ClientHandler {
@@ -115,7 +115,6 @@ impl ClientHandler {
let stream = TcpStream::connect(remote_proxy_addr)
.await
.map_err(|e| format!("Failed to connect: {}", e))?;
let _ = stream.set_nodelay(true);
let (inbound, outbound) = stream.into_split();
let mut conn = Connection::new_raw(inbound, outbound);
@@ -170,8 +169,7 @@ impl ClientHandler {
muxer.add_leg(leg_id, control_tx, data_tx);
let handler =
std::sync::Arc::new(StreamHandler::new(muxer.clone(), ConnectionRole::Client));
let handler = Arc::new(StreamHandler::new(muxer.clone(), None));
let engine = TunnelEngine {
leg_id,
@@ -198,7 +196,11 @@ impl ClientHandler {
info!("🔑 Generated Master Session ID: {}", session_id);
let muxer = Arc::new(Muxer::new(true, session_id.clone()));
// Реестр для отправки данных обратно в виртуальный сетевой стек
let registry: Arc<DashMap<u32, StreamContext>> = Arc::new(DashMap::new());
// Мапа для трансляции: локальный socket_id -> глобальный stream_id
let local_to_global: Arc<DashMap<u32, u32>> = Arc::new(DashMap::new());
info!("🚀 Netrunner Multi-Path Tunnel Initializing (10 Legs max).");
@@ -225,6 +227,11 @@ impl ClientHandler {
if m_stats.active_legs_count() == 0 {
break;
}
// 🔥 ЗАПУСК HEALTH CHECK: Пингуем все леги
m_stats.perform_health_check().await;
// После пинга выводим красивую топологию
m_stats.print_topology_tree();
}
});
@@ -232,78 +239,105 @@ impl ClientHandler {
let muxer_inner = muxer.clone();
tokio::spawn(async move {
while let Some(raw_frame) = rx_from_engine.recv().await {
// Извлекаем нужные метаданные до передачи во фрейм
let dst_ip = raw_frame.dst_ip;
let dst_port = raw_frame.dst_port;
let protocol = raw_frame.protocol;
let is_udp = protocol == LocalProtocol::Udp;
let nrxp_frame = match RawCastAdapter::to_nrxp(raw_frame) {
Ok(frame) => frame,
Err(e) => {
error!("⚠️ [Adapter] Failed to cast frame: {}", e);
continue;
}
};
if let Ok(nrxp_frame) = RawCastAdapter::to_nrxp(raw_frame) {
// Адаптер кладет socket_id от smoltcp в заголовок stream_id
let local_socket_id = nrxp_frame.header.stream_id;
let f_type = nrxp_frame.header.frame_type;
let payload = nrxp_frame.payload;
let stream_id = nrxp_frame.header.stream_id;
let f_type = nrxp_frame.header.frame_type;
let payload = nrxp_frame.payload;
match f_type {
FrameType::Connect | FrameType::UdpConnect => {
// 🔥 ШАГ 2: Выделяем глобальный ID для нового потока
let global_stream_id = muxer_inner.next_stream_id();
match f_type {
FrameType::Connect | FrameType::UdpConnect => {
info!(
"🔗 [Muxer] Forwarding {} request to {} (stream {})",
if is_udp { "UDP" } else { "TCP" },
String::from_utf8_lossy(&payload),
stream_id
);
info!(
"🔗 [Muxer] Forwarding {} request to {} (Local ID: {} -> Global ID: {})",
if is_udp { "UDP" } else { "TCP" },
String::from_utf8_lossy(&payload),
local_socket_id,
global_stream_id
);
registry.insert(stream_id, (dst_ip, dst_port, protocol));
// Запоминаем связь
local_to_global.insert(local_socket_id, global_stream_id);
registry.insert(
global_stream_id,
(local_socket_id, dst_ip, dst_port, protocol),
);
let (v_tx, mut v_rx) =
mpsc::channel(NetworkConfig::global().tcp_stream_capacity);
muxer_inner.register_stream(stream_id, v_tx);
let (v_tx, mut v_rx) =
mpsc::channel(NetworkConfig::global().tcp_stream_capacity);
let tx_to_tun = tx_to_engine.clone();
let reg = registry.clone();
// Регистрируем локальный поток под глобальным ID
muxer_inner.register_stream(global_stream_id, v_tx);
tokio::spawn(async move {
while let Some(back_payload) = v_rx.recv().await {
if let Some(ctx) = reg.get(&stream_id) {
let (ip, port, proto) = *ctx;
let out_f_type = if proto == LocalProtocol::Udp {
FrameType::UdpData
} else {
FrameType::Data
};
let mock_nrxp = Frame::new(stream_id, out_f_type, back_payload);
let tx_to_tun = tx_to_engine.clone();
let reg = registry.clone();
if let Ok(raw) = RawCastAdapter::from_nrxp(
mock_nrxp,
ip,
port,
proto == LocalProtocol::Udp,
) {
let _ = tx_to_tun.send(raw).await;
tokio::spawn(async move {
while let Some(back_payload) = v_rx.recv().await {
if let Some(ctx) = reg.get(&global_stream_id) {
let (orig_local_id, ip, port, proto) = *ctx;
let out_f_type = if proto == LocalProtocol::Udp {
FrameType::UdpData
} else {
FrameType::Data
};
// Оборачиваем обратно в оригинальный локальный ID,
// чтобы smoltcp и tun-адаптер поняли, кому это принадлежит
let mock_nrxp =
Frame::new(orig_local_id, out_f_type, back_payload);
if let Ok(raw) = RawCastAdapter::from_nrxp(
mock_nrxp,
ip,
port,
proto == LocalProtocol::Udp,
) {
let _ = tx_to_tun.send(raw).await;
}
}
}
});
let _ = muxer_inner.send_control(global_stream_id, f_type, payload);
}
FrameType::Data | FrameType::UdpData => {
// Ищем глобальный ID по локальному
if let Some(global_ref) = local_to_global.get(&local_socket_id) {
let global_stream_id = *global_ref;
let _ =
muxer_inner.send_data_safe(global_stream_id, payload, is_udp);
} else {
warn!(
"Data received for unknown local socket_id: {}",
local_socket_id
);
}
});
let _ = muxer_inner.send_control(stream_id, f_type, payload);
}
FrameType::Data | FrameType::UdpData => {
let _ = muxer_inner.send_data_safe(stream_id, payload, is_udp);
}
FrameType::Close => {
let _ = muxer_inner.send_control(stream_id, FrameType::Close, Bytes::new());
muxer_inner.remove_stream(stream_id);
}
_ => {
warn!("Unhandled FrameType from adapter: {:?}", f_type);
}
FrameType::Close => {
if let Some((_, global_stream_id)) =
local_to_global.remove(&local_socket_id)
{
let _ = muxer_inner.send_control(
global_stream_id,
FrameType::Close,
Bytes::new(),
);
muxer_inner.remove_stream(global_stream_id);
registry.remove(&global_stream_id);
}
}
_ => {
warn!("Unhandled FrameType from adapter: {:?}", f_type);
}
}
}
}
@@ -379,6 +413,8 @@ impl TunnelHandler for ServerHandler {
info!("Acting as TLS Server with Stealth Fallback");
let handshake_timeout = std::time::Duration::from_secs(1);
// 1. ФАЗА TLS HANDSHAKE
let hello = loop {
let buf_snapshot = self.conn.read_buf.clone().freeze();
match self
@@ -387,17 +423,35 @@ impl TunnelHandler for ServerHandler {
.make_server_handshake(&mut self.conn.read_buf)
{
Ok(b) => break b,
Err(e) if e.action == ErrorAction::Wait => {
let res = tokio::time::timeout(
handshake_timeout,
self.conn.inbound.read_buf(&mut self.conn.read_buf),
)
.await;
match res {
Ok(Ok(0)) => return Err("Client closed".into()),
Ok(Ok(_)) => continue,
_ => {
ServerHandler::handle_stealth_fallback(
Err(e) => {
// 🔥 ВЫЗЫВАЕМ СТРАТЕГИЮ (Она заодно и залогирует ошибку)
match e.execute_strategy() {
ErrorAction::Wait => {
let res = tokio::time::timeout(
handshake_timeout,
self.conn.inbound.read_buf(&mut self.conn.read_buf),
)
.await;
match res {
Ok(Ok(0)) => return Err("Client closed".into()),
Ok(Ok(_)) => continue,
_ => {
// По таймауту тоже редиректим, так как сканеры часто
// присылают пол-пакета и "висят"
Self::handle_stealth_fallback(
self.conn.inbound,
self.conn.outbound,
buf_snapshot,
)
.await;
return Ok(());
}
}
}
ErrorAction::Redirect => {
// Пришел валидный TLS пакет, но от чужака -> притворяемся сайтом
Self::handle_stealth_fallback(
self.conn.inbound,
self.conn.outbound,
buf_snapshot,
@@ -405,26 +459,23 @@ impl TunnelHandler for ServerHandler {
.await;
return Ok(());
}
ErrorAction::Drop => {
// Полный мусор, отключаем немедленно (экономим сокеты ОС)
return Err("Dropped by security strategy (TLS Phase)".into());
}
}
}
Err(_) => {
ServerHandler::handle_stealth_fallback(
self.conn.inbound,
self.conn.outbound,
buf_snapshot,
)
.await;
return Ok(());
}
}
};
// Отвечаем ServerHello
self.conn
.outbound
.write_all(&hello)
.await
.map_err(|e| e.to_string())?;
// 2. ФАЗА NETRUNNER HANDSHAKE (Внутри туннеля)
let (session_id, leg_id) = loop {
let n = tokio::time::timeout(
std::time::Duration::from_secs(3),
@@ -452,32 +503,50 @@ impl TunnelHandler for ServerHandler {
sid, lid
);
break (sid, lid);
} else {
return Err("Invalid Handshake format".into());
}
} else {
return Err("Invalid first frame (Expected Handshake)".into());
return Err("Invalid Handshake format".into());
}
return Err("Expected Handshake frame".into());
}
Ok(None) => continue,
Err(e) => return Err(format!("Decryption/Auth failed during Handshake: {:?}", e)),
Err(e) => {
let buf_snapshot = self.conn.read_buf.clone().freeze();
match e.execute_strategy() {
ErrorAction::Redirect => {
// Если мы решаем прятать VPN даже на уровне неудачной криптографии
Self::handle_stealth_fallback(
self.conn.inbound,
self.conn.outbound,
buf_snapshot, // Отправляем зашифрованный мусор на целевой сервак (он сам его дропнет)
)
.await;
return Ok(());
}
_ => {
// Wait или Drop на этапе Handshake лучше просто дропать
return Err("Dropped by security strategy (Auth Phase)".into());
}
}
}
}
};
// ... Инициализация Muxer и TunnelEngine (остается без изменений)
let muxer = self.session_manager.get_or_create(&session_id);
let (control_tx, control_rx) = mpsc::channel(NetworkConfig::global().muxer_capacity);
let (data_tx, data_rx) = mpsc::channel(NetworkConfig::global().muxer_capacity);
muxer.add_leg(leg_id, control_tx, data_tx);
let handler =
std::sync::Arc::new(StreamHandler::new(muxer.clone(), ConnectionRole::Server));
let opener = Arc::new(RemoteOpener {
muxer: muxer.clone(),
});
let handler = Arc::new(StreamHandler::new(muxer.clone(), Some(opener)));
let m_weak = Arc::downgrade(&muxer);
tokio::spawn(async move {
while let Some(m_stats) = m_weak.upgrade() {
tokio::time::sleep(Duration::from_secs(15)).await;
tokio::time::sleep(std::time::Duration::from_secs(15)).await;
if m_stats.active_legs_count() == 0 {
break;
}
@@ -497,13 +566,13 @@ impl TunnelHandler for ServerHandler {
muxer: muxer.clone(),
};
let _ = engine.run().await;
let res = engine.run().await;
muxer.remove_leg(leg_id);
if muxer.active_legs_count() == 0 {
self.session_manager.remove(&session_id);
}
Ok(())
res
}
}
core/src/net/connection/engine.rs
+6 -5
View File
@@ -12,7 +12,7 @@ use tokio_util::sync::CancellationToken;
use crate::{
net::{
connection::{handler::StreamHandler, muxer::MuxMessage},
network::NetworkConfig,
NetworkConfig,
},
nrxp::{Codec, ErrorAction, FrameType},
};
@@ -176,15 +176,16 @@ impl TunnelEngine {
{
let mut c = codec.lock().await;
if data.is_empty() {
match c.encrypt_data(stream_id, frame_type.clone(), Bytes::new()) {
if frame_type == FrameType::UdpData {
match c.encrypt_data(stream_id, frame_type.clone(), data) {
Ok(pkt) => packets.push(pkt),
Err(e) => {
error!(stream_id, error = ?e, "Encryption failed for empty message");
error!(stream_id, error = ?e, "Encryption failed for UDP datagram");
return Err(format!("Encryption error: {:?}", e));
}
}
} else {
// TCP - потоковый, режем на чанки по 16KB для совместимости с TLS Record
while !data.is_empty() {
let chunk_size =
std::cmp::min(data.len(), NetworkConfig::global().tcp_chunk_size);
@@ -193,7 +194,7 @@ impl TunnelEngine {
match c.encrypt_data(stream_id, frame_type.clone(), chunk) {
Ok(pkt) => packets.push(pkt),
Err(e) => {
error!(stream_id, error = ?e, "Encryption failed for chunked message");
error!(stream_id, error = ?e, "Encryption failed for TCP chunk");
return Err(format!("Encryption error: {:?}", e));
}
}
core/src/net/connection/handler.rs
+103 -143
View File
@@ -1,174 +1,134 @@
use std::sync::Arc;
use bytes::Bytes;
use netrunner_logger::{debug, error, info, warn};
use std::sync::Arc;
use std::time::{Duration, Instant};
use tokio::net::{TcpStream, UdpSocket};
use tokio::sync::mpsc;
use crate::{
net::{
connection::{
bridge::{run_tcp_bridge, run_udp_bridge},
connection::ConnectionRole,
muxer::Muxer,
},
network::NetworkConfig,
use crate::net::{
connection::{
bridge::{run_tcp_bridge, run_udp_bridge},
muxer::Muxer,
},
nrxp::{Frame, FrameType},
NetworkConfig,
};
use crate::nrxp::{Frame, FrameType};
// --- ОТКРЫВАШКА ДЛЯ СЕРВЕРА (Remote) ---
// Этот парень реально идет в интернет через сокеты ОС
pub struct RemoteOpener {
pub muxer: Arc<Muxer>,
}
impl RemoteOpener {
pub async fn open_tcp(&self, stream_id: u32, target: String, v_rx: mpsc::Receiver<Bytes>) {
let muxer = self.muxer.clone();
tokio::spawn(async move {
info!(stream_id, "🌐 [Remote] Connecting to {}", target);
let start = Instant::now();
let connect_res =
tokio::time::timeout(Duration::from_secs(7), TcpStream::connect(&target)).await;
match connect_res {
Ok(Ok(stream)) => {
info!(stream_id, "✅ [Remote] Connected in {:?}", start.elapsed());
let (r, w) = stream.into_split();
run_tcp_bridge(stream_id, r, w, muxer.clone(), v_rx).await;
}
_ => {
error!(
stream_id,
"❌ [Remote] Target connection failed: {}", target
);
let _ = muxer.send_control(stream_id, FrameType::Close, Bytes::new());
}
}
muxer.remove_stream(stream_id);
});
}
pub async fn open_udp(&self, stream_id: u32, target: String, v_rx: mpsc::Receiver<Bytes>) {
let muxer = self.muxer.clone();
tokio::spawn(async move {
info!(stream_id, "🚀 [Remote] Binding UDP for {}", target);
let socket = UdpSocket::bind("0.0.0.0:0").await.ok();
if let Some(s) = socket {
if s.connect(&target).await.is_ok() {
run_udp_bridge(stream_id, s, muxer.clone(), v_rx).await;
}
}
muxer.remove_stream(stream_id);
});
}
}
// --- ЕДИНЫЙ ХЭНДЛЕР ---
pub(crate) struct StreamHandler {
muxer: Arc<Muxer>,
role: ConnectionRole,
opener: Option<Arc<RemoteOpener>>, // На сервере Some, на клиенте None
}
impl StreamHandler {
pub(crate) fn new(muxer: Arc<Muxer>, role: ConnectionRole) -> Self {
Self { muxer, role }
pub(crate) fn new(muxer: Arc<Muxer>, opener: Option<Arc<RemoteOpener>>) -> Self {
Self { muxer, opener }
}
pub(crate) async fn handle(&self, frame: Frame) {
let stream_id = frame.header.stream_id;
info!(
stream_id,
"📥 [Tunnel] Received frame: {:?}", frame.header.frame_type
);
match frame.header.frame_type {
FrameType::Connect => self.on_connect(stream_id, frame.payload).await,
FrameType::UdpConnect => self.on_udp_connect(stream_id, frame.payload).await,
FrameType::Data => self.on_data(stream_id, frame.payload).await,
FrameType::UdpData => self.on_udp_data(stream_id, frame.payload).await,
FrameType::Close => self.on_close(stream_id).await,
_ => debug!(stream_id, "Unhandled frame type"),
}
}
async fn on_connect(&self, stream_id: u32, payload: Bytes) {
let target_str = String::from_utf8_lossy(&payload).to_string();
FrameType::Connect => {
self.handle_conn_request(stream_id, frame.payload, false)
.await
}
FrameType::UdpConnect => {
self.handle_conn_request(stream_id, frame.payload, true)
.await
}
if self.role == ConnectionRole::Server {
info!(stream_id, target = %target_str, "🌐 [TCP] Request to establish remote connection");
let muxer = self.muxer.clone();
FrameType::Data | FrameType::UdpData => {
// Унифицировано: просто кидаем в локальный канал
self.muxer.dispatch_to_local(stream_id, frame.payload);
}
let (v_tx, v_rx) =
tokio::sync::mpsc::channel(NetworkConfig::global().tcp_stream_capacity);
muxer.register_stream(stream_id, v_tx);
tokio::spawn(async move {
let start = std::time::Instant::now();
info!(
stream_id,
"⏳ [TCP Worker] Attempting connection to {}", target_str
);
let connect_res = tokio::time::timeout(
std::time::Duration::from_secs(7),
tokio::net::TcpStream::connect(&target_str),
)
.await;
match connect_res {
Ok(Ok(stream)) => {
let elapsed = start.elapsed();
info!(stream_id, target = %target_str, "✅ [TCP] Established in {:?}. Starting bridge.", elapsed);
let (r, w) = stream.into_split();
run_tcp_bridge(stream_id, r, w, muxer.clone(), v_rx).await;
info!(stream_id, "🔚 [TCP Worker] Bridge task finished");
Self::close_stream(&muxer, stream_id).await;
}
Ok(Err(e)) => {
error!(stream_id, target = %target_str, error = %e, "❌ [TCP] Connection failed");
Self::close_stream(&muxer, stream_id).await;
}
Err(_) => {
error!(stream_id, target = %target_str, "⏰ [TCP] Connection timeout (DNS or IP unreachable)");
Self::close_stream(&muxer, stream_id).await;
}
}
});
} else {
info!(stream_id, "📲 [TCP] Dispatching payload to local stack");
self.muxer.dispatch_to_local(stream_id, payload);
FrameType::Close => {
debug!(stream_id, "🏁 [Tunnel] Peer closed stream");
self.muxer.remove_stream(stream_id);
}
_ => debug!(stream_id, "Unhandled frame: {:?}", frame.header.frame_type),
}
}
async fn on_udp_connect(&self, stream_id: u32, payload: Bytes) {
let target_str = String::from_utf8_lossy(&payload).to_string();
async fn handle_conn_request(&self, stream_id: u32, payload: Bytes, is_udp: bool) {
let target = String::from_utf8_lossy(&payload).to_string();
if self.role == ConnectionRole::Server {
info!(stream_id, target = %target_str, "🚀 [UDP] Request to establish remote bridge");
let muxer = self.muxer.clone();
if let Some(opener) = &self.opener {
// Создаем виртуальную трубу для этого стрима
let capacity = if is_udp {
NetworkConfig::global().udp_stream_capacity
} else {
NetworkConfig::global().tcp_stream_capacity
};
let (v_tx, v_rx) =
tokio::sync::mpsc::channel(NetworkConfig::global().udp_stream_capacity);
muxer.register_stream(stream_id, v_tx);
let (v_tx, v_rx) = mpsc::channel(capacity);
self.muxer.register_stream(stream_id, v_tx);
tokio::spawn(async move {
info!(
stream_id,
"⏳ [UDP Worker] Binding socket for {}", target_str
);
let socket_res = tokio::net::UdpSocket::bind("[::]:0").await;
let socket = match socket_res {
Ok(s) => s,
Err(e) => {
warn!(
stream_id,
"⚠️ [UDP] IPv6 bind failed, falling back to IPv4: {}", e
);
tokio::net::UdpSocket::bind("0.0.0.0:0")
.await
.expect("UDP bind fail")
}
};
match socket.connect(&target_str).await {
Ok(_) => {
let local = socket.local_addr().unwrap();
let remote = socket.peer_addr().unwrap();
info!(
"✅ [UDP {}] Socket ready. Local: {}, Remote: {}. Starting bridge.",
stream_id, local, remote
);
run_udp_bridge(stream_id, socket, muxer, v_rx).await;
info!(stream_id, "🔚 [UDP Worker] Bridge task finished");
}
Err(e) => {
error!(stream_id, target = %target_str, error = %e, "❌ [UDP] Target connect failed");
Self::close_stream(&muxer, stream_id).await;
}
}
});
if is_udp {
opener.open_udp(stream_id, target, v_rx).await;
} else {
opener.open_tcp(stream_id, target, v_rx).await;
}
} else {
info!(
warn!(
stream_id,
"📲 [UDP] Dispatching connection payload to local stack"
"⚠️ [Tunnel] Rejected incoming connection to {} (Client mode)", target
);
self.muxer.dispatch_to_local(stream_id, payload);
// Если мы клиент, мы не принимаем входящие соединения из туннеля
let _ = self
.muxer
.send_control(stream_id, FrameType::Close, Bytes::new());
}
}
async fn on_data(&self, stream_id: u32, payload: Bytes) {
debug!(stream_id, "📦 [TCP Data] Size: {} bytes", payload.len());
self.muxer.dispatch_to_local(stream_id, payload);
}
async fn on_udp_data(&self, stream_id: u32, payload: Bytes) {
debug!(stream_id, "📦 [UDP Data] Size: {} bytes", payload.len());
self.muxer.dispatch_to_local(stream_id, payload);
}
async fn on_close(&self, stream_id: u32) {
info!(stream_id, "🏁 [Close] Received close signal for stream");
self.muxer.remove_stream(stream_id);
}
async fn close_stream(muxer: &Muxer, stream_id: u32) {
info!(
stream_id,
"📡 [Control] Sending CLOSE signal to remote peer"
);
let _ = muxer.send_control(stream_id, FrameType::Close, Bytes::new());
muxer.remove_stream(stream_id);
}
}
core/src/net/connection/mod.rs
+1 -1
View File
@@ -4,4 +4,4 @@ mod engine;
mod handler;
mod muxer;
pub use connection::{ClientHandler, Connection, ConnectionRole, ServerHandler, TunnelHandler};
pub use connection::{ClientHandler, Connection, ServerHandler, TunnelHandler};
core/src/net/connection/muxer.rs
+12 -15
View File
@@ -6,7 +6,7 @@ use std::sync::atomic::{AtomicU32, AtomicU64, Ordering};
use std::sync::Arc;
use tokio::sync::mpsc::Sender;
use crate::net::network::NetworkConfig;
use crate::net::NetworkConfig;
use crate::nrxp::FrameType;
@@ -56,7 +56,6 @@ pub struct MuxMessage {
#[derive(Clone)]
pub struct Muxer {
legs: Arc<DashMap<u32, MuxLeg>>,
streams: Arc<DashMap<u32, (Sender<Bytes>, Arc<StreamStats>)>>,
id_gen: Arc<IdGenerator>,
session_id: Arc<String>,
@@ -195,27 +194,21 @@ impl Muxer {
pub fn send_data_safe(
&self,
stream_id: u32,
mut data: Bytes,
data: Bytes, // Больше не mut
is_udp: bool,
) -> Result<(), String> {
let max_chunk = NetworkConfig::global().tcp_chunk_size;
let frame_type = if is_udp {
FrameType::UdpData
} else {
FrameType::Data
};
while !data.is_empty() {
let chunk_size = std::cmp::min(data.len(), max_chunk);
let chunk = data.split_to(chunk_size);
self.send_to_network(MuxMessage {
stream_id,
frame_type: frame_type.clone(),
data: chunk,
})?;
}
Ok(())
// Отправляем как есть, целиком!
self.send_to_network(MuxMessage {
stream_id,
frame_type,
data,
})
}
pub fn send_control(
@@ -274,6 +267,10 @@ pub fn dispatch_to_local(&self, stream_id: u32, data: Bytes) {
}
}
pub fn next_stream_id(&self) -> u32 {
self.id_gen.next()
}
pub async fn perform_health_check(&self) {
let legs: Vec<(u32, Sender<MuxMessage>)> = self.legs.iter()