zookeeper_async/
listeners.rs

1use snowflake::ProcessUniqueId;
2use std::collections::HashMap;
3use std::sync::{Arc, Mutex};
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>
21where
22    T: Send,
23{
24    listeners: Arc<Mutex<ListenerMap<T>>>,
25}
26
27impl<T> ListenerSet<T>
28where
29    T: Send + Clone,
30{
31    pub fn new() -> Self {
32        ListenerSet {
33            listeners: Arc::new(Mutex::new(ListenerMap::new())),
34        }
35    }
36
37    pub fn subscribe<Listener: Fn(T) + Send + 'static>(&self, listener: Listener) -> Subscription {
38        let mut acquired_listeners = self.listeners.lock().unwrap();
39
40        let subscription = Subscription::new();
41        acquired_listeners.insert(subscription, Box::new(listener));
42
43        subscription
44    }
45
46    pub fn unsubscribe(&self, sub: Subscription) {
47        let mut acquired_listeners = self.listeners.lock().unwrap();
48
49        // channel will be close here automatically
50        acquired_listeners.remove(&sub);
51    }
52
53    pub fn notify(&self, payload: &T) {
54        let listeners = self.listeners.lock().unwrap();
55
56        for listener in listeners.values() {
57            listener(payload.clone())
58        }
59    }
60
61    #[allow(dead_code)]
62    pub fn len(&self) -> usize {
63        self.listeners.lock().unwrap().len()
64    }
65}
66
67#[cfg(test)]
68mod tests {
69    use super::ListenerSet;
70    use std::sync::mpsc;
71
72    #[test]
73    fn test_new_listener_set() {
74        let ls = ListenerSet::<()>::new();
75        assert_eq!(ls.len(), 0);
76    }
77
78    #[test]
79    fn test_new_listener_for_chan() {
80        let ls = ListenerSet::<bool>::new();
81
82        ls.subscribe(|_e| {});
83        assert_eq!(ls.len(), 1);
84    }
85
86    #[test]
87    fn test_add_listener_to_set() {
88        let (tx, rx) = mpsc::channel();
89        let ls = ListenerSet::<bool>::new();
90
91        ls.subscribe(move |e| tx.send(e).unwrap());
92        assert_eq!(ls.len(), 1);
93
94        ls.notify(&true);
95        assert!(rx.recv().is_ok());
96    }
97
98    #[test]
99    fn test_remove_listener_from_set() {
100        let (tx, rx) = mpsc::channel();
101        let ls = ListenerSet::<bool>::new();
102
103        let sub = ls.subscribe(move |e| tx.send(e).unwrap());
104        ls.unsubscribe(sub);
105        assert_eq!(ls.len(), 0);
106
107        ls.notify(&true);
108        assert!(rx.recv().is_err());
109    }
110}
zookeeper_async/listeners.rs

zookeeper_async/
listeners.rs
