1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
// Generated by Qleany v1.3.10 from frontend_event_hub_client.tera
use common::event::{Event, EventHub, Origin};
use flume::{Receiver, Selector};
use std::collections::HashMap;
use std::sync::{Arc, Mutex};
use std::thread;
/// Event callback type
pub type EventCallback = Box<dyn Fn(Event) + Send>;
/// EventHubClient - handles event passing from backend to UI
/// Subscribe callbacks to specific event origins and start the event loop.
#[derive(Clone)]
pub struct EventHubClient {
subscribers: Arc<Mutex<HashMap<Origin, Vec<EventCallback>>>>,
receiver: Receiver<Event>,
}
impl EventHubClient {
/// Create a new event hub client
pub fn new(event_hub: &EventHub) -> Self {
EventHubClient {
subscribers: Arc::new(Mutex::new(HashMap::new())),
receiver: event_hub.subscribe_receiver(),
}
}
/// Subscribe a callback to an origin
pub fn subscribe<F>(&self, origin: Origin, callback: F)
where
F: Fn(Event) + Send + 'static,
{
let mut subs = self.subscribers.lock().unwrap();
subs.entry(origin).or_default().push(Box::new(callback));
}
/// Start the event loop in a background thread.
///
/// The thread blocks on a `flume::Selector` that waits on either an
/// incoming event or the shutdown receiver. Zero CPU while idle —
/// no polling, no timeout. The thread exits when the matching
/// shutdown `Sender` is dropped (which makes `shutdown_rx`
/// `Disconnected`) or when the event hub's sender drops.
pub fn start(&self, shutdown_rx: Receiver<()>) {
let receiver = self.receiver.clone();
let subscribers = Arc::clone(&self.subscribers);
log::info!("EventHubClient starting event loop");
thread::spawn(move || {
log::info!("EventHubClient event loop started");
loop {
// True blocking wait. Both branches reduce to
// `Result<Option<Event>, ()>`: `Ok(Some(event))` for a
// real delivery, `Ok(None)` for shutdown (either the
// shutdown sender was dropped or a `()` was actually
// sent), `Err(())` for event-channel disconnect.
let outcome: Result<Option<Event>, ()> = Selector::new()
.recv(&receiver, |r| r.map(Some).map_err(|_| ()))
.recv(&shutdown_rx, |_| Ok(None))
.wait();
match outcome {
Ok(Some(event)) => {
log::debug!("EventHubClient received event: {:?}", event);
let subs = subscribers.lock().unwrap();
if let Some(callbacks) = subs.get(&event.origin) {
for callback in callbacks {
callback(event.clone());
}
}
}
Ok(None) => {
log::info!("EventHubClient quitting event loop");
break;
}
Err(()) => {
log::info!("EventHubClient channel disconnected");
break;
}
}
}
});
}
}