nbio 0.8.2

Non-Blocking I/O
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

use std::{
    sync::atomic::{AtomicUsize, Ordering},
    time::{Duration, SystemTime},
};

use nbio::{
    liveness::{HeartbeatOutcome, HeartbeatingSession, LivenessSession, LivenessStrategy},
    mock::MockSession,
    DriveOutcome, Publish, Receive, Session,
};

#[test]
fn test_heartbeat() {
    let seq = AtomicUsize::new(0);
    let mut sess = HeartbeatingSession::new(
        MockSession::<&[usize], Vec<usize>>::new(),
        Duration::from_millis(50),
        move |s| {
            s.publish(vec![seq.fetch_add(1, Ordering::AcqRel)])?;
            Ok(HeartbeatOutcome::Sent)
        },
    );

    // drive for more than 3 intervals
    let end = SystemTime::now() + Duration::from_millis(151);
    while SystemTime::now() < end {
        sess.drive().unwrap();
        std::thread::sleep(Duration::from_millis(1));
    }
    sess.drive().unwrap();

    // assert at least 2 and at most 5 heartbeats received
    assert_eq!(
        0,
        *sess
            .session_mut()
            .publish_queue
            .pop_front()
            .unwrap()
            .get(0)
            .unwrap()
    );
    assert_eq!(
        1,
        *sess
            .session_mut()
            .publish_queue
            .pop_front()
            .unwrap()
            .get(0)
            .unwrap()
    );
    assert!(sess.session().publish_queue.len() <= 3);
}

#[test]
fn test_liveness() {
    let mut sess = LivenessSession::new(
        MockSession::<&[u8], &[u8]>::new(),
        Duration::from_millis(50),
        LivenessStrategy::default(),
    );

    // send and drive for longer than interval
    let end = SystemTime::now() + Duration::from_millis(100);
    while SystemTime::now() < end {
        sess.session_mut().receive_queue.push_back(&[0]);
        sess.receive().unwrap();
        sess.publish(&[0]).unwrap();
        assert!(sess.session_mut().publish_queue.pop_front().is_some());
        sess.drive().unwrap();
        std::thread::sleep(Duration::from_millis(1));
    }

    // assert drive does not report liveness error
    while sess.drive().unwrap() == DriveOutcome::Active {}
    assert!(sess.drive().is_ok());

    // drive for longer than interval
    let end = SystemTime::now() + Duration::from_millis(100);
    while SystemTime::now() < end {
        sess.drive().ok();
        std::thread::sleep(Duration::from_millis(1));
    }

    // assert drive reports liveness error
    assert!(sess.drive().is_err());
}