dope 0.2.3

Thin io_uring adaptor with "Manifolds"
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

use dope_futures::AsyncRead;
use dope_futures::listener::Listener;
use dope_futures::stream::Stream;
use dope_runtime::{DefaultBackend, block_on, spawn};
use dope_transport::Tcp;

#[test]
fn echo_basic() {
    let mut listener = Listener::<Tcp, DefaultBackend>::bind("127.0.0.1:0").unwrap();
    let addr = listener.local_addr().unwrap();

    block_on(async move {
        let server = spawn(async move {
            let (mut tcp, _) = listener.accept().await.unwrap();
            let mut buf = [0u8; 12];
            tcp.read_exact(&mut buf).await.unwrap();
            tcp.write_all(&buf).await.unwrap();
        });

        let mut tcp = Stream::<Tcp, DefaultBackend>::connect_with(addr)
            .await
            .unwrap();
        let msg = b"hello-dope!!";
        tcp.write_all(msg).await.unwrap();
        let mut out = [0u8; 12];
        tcp.read_exact(&mut out).await.unwrap();
        assert_eq!(&out, msg);

        server.await;
    });
}

#[test]
fn echo_large_payload() {
    const SIZE: usize = 128 * 1024;

    let mut listener = Listener::<Tcp, DefaultBackend>::bind("127.0.0.1:0").unwrap();
    let addr = listener.local_addr().unwrap();

    block_on(async move {
        let server = spawn(async move {
            let (mut tcp, _) = listener.accept().await.unwrap();
            let mut received = Vec::with_capacity(SIZE);
            while received.len() < SIZE {
                let want = (SIZE - received.len()).min(16384);
                let (res, buf) = tcp.read(vec![0u8; want]).await;
                let n = res.unwrap();
                if n == 0 {
                    break;
                }
                received.extend_from_slice(&buf[..n]);
            }
            received
        });

        let mut tcp = Stream::<Tcp, DefaultBackend>::connect_with(addr)
            .await
            .unwrap();
        let payload: Vec<u8> = (0..SIZE).map(|i| (i % 251) as u8).collect();
        let (res, _) = tcp.write_all_owned(payload.clone()).await;
        res.unwrap();

        let received = server.await;
        assert_eq!(received.len(), SIZE);
        assert_eq!(received, payload);
    });
}

#[test]
fn echo_concurrent_connections() {
    const CONNS: usize = 16;
    const MSG_LEN: usize = 32;

    let mut listener = Listener::<Tcp, DefaultBackend>::bind("127.0.0.1:0").unwrap();
    let addr = listener.local_addr().unwrap();

    block_on(async move {
        let server = spawn(async move {
            for _ in 0..CONNS {
                let (mut tcp, _) = listener.accept().await.unwrap();
                spawn(async move {
                    let mut buf = [0u8; MSG_LEN];
                    tcp.read_exact(&mut buf).await.unwrap();
                    tcp.write_all(&buf).await.unwrap();
                });
            }
        });

        for i in 0..CONNS {
            let mut tcp = Stream::<Tcp, DefaultBackend>::connect_with(addr)
                .await
                .unwrap();
            let mut msg = [0u8; MSG_LEN];
            for (j, b) in msg.iter_mut().enumerate() {
                *b = ((i * 17 + j) % 251) as u8;
            }
            tcp.write_all(&msg).await.unwrap();
            let mut out = [0u8; MSG_LEN];
            tcp.read_exact(&mut out).await.unwrap();
            assert_eq!(out, msg);
        }

        server.await;
    });
}