130 lines
4.3 KiB
Rust
130 lines
4.3 KiB
Rust
use std::time::Duration;
|
|
|
|
use irtt_client::{
|
|
AsyncClient, ClientConfig, ClientEvent, OpenOutcome, OpenPolicy, SessionRequest,
|
|
};
|
|
use irtt_proto::{Clock, ReceivedStats, StampAt};
|
|
use irtt_server::{Server, ServerConfig};
|
|
use tokio::{sync::oneshot, time::timeout};
|
|
|
|
#[tokio::test(flavor = "current_thread")]
|
|
async fn async_client_opens_echoes_receives_and_closes() {
|
|
timeout(Duration::from_secs(2), exercise_client_server())
|
|
.await
|
|
.expect("client/server integration exceeded its bounded runtime");
|
|
}
|
|
|
|
async fn exercise_client_server() {
|
|
let mut server = Server::bind("127.0.0.1:0".parse().unwrap(), ServerConfig::default())
|
|
.await
|
|
.unwrap();
|
|
let server_addr = server.local_addr().unwrap();
|
|
let (shutdown_tx, shutdown_rx) = oneshot::channel();
|
|
let server_task = tokio::spawn(async move {
|
|
server
|
|
.run(async {
|
|
let _ = shutdown_rx.await;
|
|
})
|
|
.await
|
|
});
|
|
|
|
let config = ClientConfig {
|
|
open: OpenPolicy {
|
|
timeouts: vec![Duration::from_millis(200)],
|
|
..Default::default()
|
|
},
|
|
request: SessionRequest {
|
|
duration: Some(Duration::from_secs(1)),
|
|
interval: Duration::from_millis(100),
|
|
length: 64,
|
|
received_stats: ReceivedStats::Both,
|
|
stamp_at: StampAt::Both,
|
|
clock: Clock::Both,
|
|
..Default::default()
|
|
},
|
|
|
|
probe_timeout: Duration::from_millis(200),
|
|
..ClientConfig::default()
|
|
};
|
|
let mut client = AsyncClient::connect(server_addr.to_string(), config)
|
|
.await
|
|
.unwrap();
|
|
|
|
let opened = client.open().await.unwrap();
|
|
match &opened {
|
|
OpenOutcome::Started(irtt_client::SessionStarted {
|
|
remote,
|
|
token,
|
|
negotiation: negotiated,
|
|
..
|
|
}) => {
|
|
assert_eq!(*remote, server_addr);
|
|
assert_ne!(*token, 0);
|
|
assert_eq!(negotiated.accepted.received_stats, ReceivedStats::Both);
|
|
assert_eq!(negotiated.accepted.stamp_at, StampAt::Both);
|
|
assert_eq!(negotiated.accepted.clock, Clock::Both);
|
|
// A default client requests no fill, and this open succeeded under
|
|
// the default strict negotiation policy — which it could not have
|
|
// if the server had answered an absent request with its own default
|
|
// descriptor. The server still fills the payload with it.
|
|
assert_eq!(negotiated.accepted.server_fill, None);
|
|
}
|
|
OpenOutcome::NoTestCompleted(_) => panic!("normal client unexpectedly ran no-test"),
|
|
}
|
|
|
|
let receipt = client.send_probe().await.unwrap();
|
|
assert_eq!(receipt.seq, 0);
|
|
assert_eq!(receipt.remote, server_addr);
|
|
assert!(client.has_pending_probes());
|
|
let events = client.recv().await.unwrap();
|
|
let [ClientEvent::EchoReply {
|
|
seq: 0,
|
|
remote,
|
|
sent_at,
|
|
received_at,
|
|
rtt,
|
|
server_timing: Some(timing),
|
|
one_way: Some(one_way),
|
|
received_stats: Some(_),
|
|
packet_meta,
|
|
..
|
|
}] = events.as_slice()
|
|
else {
|
|
panic!("expected a measured reply, got {events:?}");
|
|
};
|
|
assert_eq!(*remote, server_addr);
|
|
assert_eq!(*sent_at, receipt.sent_at);
|
|
assert_eq!(rtt.raw, received_at.mono.duration_since(sent_at.mono));
|
|
// Verify the measurement endpoint through public reply metadata and real
|
|
// OS timestamps. This also covers the userspace fallback on other targets.
|
|
let receive_wall = packet_meta
|
|
.kernel_rx_timestamp
|
|
.filter(|kernel| {
|
|
received_at
|
|
.wall
|
|
.duration_since(*kernel)
|
|
.is_ok_and(|lag| lag <= Duration::from_secs(1))
|
|
})
|
|
.unwrap_or(received_at.wall);
|
|
let receive_ns = i128::try_from(
|
|
receive_wall
|
|
.duration_since(std::time::UNIX_EPOCH)
|
|
.unwrap()
|
|
.as_nanos(),
|
|
)
|
|
.unwrap();
|
|
assert_eq!(
|
|
one_way.server_to_client,
|
|
Some(irtt_client::SignedDuration::from_nanos(
|
|
receive_ns - i128::from(timing.send_wall_ns.unwrap())
|
|
))
|
|
);
|
|
assert!(matches!(
|
|
client.close().await.unwrap().as_slice(),
|
|
[ClientEvent::SessionClosed { remote, .. }] if *remote == server_addr
|
|
));
|
|
|
|
shutdown_tx.send(()).unwrap();
|
|
server_task.await.unwrap().unwrap();
|
|
}
|