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(); }