Struct AsyncClient 

pub struct AsyncClient { /* private fields */ }

Low-level Tokio IRTT client for one connected UDP target.

AsyncClient does not construct, own, or store a Tokio runtime. Its async methods are polled by the caller and require a current Tokio runtime with I/O and time enabled.

Example

Use this low-level adapter when the application already owns Tokio and wants to drive the session lifecycle itself. This example type-checks without contacting a server.

use irtt_client::{AsyncClient, ClientConfig};

let mut client = AsyncClient::connect("127.0.0.1:2112", ClientConfig::default()).await?;
let outcome = client.open().await?;
println!("opened: {outcome:?}");
let sent = client.send_probe().await?;
println!("sent: {sent:?}");
let received = client.recv().await?;
println!("received: {received:?}");
let timed_out = client.poll_timeouts()?;
println!("timeouts: {timed_out:?}");
let closed = client.close().await?;
println!("closed: {closed:?}");

Implementations

impl AsyncClient

pub async fn connect( endpoint: impl AsRef<str>, config: ClientConfig, ) -> Result<Self, ClientError>

Resolve the endpoint and construct one connected Tokio UDP socket.

Polling this future without a current Tokio runtime returns ClientError::NoTokioRuntime. A runtime without enabled I/O or time drivers is outside this type’s runtime contract.

The endpoint accepts a name or address; an omitted port defaults to 2112. IPv6 literals may be bracketed or unbracketed with the default port. The reusable config contains no endpoint identity.

Examples found in repository

examples/tokio_client.rs (line 60)

async fn run() {
    let config = ClientConfig {
        open: OpenPolicy {
            // A single short attempt keeps this example fast when no server is
            // reachable; production callers should generally keep the default.
            timeouts: vec![Duration::from_millis(300)],
            ..Default::default()
        },
        request: SessionRequest {
            duration: Some(Duration::from_secs(2)),
            interval: Duration::from_millis(200),
            ..Default::default()
        },
        ..Default::default()
    };

    let mut client = match AsyncClient::connect("127.0.0.1:2112", config).await {
        Ok(client) => client,
        Err(err) => {
            eprintln!("failed to prepare client socket: {err}");
            return;
        }
    };

    let negotiated = match client.open().await {
        Ok(OpenOutcome::Started(started)) => {
            println!("session opened: {started:?}");
            started.negotiation
        }
        Ok(OpenOutcome::NoTestCompleted(completed)) => {
            println!("{completed:?}");
            return;
        }
        Err(err) => {
            eprintln!("open failed (is a server running at 127.0.0.1:2112?): {err}");
            return;
        }
    };

    // This loop owns cadence and duration. The transport sends whenever called.
    let interval = negotiated.accepted.interval;
    let start = Instant::now();
    let end = start
        + negotiated
            .accepted
            .duration
            .expect("example requests a finite run");
    let deadline = end + client.probe_timeout();
    let mut next_send = start;
    while Instant::now() < deadline && !client.is_peer_closed() {
        if Instant::now() >= end && !client.has_pending_probes() {
            break;
        }
        // Bound the receive wait by the next send/timeout deadline, rather
        // than awaiting recv() unconditionally: a lost or rate-limited reply
        // must not stall pacing or timeout classification indefinitely.
        let wake_at = (next_send < end)
            .then_some(next_send)
            .into_iter()
            .chain(client.next_probe_timeout_deadline())
            .chain(std::iter::once(Instant::now() + TIMEOUT_POLL_INTERVAL))
            .min()
            .unwrap();

        let recv_result = tokio::select! {
            events = client.recv() => Some(events),
            () = tokio::time::sleep_until(tokio::time::Instant::from_std(wake_at)) => None,
        };
        if let Some(result) = recv_result {
            match result {
                Ok(events) => events.iter().for_each(print_event),
                Err(err) => {
                    eprintln!("receive failed: {err}");
                    break;
                }
            }
        }

        if Instant::now() < end && Instant::now() >= next_send {
            match client.send_probe().await {
                Ok(receipt) => {
                    print_event(&receipt.into());
                    while next_send <= Instant::now() {
                        next_send += interval;
                    }
                }
                Err(err) => {
                    if let irtt_client::SendProbeError::AfterCommit { receipt, .. } = &err {
                        print_event(&(*receipt).into());
                    }
                    eprintln!("send failed: {err}");
                    break;
                }
            }
        }
        match client.poll_timeouts() {
            Ok(events) => events.iter().for_each(print_event),
            Err(err) => {
                eprintln!("timeout polling failed: {err}");
                break;
            }
        }
    }

    if client.is_peer_closed() {
        return;
    }
    match client.close().await {
        Ok(events) => events.iter().for_each(print_event),
        Err(err) => eprintln!("close failed: {err}"),
    }
}

