atomr_cluster/
events.rs

1//! Cluster events bus.
2//!
3//! Events are published when membership transitions, leader changes, or
4//! reachability flips. Subscribers register a per-event-class callback (or
5//! a multi-class one via the [`ClusterEvent`] enum) and receive each event
6//! in publish order.
7//!
8//! The bus is a thin `RwLock<Vec<callback>>` rather than an actor
9//! because subscribers are typically a handful of long-lived objects
10//! (telemetry probes, sharding region, pubsub mediator) and the
11//! contention model is "rare write, rare read." Phase 13 may move it
12//! behind a real actor if profiling justifies it.
13
14use std::sync::Arc;
15
16use atomr_core::actor::Address;
17use parking_lot::RwLock;
18use serde::{Deserialize, Serialize};
19
20use crate::member::{Member, MemberStatus};
21
22/// Event variants published on [`ClusterEventBus`].
23#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
24#[non_exhaustive]
25pub enum ClusterEvent {
26    MemberJoined(Member),
27    MemberWeaklyUp(Member),
28    MemberUp(Member),
29    MemberLeft(Member),
30    MemberExited(Member),
31    MemberRemoved(Member, MemberStatus),
32    UnreachableMember(Member),
33    ReachableMember(Member),
34    LeaderChanged { from: Option<Address>, to: Option<Address> },
35    ClusterShuttingDown,
36    Convergence(bool),
37}
38
39impl ClusterEvent {
40    /// Translate a status transition `(old, new)` into the
41    /// corresponding `ClusterEvent`, mirroring
42    /// `ClusterCoreDaemon` event-emission rules. Returns `None` when
43    /// the transition is a no-op (`old == new`).
44    pub fn from_status_transition(member: Member, old: MemberStatus) -> Option<ClusterEvent> {
45        let new = member.status;
46        if old == new {
47            return None;
48        }
49        Some(match new {
50            MemberStatus::Joining => ClusterEvent::MemberJoined(member),
51            MemberStatus::WeaklyUp => ClusterEvent::MemberWeaklyUp(member),
52            MemberStatus::Up => ClusterEvent::MemberUp(member),
53            MemberStatus::Leaving => ClusterEvent::MemberLeft(member),
54            MemberStatus::Exiting => ClusterEvent::MemberExited(member),
55            MemberStatus::Down => ClusterEvent::UnreachableMember(member),
56            MemberStatus::Removed => ClusterEvent::MemberRemoved(member, old),
57        })
58    }
59}
60
61type Subscriber = Arc<dyn Fn(&ClusterEvent) + Send + Sync + 'static>;
62
63/// In-process cluster events bus.
64#[derive(Default, Clone)]
65pub struct ClusterEventBus {
66    inner: Arc<RwLock<Vec<Subscriber>>>,
67}
68
69impl ClusterEventBus {
70    pub fn new() -> Self {
71        Self::default()
72    }
73
74    /// Register a subscriber that fires on every event. Returns a
75    /// handle whose `Drop` removes the subscription.
76    pub fn subscribe<F>(&self, callback: F) -> SubscriptionHandle
77    where
78        F: Fn(&ClusterEvent) + Send + Sync + 'static,
79    {
80        let cb: Subscriber = Arc::new(callback);
81        let mut subs = self.inner.write();
82        subs.push(cb.clone());
83        SubscriptionHandle {
84            bus: self.inner.clone(),
85            // Use the Arc pointer identity to find this subscription on drop.
86            id: Arc::as_ptr(&cb) as *const () as usize,
87            anchor: cb,
88        }
89    }
90
91    /// Publish an event to all current subscribers, synchronously,
92    /// in registration order. Subscribers must not block.
93    pub fn publish(&self, event: ClusterEvent) {
94        let subs = self.inner.read().clone();
95        for s in &subs {
96            s(&event);
97        }
98    }
99
100    /// Number of currently registered subscribers.
101    pub fn subscriber_count(&self) -> usize {
102        self.inner.read().len()
103    }
104}
105
106/// RAII handle returned by [`ClusterEventBus::subscribe`]. Dropping it
107/// removes the corresponding subscriber.
108pub struct SubscriptionHandle {
109    bus: Arc<RwLock<Vec<Subscriber>>>,
110    id: usize,
111    /// Keeps the `Arc` alive so the pointer identity matches on drop.
112    anchor: Subscriber,
113}
114
115impl Drop for SubscriptionHandle {
116    fn drop(&mut self) {
117        let mut subs = self.bus.write();
118        subs.retain(|s| Arc::as_ptr(s) as *const () as usize != self.id);
119        // anchor goes out of scope after retain, so the Arc count drops.
120        let _ = &self.anchor;
121    }
122}
123
124#[cfg(test)]
125mod tests {
126    use super::*;
127    use std::sync::atomic::{AtomicU32, Ordering};
128
129    #[test]
130    fn publish_delivers_to_subscribers() {
131        let bus = ClusterEventBus::new();
132        let n = Arc::new(AtomicU32::new(0));
133        let n2 = n.clone();
134        let _h = bus.subscribe(move |_| {
135            n2.fetch_add(1, Ordering::SeqCst);
136        });
137        bus.publish(ClusterEvent::ClusterShuttingDown);
138        bus.publish(ClusterEvent::Convergence(true));
139        assert_eq!(n.load(Ordering::SeqCst), 2);
140    }
141
142    #[test]
143    fn drop_unsubscribes() {
144        let bus = ClusterEventBus::new();
145        let n = Arc::new(AtomicU32::new(0));
146        let n2 = n.clone();
147        let h = bus.subscribe(move |_| {
148            n2.fetch_add(1, Ordering::SeqCst);
149        });
150        assert_eq!(bus.subscriber_count(), 1);
151        drop(h);
152        assert_eq!(bus.subscriber_count(), 0);
153        bus.publish(ClusterEvent::Convergence(false));
154        assert_eq!(n.load(Ordering::SeqCst), 0);
155    }
156
157    #[test]
158    fn leader_changed_carries_old_and_new() {
159        let bus = ClusterEventBus::new();
160        let captured = Arc::new(parking_lot::Mutex::new(None));
161        let c2 = captured.clone();
162        let _h = bus.subscribe(move |e| {
163            *c2.lock() = Some(e.clone());
164        });
165        bus.publish(ClusterEvent::LeaderChanged { from: None, to: Some(Address::local("a")) });
166        let got = captured.lock().clone();
167        assert!(matches!(got, Some(ClusterEvent::LeaderChanged { from: None, to: Some(_) })));
168    }
169}
atomr_cluster/events.rs

atomr_cluster/
events.rs
