mirror of
https://github.com/fawney19/Aether.git
synced 2026-09-02 01:10:23 +08:00
diag(tunnel): writer/reader 帧计数诊断日志
- writer task 记录发送帧数、Binary 帧大小、send 失败原因 - reader 记录接收帧总数,disconnect 时输出 - 用于定位 gateway→proxy 数据流是否正常
This commit is contained in:
@@ -41,14 +41,36 @@ pub async fn handle_proxy_connection(
|
|||||||
|
|
||||||
hub.register_proxy(conn.clone());
|
hub.register_proxy(conn.clone());
|
||||||
|
|
||||||
|
let writer_conn_id = conn_id;
|
||||||
let writer = tokio::spawn(async move {
|
let writer = tokio::spawn(async move {
|
||||||
|
let mut frames_sent: u64 = 0;
|
||||||
loop {
|
loop {
|
||||||
tokio::select! {
|
tokio::select! {
|
||||||
msg = rx.recv() => match msg {
|
msg = rx.recv() => match msg {
|
||||||
Some(msg) => {
|
Some(msg) => {
|
||||||
if ws_tx.send(msg).await.is_err() {
|
let is_binary = matches!(&msg, Message::Binary(_));
|
||||||
|
let msg_len = match &msg {
|
||||||
|
Message::Binary(b) => b.len(),
|
||||||
|
_ => 0,
|
||||||
|
};
|
||||||
|
if let Err(e) = ws_tx.send(msg).await {
|
||||||
|
warn!(
|
||||||
|
conn_id = writer_conn_id,
|
||||||
|
frames_sent = frames_sent,
|
||||||
|
error = %e,
|
||||||
|
"writer ws_tx.send failed"
|
||||||
|
);
|
||||||
break;
|
break;
|
||||||
}
|
}
|
||||||
|
frames_sent += 1;
|
||||||
|
if is_binary && msg_len > protocol::HEADER_SIZE {
|
||||||
|
debug!(
|
||||||
|
conn_id = writer_conn_id,
|
||||||
|
size = msg_len,
|
||||||
|
frames_sent = frames_sent,
|
||||||
|
"writer sent binary frame"
|
||||||
|
);
|
||||||
|
}
|
||||||
}
|
}
|
||||||
None => break,
|
None => break,
|
||||||
},
|
},
|
||||||
@@ -59,6 +81,11 @@ pub async fn handle_proxy_connection(
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
info!(
|
||||||
|
conn_id = writer_conn_id,
|
||||||
|
frames_sent = frames_sent,
|
||||||
|
"writer task exiting"
|
||||||
|
);
|
||||||
let _ = ws_tx.close().await;
|
let _ = ws_tx.close().await;
|
||||||
});
|
});
|
||||||
|
|
||||||
@@ -99,11 +126,13 @@ async fn run_proxy_reader(
|
|||||||
conn: Arc<ProxyConn>,
|
conn: Arc<ProxyConn>,
|
||||||
) {
|
) {
|
||||||
let mut oversized_count = 0u32;
|
let mut oversized_count = 0u32;
|
||||||
|
let mut frames_received: u64 = 0;
|
||||||
loop {
|
loop {
|
||||||
let msg = ws_rx.next().await;
|
let msg = ws_rx.next().await;
|
||||||
|
|
||||||
match msg {
|
match msg {
|
||||||
Some(Ok(Message::Binary(data))) => {
|
Some(Ok(Message::Binary(data))) => {
|
||||||
|
frames_received += 1;
|
||||||
let mut data = data.to_vec();
|
let mut data = data.to_vec();
|
||||||
if data.len() > MAX_FRAME_SIZE {
|
if data.len() > MAX_FRAME_SIZE {
|
||||||
oversized_count += 1;
|
oversized_count += 1;
|
||||||
@@ -129,11 +158,21 @@ async fn run_proxy_reader(
|
|||||||
hub.handle_proxy_frame(conn.id, &mut data).await;
|
hub.handle_proxy_frame(conn.id, &mut data).await;
|
||||||
}
|
}
|
||||||
Some(Ok(Message::Close(_))) | None => {
|
Some(Ok(Message::Close(_))) | None => {
|
||||||
info!(conn_id = conn.id, node_id = %conn.node_id, "proxy WebSocket closed");
|
info!(
|
||||||
|
conn_id = conn.id,
|
||||||
|
node_id = %conn.node_id,
|
||||||
|
frames_received = frames_received,
|
||||||
|
"proxy WebSocket closed"
|
||||||
|
);
|
||||||
break;
|
break;
|
||||||
}
|
}
|
||||||
Some(Err(e)) => {
|
Some(Err(e)) => {
|
||||||
warn!(conn_id = conn.id, error = %e, "proxy WebSocket error");
|
warn!(
|
||||||
|
conn_id = conn.id,
|
||||||
|
frames_received = frames_received,
|
||||||
|
error = %e,
|
||||||
|
"proxy WebSocket error"
|
||||||
|
);
|
||||||
break;
|
break;
|
||||||
}
|
}
|
||||||
Some(Ok(Message::Ping(payload))) => {
|
Some(Ok(Message::Ping(payload))) => {
|
||||||
|
|||||||
Reference in New Issue
Block a user