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 parking_lot::Mutex;
5use std::collections::HashMap;
6use std::sync::Arc;
7use std::sync::atomic::{AtomicU64, Ordering};
8use std::thread;
9
10/// Event callback type
11pub type EventCallback = Box<dyn Fn(Event) + Send>;
12
13/// Internal storage: each subscriber entry pairs its id (for O(n) removal on
14/// token drop) with its callback. Vec preserves registration order so
15/// subscribers fire deterministically.
16type SubscriberList = Vec<(u64, EventCallback)>;
17
18/// EventHubClient - handles event passing from backend to UI
19/// Subscribe callbacks to specific event origins and start the event loop.
20#[derive(Clone)]
21pub struct EventHubClient {
22    subscribers: Arc<Mutex<HashMap<Origin, SubscriberList>>>,
23    receiver: Receiver<Event>,
24    next_subscriber_id: Arc<AtomicU64>,
25}
26
27impl EventHubClient {
28    /// Create a new event hub client
29    pub fn new(event_hub: &EventHub) -> Self {
30        EventHubClient {
31            subscribers: Arc::new(Mutex::new(HashMap::new())),
32            receiver: event_hub.subscribe_receiver(),
33            next_subscriber_id: Arc::new(AtomicU64::new(0)),
34        }
35    }
36
37    /// Subscribe a callback to an origin. The returned `SubscriptionToken`
38    /// removes the callback from the subscribers map when dropped — hold it
39    /// for as long as you want the callback to fire, then drop it to
40    /// unsubscribe.
41    pub fn subscribe<F>(&self, origin: Origin, callback: F) -> SubscriptionToken
42    where
43        F: Fn(Event) + Send + 'static,
44    {
45        let id = self.next_subscriber_id.fetch_add(1, Ordering::Relaxed);
46        {
47            let mut subs = self.subscribers.lock();
48            subs.entry(origin.clone())
49                .or_default()
50                .push((id, Box::new(callback)));
51        }
52        SubscriptionToken {
53            subscribers: Arc::clone(&self.subscribers),
54            origin,
55            id,
56        }
57    }
58
59    /// Start the event loop in a background thread.
60    ///
61    /// The thread blocks on a `flume::Selector` that waits on either an
62    /// incoming event or the shutdown receiver. Zero CPU while idle —
63    /// no polling, no timeout. The thread exits when the matching
64    /// shutdown `Sender` is dropped (which makes `shutdown_rx`
65    /// `Disconnected`) or when the event hub's sender drops.
66    pub fn start(&self, shutdown_rx: Receiver<()>) {
67        let receiver = self.receiver.clone();
68        let subscribers = Arc::clone(&self.subscribers);
69
70        log::info!("EventHubClient starting event loop");
71
72        thread::spawn(move || {
73            log::info!("EventHubClient event loop started");
74            loop {
75                // True blocking wait. Both branches reduce to
76                // `Result<Option<Event>, ()>`: `Ok(Some(event))` for a
77                // real delivery, `Ok(None)` for shutdown (either the
78                // shutdown sender was dropped or a `()` was actually
79                // sent), `Err(())` for event-channel disconnect.
80                let outcome: Result<Option<Event>, ()> = Selector::new()
81                    .recv(&receiver, |r| r.map(Some).map_err(|_| ()))
82                    .recv(&shutdown_rx, |_| Ok(None))
83                    .wait();
84                match outcome {
85                    Ok(Some(event)) => {
86                        log::debug!("EventHubClient received event: {:?}", event);
87                        let subs = subscribers.lock();
88                        if let Some(callbacks) = subs.get(&event.origin) {
89                            for (_id, callback) in callbacks {
90                                callback(event.clone());
91                            }
92                        }
93                    }
94                    Ok(None) => {
95                        log::info!("EventHubClient quitting event loop");
96                        break;
97                    }
98                    Err(()) => {
99                        log::info!("EventHubClient channel disconnected");
100                        break;
101                    }
102                }
103            }
104        });
105    }
106}
107
108/// Opaque handle returned by `EventHubClient::subscribe`. Dropping it removes
109/// the associated callback from the subscribers map; the origin's entry is
110/// removed entirely when its last subscriber goes away.
111pub struct SubscriptionToken {
112    subscribers: Arc<Mutex<HashMap<Origin, SubscriberList>>>,
113    origin: Origin,
114    id: u64,
115}
116
117impl Drop for SubscriptionToken {
118    fn drop(&mut self) {
119        // Best-effort removal. parking_lot mutexes don't poison, so the
120        // lock always succeeds (no double-panic risk during unwind).
121        let mut subs = self.subscribers.lock();
122        if let Some(list) = subs.get_mut(&self.origin) {
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
