netsim 0.3.0

Run tests in network-isolated threads. Intercept and meddle with their packets.
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

use crate::priv_prelude::*;

#[tokio::test(flavor = "multi_thread")]
async fn zero_random_delay_in_order() {
    const NUM_MSGS: usize = 1000;
    const MIN_DELAY: Duration = Duration::from_millis(200);

    let (chan_0, mut chan_1) = BiChannel::new(NUM_MSGS);
    let mut chan_0 = Box::pin(chan_0.with_delay(MIN_DELAY, Duration::ZERO));
    let sender_0 = async move {
        for val in 0..NUM_MSGS {
            chan_0.feed(val).await.unwrap();
        }
        chan_0.flush().await.unwrap();
        chan_0
    };
    let receiver_1 = async move {
        for expected_val in 0..NUM_MSGS {
            let val = chan_1.next().await.unwrap();
            assert_eq!(val, expected_val);
        }
        chan_1
    };
    let sender_0 = tokio::spawn(sender_0);
    let receiver_1 = tokio::spawn(receiver_1);
    let (chan_0_res, chan_1_res) = join!(sender_0, receiver_1);
    let mut chan_0 = chan_0_res.unwrap();
    let mut chan_1 = chan_1_res.unwrap();
    
    let sender_1 = async move {
        for val in 0..NUM_MSGS {
            chan_1.feed(val).await.unwrap();
        }
        chan_1.flush().await.unwrap();
        chan_1
    };
    let receiver_0 = async move {
        for expected_val in 0..NUM_MSGS {
            let val = chan_0.next().await.unwrap();
            assert_eq!(val, expected_val);
        }
        chan_0
    };
    let sender_1 = tokio::spawn(sender_1);
    let receiver_0 = tokio::spawn(receiver_0);
    let (chan_0_res, chan_1_res) = join!(receiver_0, sender_1);
    let mut chan_0 = chan_0_res.unwrap();
    let chan_1 = chan_1_res.unwrap();

    drop(chan_1);
    assert!(chan_0.next().await.is_none());
}

#[tokio::test(flavor = "multi_thread")]
async fn random_delays_are_approx_correct() {
    const NUM_MSGS: usize = 1_000;
    const MIN_DELAY: Duration = Duration::from_millis(500);
    const MEAN_RANDOM_DELAY: Duration = Duration::from_millis(500);

    let (chan_0, mut chan_1) = BiChannel::new(NUM_MSGS);
    let mut chan_0 = Box::pin(chan_0.with_delay(MIN_DELAY, MEAN_RANDOM_DELAY));

    let check_delays = |delays: Vec<Duration>| {
        let min_delay = delays.iter().copied().min().unwrap();
        assert!(MIN_DELAY <= min_delay);
        assert!(min_delay <= MIN_DELAY * 2);

        let total_random_delays: f64 = {
            delays
            .into_iter()
            .map(|delay| delay - min_delay)
            .map(|random_delay| random_delay.as_secs_f64())
            .sum()
        };
        let mean_random_delay = total_random_delays / (NUM_MSGS as f64);
        let expected_mean_random_delay = MEAN_RANDOM_DELAY.as_secs_f64();
        assert!(expected_mean_random_delay * 0.8 < mean_random_delay);
        assert!(mean_random_delay < expected_mean_random_delay * 1.2);
    };

    let mut delays = Vec::with_capacity(NUM_MSGS);
    let sender = async move {
        for _ in 0..NUM_MSGS {
            let send_instant = Instant::now();
            chan_0.feed(send_instant).await.unwrap();
        }
        chan_0.flush().await.unwrap();
        chan_0
    };
    let receiver = async move {
        for _ in 0..NUM_MSGS {
            let send_instant = chan_1.next().await.unwrap();
            let delay = Instant::now() - send_instant;
            delays.push(delay);
        }
        (chan_1, delays)
    };
    let receiver = tokio::spawn(receiver);
    tokio::time::sleep(MIN_DELAY).await;
    let sender = tokio::spawn(sender);
    let (chan_0_res, chan_1_delays_res) = join!(sender, receiver);
    let mut chan_0 = chan_0_res.unwrap();
    let (mut chan_1, delays) = chan_1_delays_res.unwrap();
    check_delays(delays);

    let mut delays = Vec::with_capacity(NUM_MSGS);
    let sender = async move {
        for _ in 0..NUM_MSGS {
            let send_instant = Instant::now();
            chan_1.feed(send_instant).await.unwrap();
        }
        chan_1.flush().await.unwrap();
        chan_1
    };
    let receiver = async move {
        for _ in 0..NUM_MSGS {
            let send_instant = chan_0.next().await.unwrap();
            let delay = Instant::now() - send_instant;
            delays.push(delay);
        }
        (chan_0, delays)
    };
    let receiver = tokio::spawn(receiver);
    tokio::time::sleep(MIN_DELAY).await;
    let sender = tokio::spawn(sender);
    let (chan_1_res, chan_0_delays_res) = join!(sender, receiver);
    let chan_1 = chan_1_res.unwrap();
    let (chan_0, delays) = chan_0_delays_res.unwrap();
    check_delays(delays);
    
    drop((chan_0, chan_1));
}