async-listen 0.1.2

Various helpers for writing production-ready servers in rust using async-std.
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

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

use rand::{Rng, thread_rng};

use async_std::stream::{from_iter, Stream, StreamExt};
use async_std::task;

use async_listen::{ListenExt, backpressure};

fn collect<S: Stream + Unpin>(mut stream: S) -> Vec<S::Item> {
    task::block_on(async {
        let mut result = Vec::new();
        while let Some(item) = stream.next().await {
            result.push(item);
        }
        result
    })
}

fn random_delay() -> Duration {
    Duration::from_millis(thread_rng().gen_range(1, 200))
}

#[test]
fn test_backpressure() {
    let current = Arc::new(AtomicUsize::new(0));
    let top = Arc::new(AtomicUsize::new(0));
    let tasks = collect(
        from_iter(0..100)
        .backpressure(10)
        .map(|(token, _index)| {
            let current = current.clone();
            let top = top.clone();
            task::spawn(async move {
                let size = current.fetch_add(1, Ordering::Acquire) + 1;
                assert!(size <= 10);
                if size == 10 {
                    top.fetch_add(1, Ordering::Relaxed);
                }
                task::sleep(random_delay()).await;
                current.fetch_sub(1, Ordering::Acquire);
                drop(token);
            })
        }));
    for item in tasks {
        task::block_on(item);
    };
    let top = top.load(Ordering::SeqCst);
    println!("Top capacity {}", top);
    assert!(30 < top && top <= 91);
}

#[test]
fn test_apply_backpressure() {
    let current = Arc::new(AtomicUsize::new(0));
    let top = Arc::new(AtomicUsize::new(0));
    let (tx, rx) = backpressure::new(10);
    let tasks = collect(
        from_iter(0..100)
        .apply_backpressure(rx)
        .map(|_index| {
            let token = tx.token();
            let current = current.clone();
            let top = top.clone();
            task::spawn(async move {
                let size = current.fetch_add(1, Ordering::Acquire) + 1;
                assert!(size <= 10);
                if size == 10 {
                    top.fetch_add(1, Ordering::Relaxed);
                }
                task::sleep(random_delay()).await;
                current.fetch_sub(1, Ordering::Acquire);
                drop(token);
            })
        }));
    for item in tasks {
        task::block_on(item);
    };
    let top = top.load(Ordering::SeqCst);
    println!("Top capacity {}", top);
    assert!(30 < top && top <= 91);
}

#[test]
fn test_change_limit() {
    let current = Arc::new(AtomicUsize::new(0));
    let top = Arc::new(AtomicUsize::new(0));
    let (tx, rx) = backpressure::new(10);
    let tasks = collect(
        from_iter(0..100)
        .apply_backpressure(rx)
        .map(|index| {
            let tx = tx.clone();
            let token = tx.token();
            let current = current.clone();
            let top = top.clone();
            task::spawn(async move {
                let size = current.fetch_add(1, Ordering::Acquire) + 1;
                assert!(size <= 20);
                if size == 20 {
                    top.fetch_add(1, Ordering::Relaxed);
                }
                task::sleep(random_delay()).await;
                if index == 20 {
                    tx.set_limit(20);
                } else if index == 60 {
                    tx.set_limit(5);
                }
                current.fetch_sub(1, Ordering::Acquire);
                drop(token);
            })
        }));
    for item in tasks {
        task::block_on(item);
    };
    let top = top.load(Ordering::SeqCst);
    println!("Top capacity {}", top);
    assert!(5 < top && top <= 81);
}