pub async fn open(&mut self) -> Result<OpenOutcome, ClientError>

Perform the IRTT open exchange.

Each configured attempt uses one absolute deadline for one successful request submission and all replies inspected during that attempt. Malformed, unrelated, wrong-direction, and unauthenticated datagrams are ignored without resending the request.

Dropping this future before trusted acceptance leaves the local machine connected with no negotiated adapter state. If it is dropped during best-effort post-token cleanup, cleanup may remain incomplete because this low-level client never detaches cleanup work.

Examples found in repository

examples/tokio_client.rs (line 68)

async fn run() {
    let config = ClientConfig {
        open: OpenPolicy {
            // A single short attempt keeps this example fast when no server is
            // reachable; production callers should generally keep the default.
            timeouts: vec![Duration::from_millis(300)],
            ..Default::default()
        },
        request: SessionRequest {
            duration: Some(Duration::from_secs(2)),
            interval: Duration::from_millis(200),
            ..Default::default()
        },
        ..Default::default()
    };

    let mut client = match AsyncClient::connect("127.0.0.1:2112", config).await {
        Ok(client) => client,
        Err(err) => {
            eprintln!("failed to prepare client socket: {err}");
            return;
        }
    };

    let negotiated = match client.open().await {
        Ok(OpenOutcome::Started(started)) => {
            println!("session opened: {started:?}");
            started.negotiation
        }
        Ok(OpenOutcome::NoTestCompleted(completed)) => {
            println!("{completed:?}");
            return;
        }
        Err(err) => {
            eprintln!("open failed (is a server running at 127.0.0.1:2112?): {err}");
            return;
        }
    };

    // This loop owns cadence and duration. The transport sends whenever called.
    let interval = negotiated.accepted.interval;
    let start = Instant::now();
    let end = start
        + negotiated
            .accepted
            .duration
            .expect("example requests a finite run");
    let deadline = end + client.probe_timeout();
    let mut next_send = start;
    while Instant::now() < deadline && !client.is_peer_closed() {
        if Instant::now() >= end && !client.has_pending_probes() {
            break;
        }
        // Bound the receive wait by the next send/timeout deadline, rather
        // than awaiting recv() unconditionally: a lost or rate-limited reply
        // must not stall pacing or timeout classification indefinitely.
        let wake_at = (next_send < end)
            .then_some(next_send)
            .into_iter()
            .chain(client.next_probe_timeout_deadline())
            .chain(std::iter::once(Instant::now() + TIMEOUT_POLL_INTERVAL))
            .min()
            .unwrap();

        let recv_result = tokio::select! {
            events = client.recv() => Some(events),
            () = tokio::time::sleep_until(tokio::time::Instant::from_std(wake_at)) => None,
        };
        if let Some(result) = recv_result {
            match result {
                Ok(events) => events.iter().for_each(print_event),
                Err(err) => {
                    eprintln!("receive failed: {err}");
                    break;
                }
            }
        }

        if Instant::now() < end && Instant::now() >= next_send {
            match client.send_probe().await {
                Ok(receipt) => {
                    print_event(&receipt.into());
                    while next_send <= Instant::now() {
                        next_send += interval;
                    }
                }
                Err(err) => {
                    if let irtt_client::SendProbeError::AfterCommit { receipt, .. } = &err {
                        print_event(&(*receipt).into());
                    }
                    eprintln!("send failed: {err}");
                    break;
                }
            }
        }
        match client.poll_timeouts() {
            Ok(events) => events.iter().for_each(print_event),
            Err(err) => {
                eprintln!("timeout polling failed: {err}");
                break;
            }
        }
    }

    if client.is_peer_closed() {
        return;
    }
    match client.close().await {
        Ok(events) => events.iter().for_each(print_event),
        Err(err) => eprintln!("close failed: {err}"),
    }
}

pub async fn send_probe(&mut self) -> Result<SendReceipt, SendProbeError>

Send one echo probe now, committing protocol state only after acceptance. The caller owns pacing and run duration.

A prepared packet is retained across readiness false positives, cancellation, and socket errors so a later call retries the same logical probe without advancing state before kernel acceptance. Cancellation or WouldBlock before submission is uncommitted. After acceptance there is no suspension before infallible commitment, and the prepared probe is cleared even if post-send processing returns SendProbeError::AfterCommit. That failure retains the authoritative receipt; earlier failures return SendProbeError::NotCommitted.

Examples found in repository

examples/tokio_client.rs (line 123)

