text_document_frontend/
event_hub_client.rs

1// Generated by Qleany v1.7.3 from frontend_event_hub_client.tera
2use common::event::{Event, EventHub, Origin};
3use flume::Receiver;
4use std::collections::HashMap;
5use std::sync::atomic::{AtomicU64, Ordering};
6use std::sync::{Arc, Mutex};
7use std::thread;
8
9/// Event callback type
10pub type EventCallback = Box<dyn Fn(Event) + Send>;
11
12/// Internal storage: each subscriber entry pairs its id (for O(n) removal on
13/// token drop) with its callback. Vec preserves registration order so
14/// subscribers fire deterministically.
15type SubscriberList = Vec<(u64, EventCallback)>;
16
17/// EventHubClient - handles event passing from backend to UI
18/// Subscribe callbacks to specific event origins and start the event loop.
19#[derive(Clone)]
20pub struct EventHubClient {
21    subscribers: Arc<Mutex<HashMap<Origin, SubscriberList>>>,
22    receiver: Receiver<Event>,
23    next_subscriber_id: Arc<AtomicU64>,
24}
25
26impl EventHubClient {
27    /// Create a new event hub client
28    pub fn new(event_hub: &EventHub) -> Self {
29        EventHubClient {
30            subscribers: Arc::new(Mutex::new(HashMap::new())),
31            receiver: event_hub.subscribe_receiver(),
32            next_subscriber_id: Arc::new(AtomicU64::new(0)),
33        }
34    }
35
36    /// Subscribe a callback to an origin. The returned `SubscriptionToken`
37    /// removes the callback from the subscribers map when dropped — hold it
38    /// for as long as you want the callback to fire, then drop it to
39    /// unsubscribe.
40    pub fn subscribe<F>(&self, origin: Origin, callback: F) -> SubscriptionToken
41    where
42        F: Fn(Event) + Send + 'static,
43    {
44        let id = self.next_subscriber_id.fetch_add(1, Ordering::Relaxed);
45        {
46            let mut subs = self.subscribers.lock().unwrap();
47            subs.entry(origin.clone())
48                .or_default()
49                .push((id, Box::new(callback)));
50        }
51        SubscriptionToken {
52            subscribers: Arc::clone(&self.subscribers),
53            origin,
54            id,
55        }
56    }
57
58    /// Start the event loop in a background thread
59    /// Blocks on the flume receiver — no polling, zero CPU when idle
60    pub fn start(&self, quit_signal: Arc<std::sync::atomic::AtomicBool>) {
61        let receiver = self.receiver.clone();
62        let subscribers = Arc::clone(&self.subscribers);
63        let quit_signal = Arc::clone(&quit_signal);
64
65        log::info!("EventHubClient starting event loop");
66
67        thread::spawn(move || {
68            log::info!("EventHubClient event loop started");
69            loop {
70                match receiver.recv_timeout(std::time::Duration::from_millis(200)) {
71                    Ok(event) => {
72                        log::debug!("EventHubClient received event: {:?}", event);
73                        let subs = subscribers.lock().unwrap();
74                        if let Some(callbacks) = subs.get(&event.origin) {
75                            for (_id, callback) in callbacks {
76                                callback(event.clone());
77                            }
78                        }
79                    }
80                    Err(flume::RecvTimeoutError::Timeout) => {
81                        // Just check quit signal
82                    }
83                    Err(flume::RecvTimeoutError::Disconnected) => {
84                        log::info!("EventHubClient channel disconnected");
85                        break;
86                    }
87                }
88
89                if quit_signal.load(std::sync::atomic::Ordering::Relaxed) {
90                    log::info!("EventHubClient quitting event loop");
91                    break;
92                }
93            }
94        });
95    }
96}
97
98/// Opaque handle returned by `EventHubClient::subscribe`. Dropping it removes
99/// the associated callback from the subscribers map; the origin's entry is
100/// removed entirely when its last subscriber goes away.
101pub struct SubscriptionToken {
102    subscribers: Arc<Mutex<HashMap<Origin, SubscriberList>>>,
103    origin: Origin,
104    id: u64,
105}
106
107impl Drop for SubscriptionToken {
108    fn drop(&mut self) {
109        // Best-effort removal. If another thread poisoned the mutex, skip
110        // cleanup rather than double-panic during unwind.
111        if let Ok(mut subs) = self.subscribers.lock()
112            && let Some(list) = subs.get_mut(&self.origin)
113        {
114            list.retain(|(id, _)| *id != self.id);
115            if list.is_empty() {
116                subs.remove(&self.origin);
117            }
118        }
119    }
120}
text_document_frontend/event_hub_client.rs

text_document_frontend/
event_hub_client.rs
