mco 0.1.48

Rust Coroutine Library like go
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

use crate::coroutine::sleep;
use crate::std::errors::Result;
use crate::std::sync::channel::Receiver;
use crate::std::sync::Mutex;
use crate::std::time::time::Time;
use std::ops::Deref;
use std::sync::Arc;
use std::time::Duration;

/// A Ticker holds a channel that delivers ``ticks'' of a clock
/// at intervals.
/// for example:
/// ```
///         use mco::coroutine::sleep;
///         use mco::std::time::tick::Ticker;
///         use std::sync::Arc;
///         use std::time::Duration;
///         use mco::co;
///
///         let mut t = Arc::new(Ticker::new(Duration::from_secs(1)));
///         let tclone = t.clone();
///         co!(move ||{
///              for x in tclone.as_ref() {
///                println!("tick {}", x);
///             }
///         });
///         sleep(Duration::from_secs(3));
///         t.stop();
///
/// ```
pub struct Ticker {
    pub d: Arc<Mutex<Duration>>,
    pub recv: Receiver<Time>,
}

impl Ticker {
    pub fn new_arc(d: Duration) -> Arc<Self> {
        Arc::new(Self::new(d))
    }

    pub fn new(d: Duration) -> Self {
        let d = Arc::new(Mutex::new(d));
        let (s, r) = chan!();
        let ticker = Self { d: d, recv: r };
        let d = ticker.d.clone();
        let tick = move || loop {
            match d.lock() {
                Ok(d) => {
                    if d.is_zero() {
                        break;
                    }
                    sleep(d.deref().clone());
                    let _ = s.send(Time::now());
                }
                Err(_) => {
                    break;
                }
            }
        };
        co!(tick);
        ticker
    }

    /// Stop turns off a ticker. After Stop, no more ticks will be sent.
    /// Stop does not close the channel, to prevent a concurrent goroutine
    /// reading from the channel from seeing an erroneous "tick".
    pub fn stop(&self) -> Result<()> {
        match self.d.lock() {
            Ok(mut d) => {
                *d = Duration::from_secs(0);
                Ok(())
            }
            Err(e) => Err(err!("lock fail: {}", e)),
        }
    }

    /// Reset stops a ticker and resets its period to the specified duration.
    /// The next tick will arrive after the new period elapses.
    pub fn reset(&self, d: Duration) -> Result<()> {
        match self.d.lock() {
            Ok(mut dur) => {
                *dur = d;
                Ok(())
            }
            Err(e) => Err(err!("lock fail: {}", e)),
        }
    }
}

impl Iterator for Ticker {
    type Item = Time;

    fn next(&mut self) -> Option<Self::Item> {
        match self.recv.recv() {
            Ok(v) => Some(v),
            Err(_) => None,
        }
    }
}

impl Iterator for &Ticker {
    type Item = Time;

    fn next(&mut self) -> Option<Self::Item> {
        match self.recv.recv() {
            Ok(v) => Some(v),
            Err(_) => None,
        }
    }
}

#[cfg(test)]
mod test {
    use crate::sleep::sleep;
    use crate::std::time::tick::Ticker;
    use std::sync::Arc;
    use std::time::Duration;

    //test --package mco --lib std::time::tick::test::test_tick -- --exact --nocapture
    #[test]
    fn test_tick() {
        let mut t = Arc::new(Ticker::new(Duration::from_secs(1)));
        let tclone = t.clone();
        co!(move || {
            for x in tclone.as_ref() {
                println!("tick {}", x);
            }
        });
        sleep(Duration::from_secs(3));
        t.stop();
    }
}