async fn run() {
    let config = ClientConfig {
        open: OpenPolicy {
            // A single short attempt keeps this example fast when no server is
            // reachable; production callers should generally keep the default.
            timeouts: vec![Duration::from_millis(300)],
            ..Default::default()
        },
        request: SessionRequest {
            duration: Some(Duration::from_secs(2)),
            interval: Duration::from_millis(200),
            ..Default::default()
        },
        ..Default::default()
    };

    let mut client = match AsyncClient::connect("127.0.0.1:2112", config).await {
        Ok(client) => client,
        Err(err) => {
            eprintln!("failed to prepare client socket: {err}");
            return;
        }
    };

    let negotiated = match client.open().await {
        Ok(OpenOutcome::Started(started)) => {
            println!("session opened: {started:?}");
            started.negotiation
        }
        Ok(OpenOutcome::NoTestCompleted(completed)) => {
            println!("{completed:?}");
            return;
        }
        Err(err) => {
            eprintln!("open failed (is a server running at 127.0.0.1:2112?): {err}");
            return;
        }
    };

    // This loop owns cadence and duration. The transport sends whenever called.
    let interval = negotiated.accepted.interval;
    let start = Instant::now();
    let end = start
        + negotiated
            .accepted
            .duration
            .expect("example requests a finite run");
    let deadline = end + client.probe_timeout();
    let mut next_send = start;
    while Instant::now() < deadline && !client.is_peer_closed() {
        if Instant::now() >= end && !client.has_pending_probes() {
            break;
        }
        // Bound the receive wait by the next send/timeout deadline, rather
        // than awaiting recv() unconditionally: a lost or rate-limited reply
        // must not stall pacing or timeout classification indefinitely.
        let wake_at = (next_send < end)
            .then_some(next_send)
            .into_iter()
            .chain(client.next_probe_timeout_deadline())
            .chain(std::iter::once(Instant::now() + TIMEOUT_POLL_INTERVAL))
            .min()
            .unwrap();

        let recv_result = tokio::select! {
            events = client.recv() => Some(events),
            () = tokio::time::sleep_until(tokio::time::Instant::from_std(wake_at)) => None,
        };
        if let Some(result) = recv_result {
            match result {
                Ok(events) => events.iter().for_each(print_event),
                Err(err) => {
                    eprintln!("receive failed: {err}");
                    break;
                }
            }
        }

        if Instant::now() < end && Instant::now() >= next_send {
            match client.send_probe().await {
                Ok(receipt) => {
                    print_event(&receipt.into());
                    while next_send <= Instant::now() {
                        next_send += interval;
                    }
                }
                Err(err) => {
                    if let irtt_client::SendProbeError::AfterCommit { receipt, .. } = &err {
                        print_event(&(*receipt).into());
                    }
                    eprintln!("send failed: {err}");
                    break;
                }
            }
        }
        match client.poll_timeouts() {
            Ok(events) => events.iter().for_each(print_event),
            Err(err) => {
                eprintln!("timeout polling failed: {err}");
                break;
            }
        }
    }

    if client.is_peer_closed() {
        return;
    }
    match client.close().await {
        Ok(events) => events.iter().for_each(print_event),
        Err(err) => eprintln!("close failed: {err}"),
    }
}

pub async fn recv(&mut self) -> Result<Vec<ClientEvent>, ClientError>

Await and classify one complete UDP datagram.

State is validated before readiness is awaited. Readiness false positives retry without changing protocol state. Authenticated peer-close events remain authoritative even if best-effort DSCP cleanup fails.

Examples found in repository

examples/tokio_client.rs (line 109)

