msrt-udp 0.1.0

Tokio UDP adapters for MSRT.
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83

#![allow(clippy::std_instead_of_core)]

use std::env;
use std::io::ErrorKind;
use std::time::{Duration, Instant};

use msrt_udp::{EngineConfig, UdpClient, UdpClientEvent};
use tokio::time::sleep;

const DEFAULT_SERVER: &str = "127.0.0.1:9000";
const SEND_INTERVAL: Duration = Duration::from_secs(1);
const LOOP_SLEEP: Duration = Duration::from_millis(10);

#[tokio::main]
async fn main() -> msrt_udp::Result<()> {
    let server = env::args()
        .nth(1)
        .unwrap_or_else(|| DEFAULT_SERVER.to_string());

    let mut client = UdpClient::bind_with_config("127.0.0.1:0", &server, demo_config()).await?;
    println!(
        "frontend local={} remote={}",
        client.local_addr()?,
        client.peer_addr()?
    );

    reconnect(&mut client)?;
    let mut sequence = 0_u64;
    let mut next_send = Instant::now();

    loop {
        if next_send <= Instant::now() {
            let payload = format!("hello udp {sequence}");
            if client.send(payload.as_bytes())? {
                println!("queued: {payload}");
                sequence = sequence.wrapping_add(1);
            }
            next_send = Instant::now() + SEND_INTERVAL;
        }

        match client.tick().await? {
            UdpClientEvent::Message(message) if message.as_bytes() != [0] => {
                println!("message: {}", String::from_utf8_lossy(message.as_bytes()));
            }
            UdpClientEvent::Message(_) | UdpClientEvent::Idle => {}
            UdpClientEvent::SendFailed(failed) => {
                println!("send failed: {failed:?}; reconnecting");
                client.disconnect();
                reconnect(&mut client)?;
                next_send = Instant::now();
            }
            UdpClientEvent::TransportUnavailable { kind } => {
                println!("transport unavailable: {kind:?}; reconnecting");
                client.disconnect();
                reconnect(&mut client)?;
                next_send = Instant::now() + reconnect_delay(kind);
            }
        }

        sleep(LOOP_SLEEP).await;
    }
}

fn reconnect_delay(kind: ErrorKind) -> Duration {
    match kind {
        ErrorKind::ConnectionRefused | ErrorKind::ConnectionReset => Duration::from_millis(250),
        _ => Duration::from_secs(1),
    }
}

fn reconnect(client: &mut UdpClient) -> msrt_udp::Result<()> {
    client.connect()?;
    println!("session started");
    Ok(())
}

fn demo_config() -> EngineConfig {
    EngineConfig {
        retransmit_timeout_ms: 250,
        max_retransmit_attempts: 8,
        ..EngineConfig::default()
    }
}