udx 0.2.0

Rust port of libudx, a protocol for reliable, multiplex, and congestion controlled streams over udp
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
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159

use std::{io, time::Duration};
use tokio::io::{AsyncReadExt, AsyncWriteExt};
use udx::{UdxSocket, UdxStream};

#[tokio::test]
async fn stream_read_write() -> io::Result<()> {
    eprintln!("ok go");
    run().await?;
    // drop(streama);
    // drop(streamb);
    // drop(socka);
    // drop(sockb);
    eprintln!("wait");
    tokio::time::sleep(Duration::from_secs(1)).await;
    eprintln!("done");
    Ok(())
}

async fn run() -> io::Result<()> {
    let ((socka, _sockb), (mut streama, mut streamb)) = create_pair().await?;
    assert_eq!(socka.local_addr().unwrap(), streamb.remote_addr());

    let msg = vec![1, 2, 3];
    streama.write_all(&msg).await?;
    let mut read = vec![0u8; 3];
    streamb.read_exact(&mut read).await?;
    assert_eq!(msg, read);
    eprintln!("now drop");
    Ok(())
}

#[tokio::test]
async fn stream_close() -> io::Result<()> {
    let ((socka, _sockb), (mut streama, mut streamb)) = create_pair().await?;
    assert_eq!(socka.local_addr().unwrap(), streamb.remote_addr());

    // write a message.
    let msg = vec![1, 2, 3];
    streama.write_all(&msg).await?;
    assert_eq!(streama.stats().inflight_packets, 1, "inflight 1 after send");

    // close the stream
    let close = streama.close();
    assert_eq!(
        streama.stats().inflight_packets,
        1,
        "inflight still 1 after send"
    );

    // wait until closing is complete == all packages flushed
    close.await;
    assert_eq!(
        streama.stats().inflight_packets,
        0,
        "inflight 0 after close await"
    );

    // ensure reading on other end still works
    let mut read = vec![0u8; 3];
    let res = streamb.read_exact(&mut read).await;
    let _res = res?;
    assert_eq!(msg, read, "read ok");

    // try to write on closed stream
    let res = streama.write_all(&msg).await;
    assert_eq!(
        res.err().unwrap().kind(),
        io::ErrorKind::ConnectionReset,
        "stream closed"
    );
    // try to read on closed stream
    let res = streama.read(&mut read).await;
    assert_eq!(
        res.err().unwrap().kind(),
        io::ErrorKind::ConnectionReset,
        "stream closed"
    );

    Ok(())
}

async fn create_pair() -> io::Result<((UdxSocket, UdxSocket), (UdxStream, UdxStream))> {
    let socka = UdxSocket::bind("127.0.0.1:0")?;
    let sockb = UdxSocket::bind("127.0.0.1:0")?;
    let addra = socka.local_addr()?;
    let addrb = sockb.local_addr()?;
    let streama = socka.connect(addrb, 1, 2)?;
    let streamb = sockb.connect(addra, 2, 1)?;
    Ok(((socka, sockb), (streama, streamb)))
}

#[tokio::test]
async fn test_sockets() -> io::Result<()> {
    let a = UdxSocket::bind("127.0.0.1:0")?;
    let b = UdxSocket::bind("127.0.0.1:0")?;

    a.send(b.local_addr()?, b"boo");
    let (_sender, msg) = b.recv().await?;
    assert_eq!(msg, b"boo");
    Ok(())
}

#[tokio::test]
async fn test_stream_msg_replayed_on_connect() -> io::Result<()> {
    let aid = 66;
    let bid = 77;
    let a = UdxSocket::bind("127.0.0.1:0")?;
    let b = UdxSocket::bind("127.0.0.1:0")?;

    let mut bstr = b.connect(a.local_addr()?, bid, aid)?;
    bstr.write_all(b"AAA").await?;
    // b"AAA" seen as socket msg bc other stream doesn't exist yet
    let (_, _x) = a.recv().await?;

    let mut astr = a.connect(b.local_addr()?, aid, bid)?;

    bstr.write_all(b"BBB").await?;
    let mut buf = vec![];
    astr.read_buf(&mut buf).await?;
    assert_eq!(&buf, b"AAABBB");
    Ok(())
}

#[tokio::test]
async fn test_halfopen_streams() -> io::Result<()> {
    let aid = 66;
    let bid = 77;
    let a = UdxSocket::bind("127.0.0.1:0")?;
    let b = UdxSocket::bind("127.0.0.1:0")?;

    let mut bstr = b.connect(a.local_addr()?, bid, aid)?;
    bstr.write_all(b"AAA").await?;
    let (_, _x) = a.recv().await?;
    dbg!(_x);

    let a_half_str = a.create_stream(aid)?;

    bstr.write_all(b"BBB").await?;

    // b_stream message not seen as socket message to a bc a_stream is half open
    assert!(
        tokio::time::timeout(Duration::from_millis(100), a.recv())
            .await
            .is_err()
    );

    let mut astr = a_half_str.connect(b.local_addr()?, bid)?;
    bstr.write_all(b"CCC").await?;
    // b_stream message not seen as socket message to a bc a_stream is open
    assert!(
        tokio::time::timeout(Duration::from_millis(100), a.recv())
            .await
            .is_err()
    );
    let mut buf = vec![];
    astr.read_buf(&mut buf).await?;
    assert_eq!(&buf, b"AAABBBCCC");
    Ok(())
}