async fn run() {
    let config = ClientConfig {
        open: OpenPolicy {
            // A single short attempt keeps this example fast when no server is
            // reachable; production callers should generally keep the default.
            timeouts: vec![Duration::from_millis(300)],
            ..Default::default()
        },
        request: SessionRequest {
            duration: Some(Duration::from_secs(2)),
            interval: Duration::from_millis(200),
            ..Default::default()
        },
        ..Default::default()
    };

    let mut client = match AsyncClient::connect("127.0.0.1:2112", config).await {
        Ok(client) => client,
        Err(err) => {
            eprintln!("failed to prepare client socket: {err}");
            return;
        }
    };

    let negotiated = match client.open().await {
        Ok(OpenOutcome::Started(started)) => {
            println!("session opened: {started:?}");
            started.negotiation
        }
        Ok(OpenOutcome::NoTestCompleted(completed)) => {
            println!("{completed:?}");
            return;
        }
        Err(err) => {
            eprintln!("open failed (is a server running at 127.0.0.1:2112?): {err}");
            return;
        }
    };

    // This loop owns cadence and duration. The transport sends whenever called.
    let interval = negotiated.accepted.interval;
    let start = Instant::now();
    let end = start
        + negotiated
            .accepted
            .duration
            .expect("example requests a finite run");
    let deadline = end + client.probe_timeout();
    let mut next_send = start;
    while Instant::now() < deadline && !client.is_peer_closed() {
        if Instant::now() >= end && !client.has_pending_probes() {
            break;
        }
        // Bound the receive wait by the next send/timeout deadline, rather
        // than awaiting recv() unconditionally: a lost or rate-limited reply
        // must not stall pacing or timeout classification indefinitely.
        let wake_at = (next_send < end)
            .then_some(next_send)
            .into_iter()
            .chain(client.next_probe_timeout_deadline())
            .chain(std::iter::once(Instant::now() + TIMEOUT_POLL_INTERVAL))
            .min()
            .unwrap();

        let recv_result = tokio::select! {
            events = client.recv() => Some(events),
            () = tokio::time::sleep_until(tokio::time::Instant::from_std(wake_at)) => None,
        };
        if let Some(result) = recv_result {
            match result {
                Ok(events) => events.iter().for_each(print_event),
                Err(err) => {
                    eprintln!("receive failed: {err}");
                    break;
                }
            }
        }

        if Instant::now() < end && Instant::now() >= next_send {
            match client.send_probe().await {
                Ok(receipt) => {
                    print_event(&receipt.into());
                    while next_send <= Instant::now() {
                        next_send += interval;
                    }
                }
                Err(err) => {
                    if let irtt_client::SendProbeError::AfterCommit { receipt, .. } = &err {
                        print_event(&(*receipt).into());
                    }
                    eprintln!("send failed: {err}");
                    break;
                }
            }
        }
        match client.poll_timeouts() {
            Ok(events) => events.iter().for_each(print_event),
            Err(err) => {
                eprintln!("timeout polling failed: {err}");
                break;
            }
        }
    }

    if client.is_peer_closed() {
        return;
    }
    match client.close().await {
        Ok(events) => events.iter().for_each(print_event),
        Err(err) => eprintln!("close failed: {err}"),
    }
}

pub fn poll_timeouts(&mut self) -> Result<Vec<ClientEvent>, ClientError>

Poll protocol timeouts using a newly captured monotonic timestamp.

Examples found in repository

examples/tokio_client.rs (line 139)

async fn run() {
    let config = ClientConfig {
        open: OpenPolicy {
            // A single short attempt keeps this example fast when no server is
            // reachable; production callers should generally keep the default.
            timeouts: vec![Duration::from_millis(300)],
            ..Default::default()
        },
        request: SessionRequest {
            duration: Some(Duration::from_secs(2)),
            interval: Duration::from_millis(200),
            ..Default::default()
        },
        ..Default::default()
    };

    let mut client = match AsyncClient::connect("127.0.0.1:2112", config).await {
        Ok(client) => client,
        Err(err) => {
            eprintln!("failed to prepare client socket: {err}");
            return;
        }
    };

    let negotiated = match client.open().await {
        Ok(OpenOutcome::Started(started)) => {
            println!("session opened: {started:?}");
            started.negotiation
        }
        Ok(OpenOutcome::NoTestCompleted(completed)) => {
            println!("{completed:?}");
            return;
        }
        Err(err) => {
            eprintln!("open failed (is a server running at 127.0.0.1:2112?): {err}");
            return;
        }
    };

    // This loop owns cadence and duration. The transport sends whenever called.
    let interval = negotiated.accepted.interval;
    let start = Instant::now();
    let end = start
        + negotiated
            .accepted
            .duration
            .expect("example requests a finite run");
    let deadline = end + client.probe_timeout();
    let mut next_send = start;
    while Instant::now() < deadline && !client.is_peer_closed() {
        if Instant::now() >= end && !client.has_pending_probes() {
            break;
        }
        // Bound the receive wait by the next send/timeout deadline, rather
        // than awaiting recv() unconditionally: a lost or rate-limited reply
        // must not stall pacing or timeout classification indefinitely.
        let wake_at = (next_send < end)
            .then_some(next_send)
            .into_iter()
            .chain(client.next_probe_timeout_deadline())
            .chain(std::iter::once(Instant::now() + TIMEOUT_POLL_INTERVAL))
            .min()
            .unwrap();

        let recv_result = tokio::select! {
            events = client.recv() => Some(events),
            () = tokio::time::sleep_until(tokio::time::Instant::from_std(wake_at)) => None,
        };
        if let Some(result) = recv_result {
            match result {
                Ok(events) => events.iter().for_each(print_event),
                Err(err) => {
                    eprintln!("receive failed: {err}");
                    break;
                }
            }
        }

        if Instant::now() < end && Instant::now() >= next_send {
            match client.send_probe().await {
                Ok(receipt) => {
                    print_event(&receipt.into());
                    while next_send <= Instant::now() {
                        next_send += interval;
                    }
                }
                Err(err) => {
                    if let irtt_client::SendProbeError::AfterCommit { receipt, .. } = &err {
                        print_event(&(*receipt).into());
                    }
                    eprintln!("send failed: {err}");
                    break;
                }
            }
        }
        match client.poll_timeouts() {
            Ok(events) => events.iter().for_each(print_event),
            Err(err) => {
                eprintln!("timeout polling failed: {err}");
                break;
            }
        }
    }

    if client.is_peer_closed() {
        return;
    }
    match client.close().await {
        Ok(events) => events.iter().for_each(print_event),
        Err(err) => eprintln!("close failed: {err}"),
    }
}

