event_stream/
lib.rs

1#[macro_use]
2extern crate crossbeam_channel;
3
4use crossbeam_channel::{unbounded, Receiver, Sender};
5
6use std::{sync::Arc, thread};
7
8pub trait EventListener<T>
9where
10    T: Sync + Send,
11{
12    fn dispatch(&self, _event: &T) {}
13}
14
15pub struct EventDispatcher<T> {
16    tx: Sender<T>,
17    rx: Receiver<T>,
18    listeners: Vec<Arc<dyn EventListener<T> + Send + Sync>>,
19}
20
21pub struct EventDispatcherBuilder<T> {
22    tx: Sender<T>,
23    rx: Receiver<T>,
24    listeners: Vec<Arc<dyn EventListener<T> + Send + Sync>>,
25}
26
27impl<T> EventDispatcherBuilder<T>
28where
29    T: Sync + Send + Sized,
30{
31    pub fn new() -> Self {
32        let (tx, rx) = unbounded::<T>();
33        EventDispatcherBuilder {
34            tx,
35            rx,
36            listeners: vec![],
37        }
38    }
39
40    pub fn add_listener(mut self, listener: Arc<dyn EventListener<T> + Send + Sync>) -> Self {
41        self.listeners.push(listener);
42        self
43    }
44
45    pub fn build(self) -> EventDispatcher<T> {
46        EventDispatcher {
47            tx: self.tx.clone(),
48            rx: self.rx.clone(),
49            listeners: self.listeners,
50        }
51    }
52}
53
54impl<T> EventDispatcher<T>
55where
56    T: Sized + Send + Sync + 'static,
57{
58    pub fn emit(&self, event: T) {
59        let _ = self.tx.send(event).expect("cannot send via channel");
60    }
61
62    pub fn start(&self) {
63        let rx_thread = self.rx.clone();
64        let listeners = self.listeners.clone();
65        let mut tried = 0;
66
67        thread::spawn(move || loop {
68            let rx_thread_c = rx_thread.clone();
69            let listeners_c = listeners.clone();
70            let h = thread::spawn(move || loop {
71                select! {
72                    recv(rx_thread_c) -> msg => {
73                        for listener in listeners_c.iter() {
74                            if let Ok(event) = &msg {
75                                listener.dispatch(&event);
76                            }
77                        }
78                    }
79                }
80            });
81
82            let r = h.join();
83            match r {
84                Ok(r) => eprintln!("event dispatcher thread ended! {:?}", r),
85                Err(e) => {
86                    eprintln!("event dispatcher thread died! {:?}", e);
87                }
88            }
89
90            eprintln!("Starting new event dispatcher thread again...");
91            tried = tried + 1;
92            if tried >= 100 {
93                eprintln!(
94                    "event dispatcher thread cannot be started after {} attempts!",
95                    tried
96                );
97                break;
98            }
99        });
100    }
101
102    pub fn get_channel(&self) -> &Sender<T> {
103        &self.tx
104    }
105}
event_stream/lib.rs

event_stream/
lib.rs
