cspcl 0.5.0

Rust bindings for the cspcl library
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

#![cfg(feature = "async-tokio")]

use std::sync::{Mutex, MutexGuard};

use cspcl::async_api::AsyncCspcl;
use cspcl::{Cspcl, CspclConfig, Error, Interface, InterfaceName};

static TEST_GUARD: Mutex<()> = Mutex::new(());

fn test_lock() -> MutexGuard<'static, ()> {
    TEST_GUARD
        .lock()
        .unwrap_or_else(|poisoned| poisoned.into_inner())
}

fn test_instance() -> Cspcl {
    Cspcl::from_config(
        CspclConfig::new(7).with_interface(Interface::Loopback(InterfaceName::new("loopback"))),
    )
    .expect("failed to initialize test cspcl instance")
}

fn assert_timeout(err: Error) {
    assert_eq!(
        err.code(),
        cspcl::cspcl_sys::cspcl_error_t_CSPCL_ERR_TIMEOUT
    );
}

fn assert_not_initialized(err: Error) {
    assert_eq!(
        err.code(),
        cspcl::cspcl_sys::cspcl_error_t_CSPCL_ERR_NOT_INITIALIZED
    );
}

fn assert_invalid_param(err: Error) {
    assert_eq!(
        err.code(),
        cspcl::cspcl_sys::cspcl_error_t_CSPCL_ERR_INVALID_PARAM
    );
}

#[tokio::test(flavor = "multi_thread")]
async fn async_wrappers_can_be_constructed_from_sync_runtime() {
    let _guard = test_lock();
    let cspcl = test_instance();
    let async_cspcl = AsyncCspcl::from_sync(cspcl.clone());
    let (_sender, _receiver) = async_cspcl.split();

    assert!(async_cspcl.is_initialized());
    assert_eq!(async_cspcl.local_addr(), cspcl.local_addr());
    assert_eq!(async_cspcl.connection_stats(), cspcl.connection_stats());
}

#[tokio::test(flavor = "multi_thread")]
async fn async_receiver_times_out_without_pending_bundle() {
    let _guard = test_lock();
    let cspcl = test_instance();
    let async_receiver = AsyncCspcl::from_sync(cspcl).receiver();

    assert_timeout(async_receiver.recv_bundle(5).await.unwrap_err());
}

#[tokio::test(flavor = "multi_thread")]
async fn async_receiver_into_times_out_without_pending_bundle() {
    let _guard = test_lock();
    let cspcl = test_instance();
    let async_receiver = AsyncCspcl::from_sync(cspcl).receiver();
    let mut buffer = [0_u8; 64];

    assert_timeout(
        async_receiver
            .recv_bundle_into(&mut buffer, 5)
            .await
            .unwrap_err(),
    );
}

#[tokio::test(flavor = "multi_thread")]
async fn async_receive_observes_sync_shutdown() {
    let _guard = test_lock();
    let cspcl = test_instance();
    let async_receiver = AsyncCspcl::from_sync(cspcl.clone()).receiver();
    let mut buffer = [0_u8; 64];

    cspcl.shutdown().unwrap();

    assert_not_initialized(async_receiver.recv_bundle(5).await.unwrap_err());
    assert_not_initialized(
        async_receiver
            .recv_bundle_into(&mut buffer, 5)
            .await
            .unwrap_err(),
    );
}

#[tokio::test(flavor = "multi_thread")]
async fn async_sender_can_send_and_report_stats() {
    let _guard = test_lock();
    let cspcl = test_instance();
    let async_cspcl = AsyncCspcl::from_sync(cspcl.clone());
    let sender = async_cspcl.sender();
    let sender_clone = sender.clone();

    sender.send_bundle(&[1, 2, 3], 42, 10).await.unwrap();
    sender_clone.send_bundle(&[4, 5, 6], 42, 10).await.unwrap();

    let stats = async_cspcl.connection_stats();
    assert!(stats.misses >= 1);
    assert!(stats.hits >= 1);
    assert_eq!(sender.connection_stats(), stats);
}

#[tokio::test(flavor = "multi_thread")]
async fn async_sender_rejects_invalid_input_and_shutdown() {
    let _guard = test_lock();
    let cspcl = test_instance();
    let async_cspcl = AsyncCspcl::from_sync(cspcl.clone());
    let sender = async_cspcl.sender();

    assert_invalid_param(sender.send_bundle(&[], 42, 10).await.unwrap_err());

    async_cspcl.shutdown().await.unwrap();

    assert_not_initialized(sender.send_bundle(&[1, 2, 3], 42, 10).await.unwrap_err());
    assert!(!cspcl.is_initialized());
}