pub fn poll_timeouts_at( &mut self, now: Instant, ) -> Result<Vec<ClientEvent>, ClientError>

Poll protocol timeouts using the caller’s monotonic timestamp.

pub async fn close(&mut self) -> Result<Vec<ClientEvent>, ClientError>

Send the retained close packet and commit local close exactly once.

DSCP is cleared only after write readiness and immediately before the nonblocking send attempt. A WouldBlock result restores it before the next await, so dropping this future at a suspension point cannot leave an otherwise-open session with cleared DSCP.

Examples found in repository

examples/tokio_client.rs (line 151)

async fn run() {
    let config = ClientConfig {
        open: OpenPolicy {
            // A single short attempt keeps this example fast when no server is
            // reachable; production callers should generally keep the default.
            timeouts: vec![Duration::from_millis(300)],
            ..Default::default()
        },
        request: SessionRequest {
            duration: Some(Duration::from_secs(2)),
            interval: Duration::from_millis(200),
            ..Default::default()
        },
        ..Default::default()
    };

    let mut client = match AsyncClient::connect("127.0.0.1:2112", config).await {
        Ok(client) => client,
        Err(err) => {
            eprintln!("failed to prepare client socket: {err}");
            return;
        }
    };

    let negotiated = match client.open().await {
        Ok(OpenOutcome::Started(started)) => {
            println!("session opened: {started:?}");
            started.negotiation
        }
        Ok(OpenOutcome::NoTestCompleted(completed)) => {
            println!("{completed:?}");
            return;
        }
        Err(err) => {
            eprintln!("open failed (is a server running at 127.0.0.1:2112?): {err}");
            return;
        }
    };

    // This loop owns cadence and duration. The transport sends whenever called.
    let interval = negotiated.accepted.interval;
    let start = Instant::now();
    let end = start
        + negotiated
            .accepted
            .duration
            .expect("example requests a finite run");
    let deadline = end + client.probe_timeout();
    let mut next_send = start;
    while Instant::now() < deadline && !client.is_peer_closed() {
        if Instant::now() >= end && !client.has_pending_probes() {
            break;
        }
        // Bound the receive wait by the next send/timeout deadline, rather
        // than awaiting recv() unconditionally: a lost or rate-limited reply
        // must not stall pacing or timeout classification indefinitely.
        let wake_at = (next_send < end)
            .then_some(next_send)
            .into_iter()
            .chain(client.next_probe_timeout_deadline())
            .chain(std::iter::once(Instant::now() + TIMEOUT_POLL_INTERVAL))
            .min()
            .unwrap();

        let recv_result = tokio::select! {
            events = client.recv() => Some(events),
            () = tokio::time::sleep_until(tokio::time::Instant::from_std(wake_at)) => None,
        };
        if let Some(result) = recv_result {
            match result {
                Ok(events) => events.iter().for_each(print_event),
                Err(err) => {
                    eprintln!("receive failed: {err}");
                    break;
                }
            }
        }

        if Instant::now() < end && Instant::now() >= next_send {
            match client.send_probe().await {
                Ok(receipt) => {
                    print_event(&receipt.into());
                    while next_send <= Instant::now() {
                        next_send += interval;
                    }
                }
                Err(err) => {
                    if let irtt_client::SendProbeError::AfterCommit { receipt, .. } = &err {
                        print_event(&(*receipt).into());
                    }
                    eprintln!("send failed: {err}");
                    break;
                }
            }
        }
        match client.poll_timeouts() {
            Ok(events) => events.iter().for_each(print_event),
            Err(err) => {
                eprintln!("timeout polling failed: {err}");
                break;
            }
        }
    }

    if client.is_peer_closed() {
        return;
    }
    match client.close().await {
        Ok(events) => events.iter().for_each(print_event),
        Err(err) => eprintln!("close failed: {err}"),
    }
}

