superfluid_daemon/
bus.rs

1//! The session event bus.
2
3use std::collections::HashMap;
4use std::sync::atomic::{AtomicBool, Ordering};
5use std::sync::mpsc::{Receiver, SyncSender, TrySendError};
6use std::sync::{Arc, Mutex};
7
8use crate::store::CommittedEvent;
9
10const SUB_QUEUE: usize = 1024;
11
12#[derive(Debug, Clone)]
13pub enum BusMsg {
14    Committed(CommittedEvent),
15    Provisional { channel: u32, text: String },
16}
17
18#[derive(Debug, Clone, Copy, PartialEq, Eq)]
19pub enum EndReason {
20    Lagged,
21    PublisherClosed,
22    Purged,
23}
24
25struct Subscriber {
26    id: u64,
27    tx: SyncSender<BusMsg>,
28    provisional: bool,
29    lagged: Arc<AtomicBool>,
30    purged: Arc<AtomicBool>,
31}
32
33pub struct Subscription {
34    pub session: u64,
35    pub id: u64,
36    pub rx: Receiver<BusMsg>,
37    lagged: Arc<AtomicBool>,
38    purged: Arc<AtomicBool>,
39}
40
41impl Subscription {
42    pub fn take_receiver(&mut self) -> Receiver<BusMsg> {
43        let (_tx, dummy) = std::sync::mpsc::channel();
44        std::mem::replace(&mut self.rx, dummy)
45    }
46
47    pub fn end_reason(&self) -> EndReason {
48        if self.purged.load(Ordering::Acquire) {
49            EndReason::Purged
50        } else if self.lagged.load(Ordering::Acquire) {
51            EndReason::Lagged
52        } else {
53            EndReason::PublisherClosed
54        }
55    }
56}
57
58#[derive(Default)]
59pub struct SessionBus {
60    inner: Mutex<BusInner>,
61}
62
63#[derive(Default)]
64struct BusInner {
65    subs: HashMap<u64, Vec<Subscriber>>,
66    next_id: u64,
67}
68
69impl SessionBus {
70    pub fn new() -> Arc<SessionBus> {
71        Arc::new(SessionBus::default())
72    }
73
74    pub fn subscribe(&self, session: u64, provisional: bool) -> Subscription {
75        let (tx, rx) = std::sync::mpsc::sync_channel(SUB_QUEUE);
76        let lagged = Arc::new(AtomicBool::new(false));
77        let purged = Arc::new(AtomicBool::new(false));
78        let mut inner = self.inner.lock().expect("bus");
79        inner.next_id += 1;
80        let id = inner.next_id;
81        inner.subs.entry(session).or_default().push(Subscriber {
82            id,
83            tx,
84            provisional,
85            lagged: Arc::clone(&lagged),
86            purged: Arc::clone(&purged),
87        });
88        Subscription {
89            session,
90            id,
91            rx,
92            lagged,
93            purged,
94        }
95    }
96
97    pub fn end_session(&self, session: u64, reason: EndReason) {
98        let mut inner = self.inner.lock().expect("bus");
99        if let Some(subs) = inner.subs.remove(&session) {
100            for s in subs {
101                match reason {
102                    EndReason::Purged => s.purged.store(true, Ordering::Release),
103                    EndReason::Lagged => s.lagged.store(true, Ordering::Release),
104                    EndReason::PublisherClosed => {}
105                }
106            }
107        }
108    }
109
110    pub fn unsubscribe(&self, session: u64, id: u64) {
111        let mut inner = self.inner.lock().expect("bus");
112        if let Some(v) = inner.subs.get_mut(&session) {
113            v.retain(|s| s.id != id);
114            if v.is_empty() {
115                inner.subs.remove(&session);
116            }
117        }
118    }
119
120    pub fn has_provisional_subscriber(&self, session: u64) -> bool {
121        let inner = self.inner.lock().expect("bus");
122        inner
123            .subs
124            .get(&session)
125            .is_some_and(|v| v.iter().any(|s| s.provisional))
126    }
127
128    pub fn publish_committed(&self, session: u64, ev: &CommittedEvent) {
129        self.publish(session, |_| Some(BusMsg::Committed(ev.clone())));
130    }
131
132    pub fn publish_provisional(&self, session: u64, channel: u32, text: &str) {
133        self.publish(session, |s| {
134            s.provisional.then(|| BusMsg::Provisional {
135                channel,
136                text: text.to_string(),
137            })
138        });
139    }
140
141    fn publish(&self, session: u64, msg_for: impl Fn(&Subscriber) -> Option<BusMsg>) {
142        let mut inner = self.inner.lock().expect("bus");
143        let Some(subs) = inner.subs.get_mut(&session) else {
144            return;
145        };
146        subs.retain(|s| {
147            let Some(msg) = msg_for(s) else {
148                return true;
149            };
150            match s.tx.try_send(msg) {
151                Ok(()) => true,
152                Err(TrySendError::Full(_)) => {
153                    s.lagged.store(true, Ordering::Release);
154                    false
155                }
156                Err(TrySendError::Disconnected(_)) => false,
157            }
158        });
159        if subs.is_empty() {
160            inner.subs.remove(&session);
161        }
162    }
163}
superfluid_daemon/bus.rs

superfluid_daemon/
bus.rs
