zookeeper/
listeners.rs

1use std::collections::HashMap;
2use std::sync::{Arc, Mutex};
3use snowflake::ProcessUniqueId;
4
5/// A unique identifier returned by `ZooKeeper::add_listener`.
6///
7/// It can be used to remove a listener with `ZooKeeper::remove_listener`.
8#[derive(Debug,Clone,Copy,Eq,PartialEq,Hash)]
9pub struct Subscription(ProcessUniqueId);
10
11impl Subscription {
12    fn new() -> Subscription {
13        Subscription(ProcessUniqueId::new())
14    }
15}
16
17type ListenerMap<T> = HashMap<Subscription, Box<dyn Fn(T) + Send + 'static>>;
18
19#[derive(Clone)]
20pub struct ListenerSet<T>
21    where T: Send
22{
23    listeners: Arc<Mutex<ListenerMap<T>>>,
24}
25
26impl<T> ListenerSet<T> where T: Send + Clone
27{
28    pub fn new() -> Self {
29        ListenerSet { listeners: Arc::new(Mutex::new(ListenerMap::new())) }
30    }
31
32    pub fn subscribe<Listener: Fn(T) + Send + 'static>(&self, listener: Listener) -> Subscription {
33        let mut acquired_listeners = self.listeners.lock().unwrap();
34
35        let subscription = Subscription::new();
36        acquired_listeners.insert(subscription, Box::new(listener));
37
38        subscription
39    }
40
41    pub fn unsubscribe(&self, sub: Subscription) {
42        let mut acquired_listeners = self.listeners.lock().unwrap();
43
44        // channel will be close here automatically
45        acquired_listeners.remove(&sub);
46    }
47
48    pub fn notify(&self, payload: &T) {
49        let listeners = self.listeners.lock().unwrap();
50
51        for listener in listeners.values() {
52            listener(payload.clone())
53        }
54    }
55
56    #[allow(dead_code)]
57    pub fn len(&self) -> usize {
58        self.listeners.lock().unwrap().len()
59    }
60}
61
62#[cfg(test)]
63mod tests {
64    use super::{ListenerSet};
65    use std::sync::mpsc;
66
67    #[test]
68    fn test_new_listener_set() {
69        let ls = ListenerSet::<()>::new();
70        assert_eq!(ls.len(), 0);
71    }
72
73    #[test]
74    fn test_new_listener_for_chan() {
75        let ls = ListenerSet::<bool>::new();
76
77        ls.subscribe(|_e| {});
78        assert_eq!(ls.len(), 1);
79    }
80
81    #[test]
82    fn test_add_listener_to_set() {
83        let (tx, rx) = mpsc::channel();
84        let ls = ListenerSet::<bool>::new();
85
86        ls.subscribe(move |e| tx.send(e).unwrap());
87        assert_eq!(ls.len(), 1);
88
89        ls.notify(&true);
90        assert_eq!(rx.recv().is_ok(), true);
91    }
92
93    #[test]
94    fn test_remove_listener_from_set() {
95        let (tx, rx) = mpsc::channel();
96        let ls = ListenerSet::<bool>::new();
97
98        let sub = ls.subscribe(move |e| tx.send(e).unwrap());
99        ls.unsubscribe(sub);
100        assert_eq!(ls.len(), 0);
101
102        ls.notify(&true);
103        assert_eq!(rx.recv().is_err(), true);
104    }
105}
zookeeper/listeners.rs

zookeeper/
listeners.rs
