ntex 0.3.1

Framework for composable network services
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

use std::{cell::RefCell, collections::BTreeMap, rc::Rc, time::Duration, time::Instant};

use futures::future::{ready, FutureExt};

use crate::framed::State;
use crate::rt::time::delay_for;
use crate::HashSet;

pub struct Timer(Rc<RefCell<Inner>>);

struct Inner {
    resolution: Duration,
    current: Option<Instant>,
    notifications: BTreeMap<Instant, HashSet<State>>,
}

impl Inner {
    fn new(resolution: Duration) -> Self {
        Inner {
            resolution,
            current: None,
            notifications: BTreeMap::default(),
        }
    }

    fn unregister(&mut self, expire: Instant, state: &State) {
        if let Some(ref mut states) = self.notifications.get_mut(&expire) {
            states.remove(state);
            if states.is_empty() {
                self.notifications.remove(&expire);
            }
        }
    }
}

impl Clone for Timer {
    fn clone(&self) -> Self {
        Timer(self.0.clone())
    }
}

impl Default for Timer {
    fn default() -> Self {
        Timer::with(Duration::from_secs(1))
    }
}

impl Timer {
    pub fn with(resolution: Duration) -> Timer {
        Timer(Rc::new(RefCell::new(Inner::new(resolution))))
    }

    pub fn register(&self, expire: Instant, previous: Instant, state: &State) {
        {
            let mut inner = self.0.borrow_mut();

            inner.unregister(previous, state);
            inner
                .notifications
                .entry(expire)
                .or_insert_with(HashSet::default)
                .insert(state.clone());
        }

        let _ = self.now();
    }

    pub fn unregister(&self, expire: Instant, state: &State) {
        self.0.borrow_mut().unregister(expire, state);
    }

    /// Get current time. This function has to be called from
    /// future's poll method, otherwise it panics.
    pub fn now(&self) -> Instant {
        let cur = self.0.borrow().current;
        if let Some(cur) = cur {
            cur
        } else {
            let now = Instant::now();
            let inner = self.0.clone();
            let interval = {
                let mut b = inner.borrow_mut();
                b.current = Some(now);
                b.resolution
            };

            crate::rt::spawn(delay_for(interval).then(move |_| {
                let empty = {
                    let mut i = inner.borrow_mut();
                    let now = i.current.take().unwrap_or_else(Instant::now);

                    // notify io dispatcher
                    while let Some(key) = i.notifications.keys().next() {
                        let key = *key;
                        if key <= now {
                            for st in i.notifications.remove(&key).unwrap() {
                                st.keepalive_timeout();
                            }
                        } else {
                            break;
                        }
                    }
                    i.notifications.is_empty()
                };

                // extra tick
                if !empty {
                    let _ = Timer(inner).now();
                }

                ready(())
            }));

            now
        }
    }
}