unsync 0.1.2

Unsynchronized synchronization primitives for async Rust.
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

use tokio::task;
use unsync::broadcast;

#[cfg(not(miri))]
const SIZE: u32 = 100_000;

#[cfg(miri)]
const SIZE: u32 = 10;

#[tokio::test]
async fn test_broadcast() -> Result<(), Box<dyn std::error::Error>> {
    let local = task::LocalSet::new();

    let mut tx = broadcast::channel(2);

    let (receivers, b) = local
        .run_until(async move {
            let mut receivers = Vec::new();

            for _ in 0..16 {
                let mut rx = tx.subscribe();

                let a = task::spawn_local(async move {
                    let mut out = Vec::new();

                    while let Some(value) = rx.recv().await {
                        out.push(value);

                        if value % 3 == 0 {
                            task::yield_now().await;
                        }
                    }

                    out
                });

                receivers.push(a);
            }

            let b = task::spawn_local(async move {
                for n in 0..SIZE {
                    let _ = tx.send(n).await;

                    if n % 5 == 0 {
                        task::yield_now().await;
                    }
                }
            });

            let mut received = Vec::new();

            for receiver in receivers {
                received.push(receiver.await.unwrap());
            }

            (received, b.await)
        })
        .await;

    b?;

    let expected = (0..SIZE).collect::<Vec<_>>();

    for actual in receivers {
        assert_eq!(actual, expected);
    }

    Ok(())
}

#[tokio::test]
async fn test_broadcast_receiver_drop() {
    let mut tx = broadcast::channel(1);
    let mut sub1 = tx.subscribe();
    let mut sub2 = tx.subscribe();

    let (result, s1, s2) = tokio::join!(tx.send(1), sub1.recv(), sub2.recv());
    drop(sub2);
    assert_eq!(result, 2);
    assert_eq!(s1, Some(1));
    assert_eq!(s2, Some(1));

    let (result, s1) = tokio::join!(tx.send(2), sub1.recv());
    assert_eq!(result, 1);
    assert_eq!(s1, Some(2));
    drop(sub1);

    let (send,) = tokio::join!(tx.send(2));
    assert_eq!(send, 0);
}