agner-actors 0.4.1

An actor toolkit inspired by Erlang/OTP (actors)
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

use agner_utils::spsc;

#[derive(Debug)]
pub struct PipeRx<T>(spsc::Receiver<T>);

#[derive(Debug)]
pub struct PipeTx<T>(spsc::Sender<T>, bool);

impl<T> PipeTx<T> {
    pub fn blocking(self) -> Self {
        Self(self.0, true)
    }
    // pub fn rejecting(self) -> Self {
    //     Self(self.0, false)
    // }
}

pub fn new<M>(max_len: usize) -> (PipeTx<M>, PipeRx<M>) {
    let (tx, rx) = spsc::channel(max_len);
    (PipeTx(tx, false), PipeRx(rx))
}

impl<T> PipeTx<T>
where
    T: Unpin,
{
    pub async fn send(&mut self, message: T) -> Result<(), T> {
        self.0.send(message, self.1).await
    }

    pub async fn len(&self) -> (usize, usize)
    where
        T: Unpin,
    {
        self.0.len().await
    }
}

impl<T> PipeRx<T>
where
    T: Unpin,
{
    pub async fn recv(&mut self) -> T {
        self.0.recv(true).await.unwrap()
    }

    pub async fn len(&self) -> (usize, usize)
    where
        T: Unpin,
    {
        self.0.len().await
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[tokio::test]
    async fn basic_test() {
        const LEN: usize = 30;
        let (mut inbox_w, mut inbox_r) = new::<usize>(LEN);

        for i in 1..1_000_000 {
            if let Err(rejected) = inbox_w.send(i).await {
                assert_eq!(rejected, i);

                let popped = inbox_r.recv().await;
                assert_eq!(popped + LEN, rejected);
                assert!(inbox_w.send(rejected).await.is_ok());
            }
        }
    }

    #[tokio::test]
    async fn futures_mpsc_test() {
        use futures::StreamExt;

        const LEN: usize = 30;
        let (mut inbox_w, mut inbox_r) = futures::channel::mpsc::channel::<usize>(LEN - 1);

        for i in 1..1_000_000 {
            if let Err(rejected) = inbox_w.try_send(i) {
                let rejected = rejected.into_inner();
                assert_eq!(rejected, i);

                let popped = inbox_r.next().await.expect("channel closed?");
                assert_eq!(popped + LEN, rejected);
                assert!(inbox_w.try_send(rejected).is_ok());
            }
        }
    }

    #[tokio::test]
    async fn tokio_mpsc_test() {
        use tokio::sync::mpsc::error::TrySendError;

        const LEN: usize = 30;
        let (inbox_w, mut inbox_r) = tokio::sync::mpsc::channel::<usize>(LEN);

        for i in 1..1_000_000 {
            if let Err(rejected) = inbox_w.try_send(i) {
                let rejected = match rejected {
                    TrySendError::Full(rejected) => rejected,
                    _ => panic!("unexpected channel error"),
                };
                assert_eq!(rejected, i);

                let popped = inbox_r.recv().await.expect("channel closed?");
                assert_eq!(popped + LEN, rejected);
                assert!(inbox_w.try_send(rejected).is_ok());
            }
        }
    }
}