i/Aether
1
0
Fork 0
mirror of https://github.com/fawney19/Aether.git synced 2026-09-02 17:30:23 +08:00
Code Issues Packages Projects Releases Wiki Activity

style: cargo fmt

Browse Source
This commit is contained in:
fawney19
2026-04-15 23:54:14 +08:00
parent 5aa8883865
commit 205f78b39c
2 changed files with 20 additions and 27 deletions
Download Patch File Download Diff File Expand all files Collapse all files

35
apps/aether-gateway/src/execution_runtime/stream_pump.rs
View File

@@ -240,8 +240,7 @@ mod tests {
use super::build_direct_execution_frame_stream; use super::build_direct_execution_frame_stream;
use crate::execution_runtime::transport::{ use crate::execution_runtime::transport::{
execute_stream_plan_via_local_tunnel, DirectSyncExecutionRuntime, execute_stream_plan_via_local_tunnel, DirectSyncExecutionRuntime, DirectUpstreamResponse,
DirectUpstreamResponse,
}; };
use crate::tunnel::{tunnel_protocol, TunnelProxyConn}; use crate::tunnel::{tunnel_protocol, TunnelProxyConn};
use crate::AppState; use crate::AppState;
@@ -630,17 +629,18 @@ mod tests {
// --- First request --- // --- First request ---
let state1 = state.clone(); let state1 = state.clone();
let plan1 = plan.clone(); let plan1 = plan.clone();
let exec1 = tokio::spawn(async move { let exec1 =
execute_stream_plan_via_local_tunnel(&state1, &plan1).await tokio::spawn(
}); async move { execute_stream_plan_via_local_tunnel(&state1, &plan1).await },
);
// Read request frames from proxy side // Read request frames from proxy side
let req1_headers = match proxy_rx.recv().await.expect("req1 headers") { let req1_headers = match proxy_rx.recv().await.expect("req1 headers") {
Message::Binary(data) => data, Message::Binary(data) => data,
other => panic!("unexpected: {other:?}"), other => panic!("unexpected: {other:?}"),
}; };
let req1_header = tunnel_protocol::FrameHeader::parse(&req1_headers) let req1_header =
.expect("req1 header parse"); tunnel_protocol::FrameHeader::parse(&req1_headers).expect("req1 header parse");
let _req1_body = proxy_rx.recv().await.expect("req1 body"); let _req1_body = proxy_rx.recv().await.expect("req1 body");
// Simulate proxy response // Simulate proxy response
@@ -689,10 +689,7 @@ mod tests {
0, 0,
&[], &[],
); );
tunnel_app tunnel_app.hub.handle_proxy_frame(900, &mut end_frame).await;
.hub
.handle_proxy_frame(900, &mut end_frame)
.await;
// Drain the body // Drain the body
while let Ok(Some(_)) = resp1.next_chunk().await {} while let Ok(Some(_)) = resp1.next_chunk().await {}
@@ -705,9 +702,10 @@ mod tests {
candidate_id: Some("cand-reuse-2".into()), candidate_id: Some("cand-reuse-2".into()),
..plan.clone() ..plan.clone()
}; };
let exec2 = tokio::spawn(async move { let exec2 =
execute_stream_plan_via_local_tunnel(&state2, &plan2).await tokio::spawn(
}); async move { execute_stream_plan_via_local_tunnel(&state2, &plan2).await },
);
// Read second request's frames // Read second request's frames
let req2_headers = tokio::time::timeout(Duration::from_secs(2), proxy_rx.recv()) let req2_headers = tokio::time::timeout(Duration::from_secs(2), proxy_rx.recv())
@@ -718,8 +716,8 @@ mod tests {
Message::Binary(data) => data, Message::Binary(data) => data,
other => panic!("unexpected: {other:?}"), other => panic!("unexpected: {other:?}"),
}; };
let req2_header = tunnel_protocol::FrameHeader::parse(&req2_data) let req2_header =
.expect("req2 header parse"); tunnel_protocol::FrameHeader::parse(&req2_data).expect("req2 header parse");
assert_eq!(req2_header.msg_type, tunnel_protocol::REQUEST_HEADERS); assert_eq!(req2_header.msg_type, tunnel_protocol::REQUEST_HEADERS);
// Simulate proxy response for second request // Simulate proxy response for second request
@@ -748,9 +746,6 @@ mod tests {
0, 0,
&[], &[],
); );
tunnel_app tunnel_app.hub.handle_proxy_frame(900, &mut end2).await;
.hub
.handle_proxy_frame(900, &mut end2)
.await;
} }
} }

12
apps/aether-gateway/src/tunnel/embedded/hub.rs
View File

@@ -1149,7 +1149,9 @@ mod tests {
// Verify stream_count went back to 0 // Verify stream_count went back to 0
assert_eq!( assert_eq!(
proxy.stream_count.load(std::sync::atomic::Ordering::Relaxed), proxy
.stream_count
.load(std::sync::atomic::Ordering::Relaxed),
0, 0,
"stream_count should be 0 after STREAM_END" "stream_count should be 0 after STREAM_END"
); );
@@ -1172,12 +1174,8 @@ mod tests {
); );
// Simulate proxy response for second stream // Simulate proxy response for second stream
let mut resp2_headers = protocol::encode_frame( let mut resp2_headers =
header.stream_id, protocol::encode_frame(header.stream_id, protocol::RESPONSE_HEADERS, 0, &resp_meta);
protocol::RESPONSE_HEADERS,
0,
&resp_meta,
);
hub.handle_proxy_frame(400, &mut resp2_headers).await; hub.handle_proxy_frame(400, &mut resp2_headers).await;
let response = stream2 let response = stream2