pub fn probe_timeout(&self) -> Duration

Return the configured local probe timeout.

Examples found in repository

examples/tokio_client.rs (line 91)

async fn run() {
    let config = ClientConfig {
        open: OpenPolicy {
            // A single short attempt keeps this example fast when no server is
            // reachable; production callers should generally keep the default.
            timeouts: vec![Duration::from_millis(300)],
            ..Default::default()
        },
        request: SessionRequest {
            duration: Some(Duration::from_secs(2)),
            interval: Duration::from_millis(200),
            ..Default::default()
        },
        ..Default::default()
    };

    let mut client = match AsyncClient::connect("127.0.0.1:2112", config).await {
        Ok(client) => client,
        Err(err) => {
            eprintln!("failed to prepare client socket: {err}");
            return;
        }
    };

    let negotiated = match client.open().await {
        Ok(OpenOutcome::Started(started)) => {
            println!("session opened: {started:?}");
            started.negotiation
        }
        Ok(OpenOutcome::NoTestCompleted(completed)) => {
            println!("{completed:?}");
            return;
        }
        Err(err) => {
            eprintln!("open failed (is a server running at 127.0.0.1:2112?): {err}");
            return;
        }
    };

    // This loop owns cadence and duration. The transport sends whenever called.
    let interval = negotiated.accepted.interval;
    let start = Instant::now();
    let end = start
        + negotiated
            .accepted
            .duration
            .expect("example requests a finite run");
    let deadline = end + client.probe_timeout();
    let mut next_send = start;
    while Instant::now() < deadline && !client.is_peer_closed() {
        if Instant::now() >= end && !client.has_pending_probes() {
            break;
        }
        // Bound the receive wait by the next send/timeout deadline, rather
        // than awaiting recv() unconditionally: a lost or rate-limited reply
        // must not stall pacing or timeout classification indefinitely.
        let wake_at = (next_send < end)
            .then_some(next_send)
            .into_iter()
            .chain(client.next_probe_timeout_deadline())
            .chain(std::iter::once(Instant::now() + TIMEOUT_POLL_INTERVAL))
            .min()
            .unwrap();

        let recv_result = tokio::select! {
            events = client.recv() => Some(events),
            () = tokio::time::sleep_until(tokio::time::Instant::from_std(wake_at)) => None,
        };
        if let Some(result) = recv_result {
            match result {
                Ok(events) => events.iter().for_each(print_event),
                Err(err) => {
                    eprintln!("receive failed: {err}");
                    break;
                }
            }
        }

        if Instant::now() < end && Instant::now() >= next_send {
            match client.send_probe().await {
                Ok(receipt) => {
                    print_event(&receipt.into());
                    while next_send <= Instant::now() {
                        next_send += interval;
                    }
                }
                Err(err) => {
                    if let irtt_client::SendProbeError::AfterCommit { receipt, .. } = &err {
                        print_event(&(*receipt).into());
                    }
                    eprintln!("send failed: {err}");
                    break;
                }
            }
        }
        match client.poll_timeouts() {
            Ok(events) => events.iter().for_each(print_event),
            Err(err) => {
                eprintln!("timeout polling failed: {err}");
                break;
            }
        }
    }

    if client.is_peer_closed() {
        return;
    }
    match client.close().await {
        Ok(events) => events.iter().for_each(print_event),
        Err(err) => eprintln!("close failed: {err}"),
    }
}

pub fn negotiation(&self) -> Option<&NegotiationResult>

Accepted semantics and exact peer parameters of the currently live session.

Returns None before Open, after Close, and after no-test completion.

pub fn has_pending_probes(&self) -> bool

Whether any sent probes still await a reply or timeout.

Examples found in repository

examples/tokio_client.rs (line 94)

