acts 0.17.2

a fast, lightweight, extensiable workflow engine
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
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146

use std::sync::{Arc, Mutex};
use tokio::sync::Notify;

#[derive(Clone)]
pub struct Signal<T> {
    data: Arc<Mutex<T>>,
    sig: Arc<Notify>,
    is_closed: Arc<Mutex<bool>>,
}

impl<T: Clone> Signal<T> {
    pub fn new(v: T) -> Self {
        let sig = Arc::new(Notify::new());
        let data = Arc::new(Mutex::new(v));
        Self {
            sig,
            data,
            is_closed: Arc::new(Mutex::new(false)),
        }
    }

    pub fn send(&self, v: T) {
        if *self.is_closed.lock().unwrap() {
            return;
        }
        *self.data.lock().unwrap() = v;
        self.close();
    }

    pub fn close(&self) {
        *self.is_closed.lock().unwrap() = true;
        self.sig.notify_one();
    }
    pub fn data(&self) -> T {
        let data = self.data.lock().unwrap();
        data.clone()
    }

    pub fn update<F: Fn(&mut T)>(&self, f: F) {
        if *self.is_closed.lock().unwrap() {
            return;
        }
        let mut data = self.data.lock().unwrap();
        f(&mut data);
    }

    pub fn double(&self) -> (Self, Self) {
        (self.clone(), self.clone())
    }

    pub fn triple(&self) -> (Self, Self, Self) {
        (self.clone(), self.clone(), self.clone())
    }

    pub async fn recv(&self) -> T {
        self.sig.notified().await;
        let v = self.data.lock().unwrap();
        v.clone()
    }

    pub async fn timeout(&self, millis: u64) -> T {
        tokio::time::sleep(tokio::time::Duration::from_millis(millis)).await;
        let v = self.data.lock().unwrap();
        v.clone()
    }
}

#[cfg(test)]
mod tests {
    use crate::Signal;

    #[test]
    fn engine_signal_new() {
        let s = Signal::new(5);
        assert_eq!(s.data(), 5);

        let s = Signal::new("abc");
        assert_eq!(s.data(), "abc");
    }

    #[tokio::test]
    async fn engine_signal_send() {
        let s = Signal::new(0);
        let s2 = s.clone();
        tokio::spawn(async move {
            s.send(10);
        });
        let ret = s2.recv().await;
        assert_eq!(ret, 10);
    }

    #[tokio::test]
    async fn engine_signal_close() {
        let s = Signal::new(0);
        let s2 = s.clone();
        tokio::spawn(async move {
            s.close();
        });
        let ret = s2.recv().await;
        assert_eq!(ret, 0);
    }

    #[tokio::test]
    async fn engine_signal_update() {
        let s = Signal::new(0);
        let s2 = s.clone();
        tokio::spawn(async move {
            s.update(|data| *data = 100);
            s.close();
        });
        let ret = s2.recv().await;
        assert_eq!(ret, 100);
    }

    #[tokio::test]
    async fn engine_signal_timeout() {
        let s = Signal::new(0);
        let s2 = s.clone();
        tokio::spawn(async move {
            s.update(|data| *data = 100);
        });
        let ret = s2.timeout(10).await;
        assert_eq!(ret, 100);
    }

    #[tokio::test]
    async fn engine_signal_double() {
        let (s1, s2) = Signal::new(0).double();
        tokio::spawn(async move {
            s1.send(10);
        });
        let ret = s2.recv().await;
        assert_eq!(ret, 10);
    }

    #[tokio::test]
    async fn engine_signal_triple() {
        let (s1, s2, s3) = Signal::new(0).triple();
        tokio::spawn(async move {
            s1.update(|data| *data = 100);
            s2.close();
        });
        let ret = s3.recv().await;
        assert_eq!(ret, 100);
    }
}