dnet_tests/
lib.rs

1//! Common tests for `dnet` transport implementations.
2
3#![warn(missing_docs)]
4
5use std::fmt::Debug;
6use std::time::Duration;
7
8use dnet_base::{Messages, Receive, Transport};
9
10use futures::{stream, SinkExt, StreamExt};
11
12pub use dportable::test::dtest;
13
14pub use dportable::test::dtest_configure;
15
16/// Test transport by sending strings and integers between two connected instances.
17pub async fn test_transport<L, R, E1, E2>(mut left: L, mut right: R)
18where
19    L: Transport<u32, String, E1> + Unpin,
20    R: Transport<String, u32, E2> + Unpin,
21    E1: Debug,
22    E2: Debug,
23{
24    init_logging(&mut left, &mut right);
25
26    left.send("Hello World!".to_string()).await.unwrap();
27    left.send("Hello World again!".to_string()).await.unwrap();
28    right.send(128).await.unwrap();
29    right.send(1).await.unwrap();
30
31    assert_eq!(right.receive().await.unwrap(), "Hello World!");
32    assert_eq!(right.receive().await.unwrap(), "Hello World again!");
33    assert_eq!(left.receive().await.unwrap(), 128);
34    assert_eq!(left.receive().await.unwrap(), 1);
35}
36
37/// Test transport by sending units (`()`) between two connected instances.
38///
39/// When used with `BincodeCodec` it can verify sending message of length 0 (in bytes) works.
40pub async fn test_unit_message<L, R, E1, E2>(mut left: L, mut right: R)
41where
42    L: Transport<(), (), E1> + Unpin,
43    R: Transport<(), (), E2> + Unpin,
44    E1: Debug,
45    E2: Debug,
46{
47    init_logging(&mut left, &mut right);
48
49    left.send(()).await.unwrap();
50    left.send(()).await.unwrap();
51    right.send(()).await.unwrap();
52    right.send(()).await.unwrap();
53
54    right.receive().await.unwrap();
55    right.receive().await.unwrap();
56    left.receive().await.unwrap();
57    left.receive().await.unwrap();
58}
59
60/// Test transport by collecting received messages while treating transport as a stream.
61///
62/// It verifies if transport closing (or dropping) on one side is
63/// communicated to the other side - without it, stream would never complete.
64pub async fn test_stream<L, R, E1, E2>(left: L, right: R)
65where
66    L: Transport<(), u32, E1> + Unpin,
67    R: Transport<u32, (), E2> + Unpin,
68    E1: Debug,
69    E2: Debug,
70{
71    test_stream_with_sleep_before_drop(left, right, Duration::ZERO).await
72}
73
74/// Same as [test_stream] except we [sleep](dportable::time::sleep) for specified
75/// duration before dropping the sending transport.
76///
77/// Used by unreliable QUIC transport - there is no mechanism for the underlying
78/// unreliable transport to wait for messages to be flushed.
79pub async fn test_stream_with_sleep_before_drop<L, R, E1, E2>(
80    mut left: L,
81    mut right: R,
82    duration: Duration,
83) where
84    L: Transport<(), u32, E1> + Unpin,
85    R: Transport<u32, (), E2> + Unpin,
86    E1: Debug,
87    E2: Debug,
88{
89    init_logging(&mut left, &mut right);
90
91    left.send_all(&mut stream::iter(vec![1, 2, 3].into_iter().map(Ok)))
92        .await
93        .unwrap();
94    if !duration.is_zero() {
95        dportable::time::sleep(duration).await;
96    }
97    drop(left);
98
99    assert_eq!(right.messages().collect::<Vec<u32>>().await, vec![1, 2, 3]);
100}
101
102/// Init tracing subscriber and enable logging for given transports.
103pub fn init_logging<T1, T2, T1I, T1O, T2I, T2O, E1, E2>(left: &mut T1, right: &mut T2)
104where
105    T1: Transport<T1I, T1O, E1>,
106    T2: Transport<T2I, T2O, E2>,
107{
108    #[cfg(not(feature = "logging"))]
109    {
110        let _ = (left, right);
111    }
112
113    #[cfg(feature = "logging")]
114    {
115        init_subscriber();
116
117        left.enable_logging();
118        left.set_logging_name("left");
119
120        right.enable_logging();
121        right.set_logging_name("right");
122    }
123}
124
125/// Init tracing subscriber.
126#[cfg(feature = "logging")]
127pub fn init_subscriber() {
128    use std::sync::Once;
129
130    static INIT: Once = Once::new();
131    INIT.call_once(|| {
132        #[cfg(target_arch = "wasm32")]
133        wasm_tracing::set_as_global_default();
134
135        #[cfg(not(target_arch = "wasm32"))]
136        tracing_subscriber::fmt().with_env_filter("trace").init();
137    });
138}
dnet_tests/lib.rs

dnet_tests/
lib.rs