async fn run() {
    let config = ClientConfig {
        open: OpenPolicy {
            // A single short attempt keeps this example fast when no server is
            // reachable; production callers should generally keep the default.
            timeouts: vec![Duration::from_millis(300)],
            ..Default::default()
        },
        request: SessionRequest {
            duration: Some(Duration::from_secs(2)),
            interval: Duration::from_millis(200),
            ..Default::default()
        },
        ..Default::default()
    };

    let mut client = match AsyncClient::connect("127.0.0.1:2112", config).await {
        Ok(client) => client,
        Err(err) => {
            eprintln!("failed to prepare client socket: {err}");
            return;
        }
    };

    let negotiated = match client.open().await {
        Ok(OpenOutcome::Started(started)) => {
            println!("session opened: {started:?}");
            started.negotiation
        }
        Ok(OpenOutcome::NoTestCompleted(completed)) => {
            println!("{completed:?}");
            return;
        }
        Err(err) => {
            eprintln!("open failed (is a server running at 127.0.0.1:2112?): {err}");
            return;
        }
    };

    // This loop owns cadence and duration. The transport sends whenever called.
    let interval = negotiated.accepted.interval;
    let start = Instant::now();
    let end = start
        + negotiated
            .accepted
            .duration
            .expect("example requests a finite run");
    let deadline = end + client.probe_timeout();
    let mut next_send = start;
    while Instant::now() < deadline && !client.is_peer_closed() {
        if Instant::now() >= end && !client.has_pending_probes() {
            break;
        }
        // Bound the receive wait by the next send/timeout deadline, rather
        // than awaiting recv() unconditionally: a lost or rate-limited reply
        // must not stall pacing or timeout classification indefinitely.
        let wake_at = (next_send < end)
            .then_some(next_send)
            .into_iter()
            .chain(client.next_probe_timeout_deadline())
            .chain(std::iter::once(Instant::now() + TIMEOUT_POLL_INTERVAL))
            .min()
            .unwrap();

        let recv_result = tokio::select! {
            events = client.recv() => Some(events),
            () = tokio::time::sleep_until(tokio::time::Instant::from_std(wake_at)) => None,
        };
        if let Some(result) = recv_result {
            match result {
                Ok(events) => events.iter().for_each(print_event),
                Err(err) => {
                    eprintln!("receive failed: {err}");
                    break;
                }
            }
        }

        if Instant::now() < end && Instant::now() >= next_send {
            match client.send_probe().await {
                Ok(receipt) => {
                    print_event(&receipt.into());
                    while next_send <= Instant::now() {
                        next_send += interval;
                    }
                }
                Err(err) => {
                    if let irtt_client::SendProbeError::AfterCommit { receipt, .. } = &err {
                        print_event(&(*receipt).into());
                    }
                    eprintln!("send failed: {err}");
                    break;
                }
            }
        }
        match client.poll_timeouts() {
            Ok(events) => events.iter().for_each(print_event),
            Err(err) => {
                eprintln!("timeout polling failed: {err}");
                break;
            }
        }
    }

    if client.is_peer_closed() {
        return;
    }
    match client.close().await {
        Ok(events) => events.iter().for_each(print_event),
        Err(err) => eprintln!("close failed: {err}"),
    }
}

pub fn is_peer_closed(&self) -> bool

Return whether an authenticated peer close ended the session.

Examples found in repository

examples/tokio_client.rs (line 93)

async fn run() {
    let config = ClientConfig {
        open: OpenPolicy {
            // A single short attempt keeps this example fast when no server is
            // reachable; production callers should generally keep the default.
            timeouts: vec![Duration::from_millis(300)],
            ..Default::default()
        },
        request: SessionRequest {
            duration: Some(Duration::from_secs(2)),
            interval: Duration::from_millis(200),
            ..Default::default()
        },
        ..Default::default()
    };

    let mut client = match AsyncClient::connect("127.0.0.1:2112", config).await {
        Ok(client) => client,
        Err(err) => {
            eprintln!("failed to prepare client socket: {err}");
            return;
        }
    };

    let negotiated = match client.open().await {
        Ok(OpenOutcome::Started(started)) => {
            println!("session opened: {started:?}");
            started.negotiation
        }
        Ok(OpenOutcome::NoTestCompleted(completed)) => {
            println!("{completed:?}");
            return;
        }
        Err(err) => {
            eprintln!("open failed (is a server running at 127.0.0.1:2112?): {err}");
            return;
        }
    };

    // This loop owns cadence and duration. The transport sends whenever called.
    let interval = negotiated.accepted.interval;
    let start = Instant::now();
    let end = start
        + negotiated
            .accepted
            .duration
            .expect("example requests a finite run");
    let deadline = end + client.probe_timeout();
    let mut next_send = start;
    while Instant::now() < deadline && !client.is_peer_closed() {
        if Instant::now() >= end && !client.has_pending_probes() {
            break;
        }
        // Bound the receive wait by the next send/timeout deadline, rather
        // than awaiting recv() unconditionally: a lost or rate-limited reply
        // must not stall pacing or timeout classification indefinitely.
        let wake_at = (next_send < end)
            .then_some(next_send)
            .into_iter()
            .chain(client.next_probe_timeout_deadline())
            .chain(std::iter::once(Instant::now() + TIMEOUT_POLL_INTERVAL))
            .min()
            .unwrap();

        let recv_result = tokio::select! {
            events = client.recv() => Some(events),
            () = tokio::time::sleep_until(tokio::time::Instant::from_std(wake_at)) => None,
        };
        if let Some(result) = recv_result {
            match result {
                Ok(events) => events.iter().for_each(print_event),
                Err(err) => {
                    eprintln!("receive failed: {err}");
                    break;
                }
            }
        }

        if Instant::now() < end && Instant::now() >= next_send {
            match client.send_probe().await {
                Ok(receipt) => {
                    print_event(&receipt.into());
                    while next_send <= Instant::now() {
                        next_send += interval;
                    }
                }
                Err(err) => {
                    if let irtt_client::SendProbeError::AfterCommit { receipt, .. } = &err {
                        print_event(&(*receipt).into());
                    }
                    eprintln!("send failed: {err}");
                    break;
                }
            }
        }
        match client.poll_timeouts() {
            Ok(events) => events.iter().for_each(print_event),
            Err(err) => {
                eprintln!("timeout polling failed: {err}");
                break;
            }
        }
    }

    if client.is_peer_closed() {
        return;
    }
    match client.close().await {
        Ok(events) => events.iter().for_each(print_event),
        Err(err) => eprintln!("close failed: {err}"),
    }
}

