shared/
events.rs

1use futures_util::StreamExt;
2use parking_lot::RwLock;
3use std::{collections::HashMap, pin::Pin, sync::Arc};
4
5type Listener<Event> = dyn Fn(
6        crate::State,
7        Arc<Event>,
8    ) -> Pin<Box<dyn Future<Output = Result<(), anyhow::Error>> + Send + 'static>>
9    + Send
10    + Sync;
11
12#[derive(Clone)]
13pub struct EventHandlerHandle {
14    listeners_ref: Arc<dyn DisconnectEventHandler + Send + Sync>,
15    id: uuid::Uuid,
16}
17
18impl EventHandlerHandle {
19    #[inline]
20    pub fn disconnect(self) {
21        self.listeners_ref.disconnect(self.id);
22    }
23}
24
25pub(crate) trait DisconnectEventHandler {
26    fn disconnect(&self, id: uuid::Uuid);
27}
28
29impl<Event> DisconnectEventHandler for RwLock<HashMap<uuid::Uuid, Box<Listener<Event>>>> {
30    #[inline]
31    fn disconnect(&self, id: uuid::Uuid) {
32        self.write().remove(&id);
33    }
34}
35
36pub struct EventEmitter<Event: 'static + Send + Sync> {
37    listeners: Arc<RwLock<HashMap<uuid::Uuid, Box<Listener<Event>>>>>,
38    event_channel: tokio::sync::mpsc::Sender<(crate::State, Event)>,
39    task: tokio::task::JoinHandle<()>,
40}
41
42impl<Event: 'static + Send + Sync> Default for EventEmitter<Event> {
43    fn default() -> Self {
44        let listeners = Arc::new(RwLock::new(HashMap::new()));
45        let (event_channel_sender, mut event_channel_receiver) = tokio::sync::mpsc::channel(64);
46
47        Self {
48            listeners: listeners.clone(),
49            event_channel: event_channel_sender,
50            task: tokio::spawn(async move {
51                let semaphore = Arc::new(tokio::sync::Semaphore::new(8));
52
53                while let Some((state, event)) = event_channel_receiver.recv().await {
54                    tracing::debug!("emitting event {:?}", std::any::type_name::<Event>());
55
56                    let listeners = listeners.clone();
57                    let permit = match semaphore.clone().acquire_owned().await {
58                        Ok(permit) => permit,
59                        Err(_) => {
60                            tracing::error!("semaphore closed, shutting down event emitter");
61                            break;
62                        }
63                    };
64
65                    tokio::spawn(async move {
66                        let event = Arc::new(event);
67                        let listeners = listeners
68                            .read()
69                            .values()
70                            .map(|listener| listener(state.clone(), event.clone()))
71                            .collect::<Vec<_>>();
72
73                        let mut result_stream =
74                            futures_util::stream::iter(listeners).buffer_unordered(8);
75
76                        while let Some(result) = result_stream.next().await {
77                            if let Err(err) = result {
78                                tracing::error!(
79                                    "event listener error for {:?}: {:?}",
80                                    std::any::type_name::<Event>(),
81                                    err
82                                );
83                            }
84                        }
85
86                        drop(permit);
87                    });
88                }
89            }),
90        }
91    }
92}
93
94impl<Event: 'static + Send + Sync> EventEmitter<Event> {
95    pub fn register_event_handler<
96        F: Fn(crate::State, Arc<Event>) -> Fut + Send + Sync + 'static,
97        Fut: Future<Output = Result<(), anyhow::Error>> + Send + 'static,
98    >(
99        &self,
100        listener: F,
101    ) -> EventHandlerHandle {
102        let id = uuid::Uuid::new_v4();
103        let listener_box = Box::new(move |state: crate::State, event: Arc<Event>| {
104            Box::pin(listener(state, event))
105                as Pin<Box<dyn Future<Output = Result<(), anyhow::Error>> + Send + 'static>>
106        }) as Box<Listener<Event>>;
107
108        self.listeners.write().insert(id, listener_box);
109
110        EventHandlerHandle {
111            listeners_ref: self.listeners.clone(),
112            id,
113        }
114    }
115
116    #[inline]
117    pub fn emit(&self, state: crate::State, event: Event) {
118        let _ = self.event_channel.try_send((state, event));
119    }
120}
121
122impl<Event: 'static + Send + Sync> Drop for EventEmitter<Event> {
123    #[inline]
124    fn drop(&mut self) {
125        self.task.abort();
126    }
127}
shared/events.rs

shared/
events.rs
