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, Selector};
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    ///
60    /// The thread blocks on a `flume::Selector` that waits on either an
61    /// incoming event or the shutdown receiver. Zero CPU while idle —
62    /// no polling, no timeout. The thread exits when the matching
63    /// shutdown `Sender` is dropped (which makes `shutdown_rx`
64    /// `Disconnected`) or when the event hub's sender drops.
65    pub fn start(&self, shutdown_rx: Receiver<()>) {
66        let receiver = self.receiver.clone();
67        let subscribers = Arc::clone(&self.subscribers);
68
69        log::info!("EventHubClient starting event loop");
70
71        thread::spawn(move || {
72            log::info!("EventHubClient event loop started");
73            loop {
74                // True blocking wait. Both branches reduce to
75                // `Result<Option<Event>, ()>`: `Ok(Some(event))` for a
76                // real delivery, `Ok(None)` for shutdown (either the
77                // shutdown sender was dropped or a `()` was actually
78                // sent), `Err(())` for event-channel disconnect.
79                let outcome: Result<Option<Event>, ()> = Selector::new()
80                    .recv(&receiver, |r| r.map(Some).map_err(|_| ()))
81                    .recv(&shutdown_rx, |_| Ok(None))
82                    .wait();
83                match outcome {
84                    Ok(Some(event)) => {
85                        log::debug!("EventHubClient received event: {:?}", event);
86                        let subs = subscribers.lock().unwrap();
87                        if let Some(callbacks) = subs.get(&event.origin) {
88                            for (_id, callback) in callbacks {
89                                callback(event.clone());
90                            }
91                        }
92                    }
93                    Ok(None) => {
94                        log::info!("EventHubClient quitting event loop");
95                        break;
96                    }
97                    Err(()) => {
98                        log::info!("EventHubClient channel disconnected");
99                        break;
100                    }
101                }
102            }
103        });
104    }
105}
106
107/// Opaque handle returned by `EventHubClient::subscribe`. Dropping it removes
108/// the associated callback from the subscribers map; the origin's entry is
109/// removed entirely when its last subscriber goes away.
110pub struct SubscriptionToken {
111    subscribers: Arc<Mutex<HashMap<Origin, SubscriberList>>>,
112    origin: Origin,
113    id: u64,
114}
115
116impl Drop for SubscriptionToken {
117    fn drop(&mut self) {
118        // Best-effort removal. If another thread poisoned the mutex, skip
119        // cleanup rather than double-panic during unwind.
120        if let Ok(mut subs) = self.subscribers.lock()
121            && let Some(list) = subs.get_mut(&self.origin)
122        {
123            list.retain(|(id, _)| *id != self.id);
124            if list.is_empty() {
125                subs.remove(&self.origin);
126            }
127        }
128    }
129}
text_document_frontend/event_hub_client.rs

text_document_frontend/
event_hub_client.rs