pub fn next_probe_timeout_deadline(&self) -> Option<Instant>

Earliest timeout deadline among probes still awaiting a reply.

Examples found in repository

examples/tokio_client.rs (line 103)

async fn run() {
    let config = ClientConfig {
        open: OpenPolicy {
            // A single short attempt keeps this example fast when no server is
            // reachable; production callers should generally keep the default.
            timeouts: vec![Duration::from_millis(300)],
            ..Default::default()
        },
        request: SessionRequest {
            duration: Some(Duration::from_secs(2)),
            interval: Duration::from_millis(200),
            ..Default::default()
        },
        ..Default::default()
    };

    let mut client = match AsyncClient::connect("127.0.0.1:2112", config).await {
        Ok(client) => client,
        Err(err) => {
            eprintln!("failed to prepare client socket: {err}");
            return;
        }
    };

    let negotiated = match client.open().await {
        Ok(OpenOutcome::Started(started)) => {
            println!("session opened: {started:?}");
            started.negotiation
        }
        Ok(OpenOutcome::NoTestCompleted(completed)) => {
            println!("{completed:?}");
            return;
        }
        Err(err) => {
            eprintln!("open failed (is a server running at 127.0.0.1:2112?): {err}");
            return;
        }
    };

    // This loop owns cadence and duration. The transport sends whenever called.
    let interval = negotiated.accepted.interval;
    let start = Instant::now();
    let end = start
        + negotiated
            .accepted
            .duration
            .expect("example requests a finite run");
    let deadline = end + client.probe_timeout();
    let mut next_send = start;
    while Instant::now() < deadline && !client.is_peer_closed() {
        if Instant::now() >= end && !client.has_pending_probes() {
            break;
        }
        // Bound the receive wait by the next send/timeout deadline, rather
        // than awaiting recv() unconditionally: a lost or rate-limited reply
        // must not stall pacing or timeout classification indefinitely.
        let wake_at = (next_send < end)
            .then_some(next_send)
            .into_iter()
            .chain(client.next_probe_timeout_deadline())
            .chain(std::iter::once(Instant::now() + TIMEOUT_POLL_INTERVAL))
            .min()
            .unwrap();

        let recv_result = tokio::select! {
            events = client.recv() => Some(events),
            () = tokio::time::sleep_until(tokio::time::Instant::from_std(wake_at)) => None,
        };
        if let Some(result) = recv_result {
            match result {
                Ok(events) => events.iter().for_each(print_event),
                Err(err) => {
                    eprintln!("receive failed: {err}");
                    break;
                }
            }
        }

        if Instant::now() < end && Instant::now() >= next_send {
            match client.send_probe().await {
                Ok(receipt) => {
                    print_event(&receipt.into());
                    while next_send <= Instant::now() {
                        next_send += interval;
                    }
                }
                Err(err) => {
                    if let irtt_client::SendProbeError::AfterCommit { receipt, .. } = &err {
                        print_event(&(*receipt).into());
                    }
                    eprintln!("send failed: {err}");
                    break;
                }
            }
        }
        match client.poll_timeouts() {
            Ok(events) => events.iter().for_each(print_event),
            Err(err) => {
                eprintln!("timeout polling failed: {err}");
                break;
            }
        }
    }

    if client.is_peer_closed() {
        return;
    }
    match client.close().await {
        Ok(events) => events.iter().for_each(print_event),
        Err(err) => eprintln!("close failed: {err}"),
    }
}

Trait Implementations

impl Debug for AsyncClient

fn fmt(&self, f: &mut Formatter<'_>) -> Result

Formats the value using the given formatter.