tokio_client/
tokio_client.rs

1//! Low-level Tokio client example (`tokio` feature).
2//!
3//! Drives [`AsyncClient`] directly on a caller-owned runtime. This is the
4//! adapter to reach for when you already have a Tokio runtime and want to
5//! await sends/receives yourself, rather than use the higher-level `managed`
6//! driver (see the `managed_client` example).
7//!
8//! Run a server first, then run this example:
9//!
10//! ```text
11//! cargo run -p irtt-rs --bin irtt-server --features server
12//! cargo run -p irtt-client --example tokio_client --features tokio
13//! ```
14//!
15//! Without a reachable server this exits quickly with an open-timeout error
16//! instead of hanging, because the example shortens `open_timeouts` for a
17//! fast demonstration.
18
19use std::time::{Duration, Instant};
20
21use irtt_client::{AsyncClient, ClientConfig, ClientEvent};
22
23/// Upper bound on how long the receive wait can go without also calling
24/// `poll_timeouts()`. `AsyncClient` does not expose the earliest pending
25/// probe's own timeout deadline, so this example polls on a short fixed
26/// cadence instead of waiting for the full `probe_timeout()` from whenever
27/// this loop last ran — that would let poll_timeouts() classify a lost
28/// probe (and its dependent statistics/session state) far later than it
29/// actually became lost.
30const TIMEOUT_POLL_INTERVAL: Duration = Duration::from_millis(50);
31
32fn main() {
33    let runtime = tokio::runtime::Builder::new_current_thread()
34        .enable_io()
35        .enable_time()
36        .build()
37        .expect("failed to build Tokio runtime");
38
39    runtime.block_on(run());
40}
41
42async fn run() {
43    let config = ClientConfig {
44        server_addr: "127.0.0.1:2112".to_owned(),
45        duration: Some(Duration::from_secs(2)),
46        interval: Duration::from_millis(200),
47        // A single short attempt keeps this example fast when no server is
48        // reachable; production callers should generally keep the default.
49        open_timeouts: vec![Duration::from_millis(300)],
50        ..Default::default()
51    };
52
53    let mut client = match AsyncClient::connect(config).await {
54        Ok(client) => client,
55        Err(err) => {
56            eprintln!("failed to prepare client socket: {err}");
57            return;
58        }
59    };
60
61    match client.open().await {
62        Ok(outcome) => println!("session opened: {outcome:?}"),
63        Err(err) => {
64            eprintln!("open failed (is a server running at 127.0.0.1:2112?): {err}");
65            return;
66        }
67    }
68
69    while !client.is_run_complete() {
70        // Bound the receive wait by the next send/timeout deadline, rather
71        // than awaiting recv() unconditionally: a lost or rate-limited reply
72        // must not stall pacing or timeout classification indefinitely.
73        let wake_at = client
74            .next_send_deadline()
75            .into_iter()
76            .chain(std::iter::once(Instant::now() + TIMEOUT_POLL_INTERVAL))
77            .min()
78            .expect("the fixed poll-interval deadline is always present");
79
80        let recv_result = tokio::select! {
81            events = client.recv() => Some(events),
82            () = tokio::time::sleep_until(tokio::time::Instant::from_std(wake_at)) => None,
83        };
84        if let Some(result) = recv_result {
85            match result {
86                Ok(events) => events.iter().for_each(print_event),
87                Err(err) => {
88                    eprintln!("receive failed: {err}");
89                    break;
90                }
91            }
92        }
93
94        if client
95            .next_send_deadline()
96            .is_some_and(|send_at| Instant::now() >= send_at)
97        {
98            // A failed send retains its prepared probe without advancing the
99            // schedule, so a persistent error (e.g. the interface going
100            // down) would otherwise make this loop retry forever instead of
101            // reporting it.
102            match client.send_probe().await {
103                Ok(events) => events.iter().for_each(print_event),
104                Err(err) => {
105                    eprintln!("send failed: {err}");
106                    break;
107                }
108            }
109        }
110        match client.poll_timeouts() {
111            Ok(events) => events.iter().for_each(print_event),
112            Err(err) => {
113                eprintln!("timeout polling failed: {err}");
114                break;
115            }
116        }
117    }
118
119    match client.close().await {
120        Ok(events) => events.iter().for_each(print_event),
121        Err(err) => eprintln!("close failed: {err}"),
122    }
123}
124
125fn print_event(event: &ClientEvent) {
126    match event {
127        ClientEvent::EchoReply { seq, rtt, .. } => {
128            println!("seq {seq}: rtt {:?}", rtt.effective);
129        }
130        ClientEvent::EchoLoss { seq, .. } => println!("seq {seq}: lost"),
131        other => println!("{other:?}"),
132    }
133}
tokio_client/tokio_client.rs

tokio_client/
tokio_client.rs
