hive_btle/mesh/
manager.rs

1//! Mesh Manager
2//!
3//! Manages the mesh topology, connections, and provides parent failover.
4
5#[cfg(not(feature = "std"))]
6use alloc::{boxed::Box, collections::BTreeMap, vec::Vec};
7#[cfg(feature = "std")]
8use std::collections::HashMap;
9
10use core::sync::atomic::{AtomicUsize, Ordering};
11
12#[cfg(feature = "std")]
13use std::sync::RwLock;
14
15use crate::discovery::HiveBeacon;
16use crate::error::{BleError, Result};
17use crate::{HierarchyLevel, NodeId};
18
19use super::topology::{
20    ConnectionState, DisconnectReason, MeshTopology, ParentCandidate, PeerInfo, PeerRole,
21    TopologyConfig, TopologyEvent,
22};
23
24/// Callback type for topology events
25pub type TopologyCallback = Box<dyn Fn(&TopologyEvent) + Send + Sync>;
26
27/// Mesh manager state
28#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
29pub enum ManagerState {
30    /// Not started
31    #[default]
32    Stopped,
33    /// Starting up
34    Starting,
35    /// Running and managing topology
36    Running,
37    /// In parent failover mode
38    Failover,
39    /// Stopping
40    Stopping,
41}
42
43/// Manages the BLE mesh topology
44///
45/// Responsible for:
46/// - Tracking parent/child/peer connections
47/// - Parent selection and failover
48/// - Connection lifecycle management
49/// - Publishing topology events
50#[cfg(feature = "std")]
51pub struct MeshManager {
52    /// Our node ID
53    node_id: NodeId,
54    /// Our hierarchy level
55    my_level: HierarchyLevel,
56    /// Configuration
57    config: TopologyConfig,
58    /// Current topology state
59    topology: RwLock<MeshTopology>,
60    /// Connected peer info
61    peers: RwLock<HashMap<NodeId, PeerInfo>>,
62    /// Parent candidates from beacons
63    candidates: RwLock<Vec<ParentCandidate>>,
64    /// Current state
65    state: RwLock<ManagerState>,
66    /// Event callbacks
67    callbacks: RwLock<Vec<TopologyCallback>>,
68    /// Monotonic time in milliseconds (for testing without system time)
69    /// Using AtomicUsize for 32-bit platform compatibility (ESP32)
70    current_time_ms: AtomicUsize,
71}
72
73#[cfg(feature = "std")]
74impl MeshManager {
75    /// Create a new mesh manager
76    pub fn new(node_id: NodeId, my_level: HierarchyLevel, config: TopologyConfig) -> Self {
77        let topology = MeshTopology::new(my_level, config.max_children, config.max_connections);
78
79        Self {
80            node_id,
81            my_level,
82            config,
83            topology: RwLock::new(topology),
84            peers: RwLock::new(HashMap::new()),
85            candidates: RwLock::new(Vec::new()),
86            state: RwLock::new(ManagerState::Stopped),
87            callbacks: RwLock::new(Vec::new()),
88            current_time_ms: AtomicUsize::new(0),
89        }
90    }
91
92    /// Get our node ID
93    pub fn node_id(&self) -> &NodeId {
94        &self.node_id
95    }
96
97    /// Get our hierarchy level
98    pub fn my_level(&self) -> HierarchyLevel {
99        self.my_level
100    }
101
102    /// Get current state
103    pub fn state(&self) -> ManagerState {
104        *self.state.read().unwrap()
105    }
106
107    /// Start the mesh manager
108    pub fn start(&self) -> Result<()> {
109        let mut state = self.state.write().unwrap();
110        match *state {
111            ManagerState::Stopped => {
112                *state = ManagerState::Running;
113                Ok(())
114            }
115            _ => Err(BleError::InvalidState("Already started".into())),
116        }
117    }
118
119    /// Stop the mesh manager
120    pub fn stop(&self) -> Result<()> {
121        let mut state = self.state.write().unwrap();
122        *state = ManagerState::Stopped;
123
124        // Clear topology
125        let mut topology = self.topology.write().unwrap();
126        topology.parent = None;
127        topology.children.clear();
128        topology.peers.clear();
129
130        // Clear peers
131        self.peers.write().unwrap().clear();
132
133        // Clear candidates
134        self.candidates.write().unwrap().clear();
135
136        Ok(())
137    }
138
139    /// Register a callback for topology events
140    pub fn on_topology_event(&self, callback: TopologyCallback) {
141        self.callbacks.write().unwrap().push(callback);
142    }
143
144    /// Emit a topology event to all listeners
145    fn emit_event(&self, event: TopologyEvent) {
146        let callbacks = self.callbacks.read().unwrap();
147        for callback in callbacks.iter() {
148            callback(&event);
149        }
150    }
151
152    /// Set the current time (for testing or embedded without RTC)
153    /// Note: Uses usize internally for 32-bit platform compatibility
154    pub fn set_time_ms(&self, time_ms: u64) {
155        self.current_time_ms
156            .store(time_ms as usize, Ordering::SeqCst);
157    }
158
159    /// Get the current time
160    /// Note: Returns u64 but internally stored as usize for 32-bit compatibility
161    pub fn time_ms(&self) -> u64 {
162        self.current_time_ms.load(Ordering::SeqCst) as u64
163    }
164
165    /// Get a snapshot of the current topology
166    pub fn topology(&self) -> MeshTopology {
167        self.topology.read().unwrap().clone()
168    }
169
170    /// Check if we have a parent
171    pub fn has_parent(&self) -> bool {
172        self.topology.read().unwrap().has_parent()
173    }
174
175    /// Get our parent's node ID
176    pub fn parent(&self) -> Option<NodeId> {
177        self.topology.read().unwrap().parent
178    }
179
180    /// Get list of children
181    pub fn children(&self) -> Vec<NodeId> {
182        self.topology.read().unwrap().children.clone()
183    }
184
185    /// Get number of children
186    pub fn child_count(&self) -> usize {
187        self.topology.read().unwrap().children.len()
188    }
189
190    /// Check if we can accept more children
191    pub fn can_accept_child(&self) -> bool {
192        self.topology.read().unwrap().can_accept_child()
193    }
194
195    /// Get all connected peer IDs
196    pub fn connected_peers(&self) -> Vec<NodeId> {
197        self.topology.read().unwrap().all_connected()
198    }
199
200    /// Get peer info for a node
201    pub fn get_peer_info(&self, node_id: &NodeId) -> Option<PeerInfo> {
202        self.peers.read().unwrap().get(node_id).cloned()
203    }
204
205    /// Process a beacon from a discovered node
206    ///
207    /// This updates our list of potential parents
208    pub fn process_beacon(&self, beacon: &HiveBeacon, rssi: i8) {
209        // Only consider nodes at higher hierarchy levels as parent candidates
210        if beacon.hierarchy_level > self.my_level {
211            let candidate = ParentCandidate {
212                node_id: beacon.node_id,
213                level: beacon.hierarchy_level,
214                rssi,
215                age_ms: 0,
216                failure_count: self
217                    .peers
218                    .read()
219                    .unwrap()
220                    .get(&beacon.node_id)
221                    .map(|p| p.failure_count)
222                    .unwrap_or(0),
223            };
224
225            let mut candidates = self.candidates.write().unwrap();
226
227            // Update existing or add new
228            if let Some(existing) = candidates.iter_mut().find(|c| c.node_id == beacon.node_id) {
229                existing.rssi = rssi;
230                existing.age_ms = 0;
231                existing.level = beacon.hierarchy_level;
232            } else {
233                candidates.push(candidate);
234            }
235        }
236    }
237
238    /// Select best parent from candidates
239    ///
240    /// Returns the best candidate based on RSSI, age, and failure history
241    pub fn select_best_parent(&self) -> Option<ParentCandidate> {
242        let candidates = self.candidates.read().unwrap();
243
244        candidates
245            .iter()
246            .filter(|c| {
247                c.rssi >= self.config.min_parent_rssi
248                    && c.age_ms <= self.config.max_beacon_age_ms
249                    && c.failure_count < self.config.max_failures
250            })
251            .max_by_key(|c| c.score(self.my_level))
252            .cloned()
253    }
254
255    /// Connect to a node as our parent
256    pub fn connect_parent(&self, node_id: NodeId, level: HierarchyLevel, rssi: i8) -> Result<()> {
257        let mut topology = self.topology.write().unwrap();
258
259        if topology.has_parent() {
260            return Err(BleError::InvalidState("Already have a parent".into()));
261        }
262
263        if !topology.set_parent(node_id) {
264            return Err(BleError::ConnectionFailed(
265                "Cannot accept connection".into(),
266            ));
267        }
268
269        // Add peer info
270        let mut peer_info = PeerInfo::new(node_id, PeerRole::Parent, level);
271        peer_info.state = ConnectionState::Connected;
272        peer_info.rssi = Some(rssi);
273        peer_info.connected_at = Some(self.time_ms());
274        peer_info.last_seen_ms = self.time_ms();
275
276        self.peers.write().unwrap().insert(node_id, peer_info);
277
278        // Emit event
279        drop(topology); // Release lock before emitting
280        self.emit_event(TopologyEvent::ParentConnected {
281            node_id,
282            level,
283            rssi: Some(rssi),
284        });
285
286        self.emit_topology_changed();
287        Ok(())
288    }
289
290    /// Disconnect from our parent
291    pub fn disconnect_parent(&self, reason: DisconnectReason) -> Option<NodeId> {
292        let old_parent = {
293            let mut topology = self.topology.write().unwrap();
294            topology.clear_parent()
295        };
296
297        if let Some(ref parent_id) = old_parent {
298            self.peers.write().unwrap().remove(parent_id);
299
300            self.emit_event(TopologyEvent::ParentDisconnected {
301                node_id: *parent_id,
302                reason,
303            });
304            self.emit_topology_changed();
305        }
306
307        old_parent
308    }
309
310    /// Accept a child connection
311    pub fn accept_child(&self, node_id: NodeId, level: HierarchyLevel) -> Result<()> {
312        let mut topology = self.topology.write().unwrap();
313
314        if !topology.add_child(node_id) {
315            return Err(BleError::ConnectionFailed("Cannot accept child".into()));
316        }
317
318        // Add peer info
319        let mut peer_info = PeerInfo::new(node_id, PeerRole::Child, level);
320        peer_info.state = ConnectionState::Connected;
321        peer_info.connected_at = Some(self.time_ms());
322        peer_info.last_seen_ms = self.time_ms();
323
324        self.peers.write().unwrap().insert(node_id, peer_info);
325
326        // Emit event
327        drop(topology);
328        self.emit_event(TopologyEvent::ChildConnected { node_id, level });
329
330        self.emit_topology_changed();
331        Ok(())
332    }
333
334    /// Remove a child
335    pub fn remove_child(&self, node_id: &NodeId, reason: DisconnectReason) -> bool {
336        let removed = {
337            let mut topology = self.topology.write().unwrap();
338            topology.remove_child(node_id)
339        };
340
341        if removed {
342            self.peers.write().unwrap().remove(node_id);
343
344            self.emit_event(TopologyEvent::ChildDisconnected {
345                node_id: *node_id,
346                reason,
347            });
348            self.emit_topology_changed();
349        }
350
351        removed
352    }
353
354    /// Start parent failover process
355    pub fn start_failover(&self) -> Result<()> {
356        let mut state = self.state.write().unwrap();
357        if *state != ManagerState::Running {
358            return Err(BleError::InvalidState("Not running".into()));
359        }
360
361        let old_parent = self.disconnect_parent(DisconnectReason::LinkLoss);
362
363        if let Some(old_parent_id) = old_parent {
364            *state = ManagerState::Failover;
365            drop(state);
366
367            self.emit_event(TopologyEvent::ParentFailoverStarted {
368                old_parent: old_parent_id,
369            });
370        }
371
372        Ok(())
373    }
374
375    /// Complete failover by connecting to new parent
376    pub fn complete_failover(
377        &self,
378        new_parent: Option<(NodeId, HierarchyLevel, i8)>,
379    ) -> Result<()> {
380        let old_parent = {
381            // Get old parent from candidates list (it was stored there)
382            self.candidates
383                .read()
384                .unwrap()
385                .first()
386                .map(|c| c.node_id)
387                .unwrap_or_else(|| NodeId::new(0))
388        };
389
390        if let Some((node_id, level, rssi)) = new_parent {
391            self.connect_parent(node_id, level, rssi)?;
392
393            let mut state = self.state.write().unwrap();
394            *state = ManagerState::Running;
395            drop(state);
396
397            self.emit_event(TopologyEvent::ParentFailoverCompleted {
398                old_parent,
399                new_parent: Some(node_id),
400            });
401        } else {
402            let mut state = self.state.write().unwrap();
403            *state = ManagerState::Running;
404            drop(state);
405
406            self.emit_event(TopologyEvent::ParentFailoverCompleted {
407                old_parent,
408                new_parent: None,
409            });
410        }
411
412        Ok(())
413    }
414
415    /// Update RSSI for a connected peer
416    pub fn update_rssi(&self, node_id: &NodeId, rssi: i8) {
417        let mut peers = self.peers.write().unwrap();
418        if let Some(peer) = peers.get_mut(node_id) {
419            peer.update_rssi(rssi);
420            peer.last_seen_ms = self.time_ms();
421        }
422        drop(peers);
423
424        self.emit_event(TopologyEvent::ConnectionQualityChanged {
425            node_id: *node_id,
426            rssi,
427        });
428    }
429
430    /// Record a connection failure for a node
431    pub fn record_failure(&self, node_id: &NodeId) {
432        let mut peers = self.peers.write().unwrap();
433        if let Some(peer) = peers.get_mut(node_id) {
434            peer.record_failure();
435        }
436
437        // Also update candidate failure count
438        let mut candidates = self.candidates.write().unwrap();
439        if let Some(candidate) = candidates.iter_mut().find(|c| &c.node_id == node_id) {
440            candidate.failure_count = candidate.failure_count.saturating_add(1);
441        }
442    }
443
444    /// Age all candidates (call periodically)
445    pub fn age_candidates(&self, elapsed_ms: u64) {
446        let mut candidates = self.candidates.write().unwrap();
447        for candidate in candidates.iter_mut() {
448            candidate.age_ms = candidate.age_ms.saturating_add(elapsed_ms);
449        }
450
451        // Remove candidates that are too old
452        candidates.retain(|c| c.age_ms <= self.config.max_beacon_age_ms * 2);
453    }
454
455    /// Check if we should switch parents (better option available)
456    pub fn should_switch_parent(&self) -> Option<ParentCandidate> {
457        let topology = self.topology.read().unwrap();
458        let current_parent = topology.parent?;
459        drop(topology);
460
461        let peers = self.peers.read().unwrap();
462        let current_rssi = peers.get(&current_parent)?.rssi?;
463        drop(peers);
464
465        // Find best alternative
466        let best = self.select_best_parent()?;
467
468        // Only switch if significantly better (hysteresis)
469        if best.rssi > current_rssi + self.config.rssi_hysteresis as i8 {
470            Some(best)
471        } else {
472            None
473        }
474    }
475
476    /// Helper to emit topology changed event
477    fn emit_topology_changed(&self) {
478        let topology = self.topology.read().unwrap();
479        self.emit_event(TopologyEvent::TopologyChanged {
480            child_count: topology.children.len(),
481            peer_count: topology.peers.len(),
482            has_parent: topology.has_parent(),
483        });
484    }
485}
486
487#[cfg(test)]
488mod tests {
489    use super::*;
490
491    fn create_manager() -> MeshManager {
492        MeshManager::new(
493            NodeId::new(0x1234),
494            HierarchyLevel::Platform,
495            TopologyConfig::default(),
496        )
497    }
498
499    #[test]
500    fn test_manager_creation() {
501        let manager = create_manager();
502        assert_eq!(manager.node_id().as_u32(), 0x1234);
503        assert_eq!(manager.my_level(), HierarchyLevel::Platform);
504        assert_eq!(manager.state(), ManagerState::Stopped);
505    }
506
507    #[test]
508    fn test_start_stop() {
509        let manager = create_manager();
510
511        assert!(manager.start().is_ok());
512        assert_eq!(manager.state(), ManagerState::Running);
513
514        assert!(manager.stop().is_ok());
515        assert_eq!(manager.state(), ManagerState::Stopped);
516    }
517
518    #[test]
519    fn test_connect_parent() {
520        let manager = create_manager();
521        manager.start().unwrap();
522
523        let parent_id = NodeId::new(0x5678);
524        assert!(manager
525            .connect_parent(parent_id, HierarchyLevel::Squad, -50)
526            .is_ok());
527
528        assert!(manager.has_parent());
529        assert_eq!(manager.parent(), Some(parent_id));
530
531        // Can't connect another parent
532        assert!(manager
533            .connect_parent(NodeId::new(0x9999), HierarchyLevel::Squad, -50)
534            .is_err());
535    }
536
537    #[test]
538    fn test_disconnect_parent() {
539        let manager = create_manager();
540        manager.start().unwrap();
541
542        let parent_id = NodeId::new(0x5678);
543        manager
544            .connect_parent(parent_id, HierarchyLevel::Squad, -50)
545            .unwrap();
546
547        let old = manager.disconnect_parent(DisconnectReason::Requested);
548        assert_eq!(old, Some(parent_id));
549        assert!(!manager.has_parent());
550    }
551
552    #[test]
553    fn test_accept_child() {
554        let manager = MeshManager::new(
555            NodeId::new(0x1234),
556            HierarchyLevel::Squad,
557            TopologyConfig::default(),
558        );
559        manager.start().unwrap();
560
561        let child_id = NodeId::new(0x0001);
562        assert!(manager
563            .accept_child(child_id, HierarchyLevel::Platform)
564            .is_ok());
565
566        assert_eq!(manager.child_count(), 1);
567        assert_eq!(manager.children(), vec![child_id]);
568    }
569
570    #[test]
571    fn test_max_children() {
572        let config = TopologyConfig {
573            max_children: 2,
574            ..Default::default()
575        };
576
577        let manager = MeshManager::new(NodeId::new(0x1234), HierarchyLevel::Squad, config);
578        manager.start().unwrap();
579
580        assert!(manager
581            .accept_child(NodeId::new(0x0001), HierarchyLevel::Platform)
582            .is_ok());
583        assert!(manager
584            .accept_child(NodeId::new(0x0002), HierarchyLevel::Platform)
585            .is_ok());
586        assert!(manager
587            .accept_child(NodeId::new(0x0003), HierarchyLevel::Platform)
588            .is_err());
589    }
590
591    #[test]
592    fn test_process_beacon() {
593        let manager = create_manager();
594        manager.start().unwrap();
595
596        let beacon = HiveBeacon {
597            node_id: NodeId::new(0x5678),
598            hierarchy_level: HierarchyLevel::Squad,
599            version: 1,
600            seq_num: 1,
601            capabilities: 0,
602            battery_percent: 100,
603            geohash: 0,
604        };
605
606        manager.process_beacon(&beacon, -50);
607
608        let best = manager.select_best_parent();
609        assert!(best.is_some());
610        assert_eq!(best.unwrap().node_id.as_u32(), 0x5678);
611    }
612
613    #[test]
614    fn test_select_best_parent_rssi() {
615        let manager = create_manager();
616        manager.start().unwrap();
617
618        // Add two candidates
619        let beacon1 = HiveBeacon {
620            node_id: NodeId::new(0x1111),
621            hierarchy_level: HierarchyLevel::Squad,
622            version: 1,
623            seq_num: 1,
624            capabilities: 0,
625            battery_percent: 100,
626            geohash: 0,
627        };
628
629        let beacon2 = HiveBeacon {
630            node_id: NodeId::new(0x2222),
631            hierarchy_level: HierarchyLevel::Squad,
632            version: 1,
633            seq_num: 1,
634            capabilities: 0,
635            battery_percent: 100,
636            geohash: 0,
637        };
638
639        manager.process_beacon(&beacon1, -70);
640        manager.process_beacon(&beacon2, -50); // Better RSSI
641
642        let best = manager.select_best_parent().unwrap();
643        assert_eq!(best.node_id.as_u32(), 0x2222);
644    }
645
646    #[test]
647    fn test_failover() {
648        let manager = create_manager();
649        manager.start().unwrap();
650
651        let parent_id = NodeId::new(0x5678);
652        manager
653            .connect_parent(parent_id, HierarchyLevel::Squad, -50)
654            .unwrap();
655
656        // Start failover
657        assert!(manager.start_failover().is_ok());
658        assert_eq!(manager.state(), ManagerState::Failover);
659        assert!(!manager.has_parent());
660
661        // Complete without new parent
662        assert!(manager.complete_failover(None).is_ok());
663        assert_eq!(manager.state(), ManagerState::Running);
664    }
665
666    #[test]
667    fn test_event_callback() {
668        use std::sync::atomic::{AtomicBool, Ordering};
669        use std::sync::Arc;
670
671        let manager = create_manager();
672        manager.start().unwrap();
673
674        let called = Arc::new(AtomicBool::new(false));
675        let called_clone = called.clone();
676
677        manager.on_topology_event(Box::new(move |event| {
678            if matches!(event, TopologyEvent::ParentConnected { .. }) {
679                called_clone.store(true, Ordering::SeqCst);
680            }
681        }));
682
683        manager
684            .connect_parent(NodeId::new(0x5678), HierarchyLevel::Squad, -50)
685            .unwrap();
686
687        assert!(called.load(Ordering::SeqCst));
688    }
689
690    #[test]
691    fn test_update_rssi() {
692        let manager = create_manager();
693        manager.start().unwrap();
694
695        let parent_id = NodeId::new(0x5678);
696        manager
697            .connect_parent(parent_id, HierarchyLevel::Squad, -50)
698            .unwrap();
699
700        manager.update_rssi(&parent_id, -60);
701
702        let info = manager.get_peer_info(&parent_id).unwrap();
703        assert_eq!(info.rssi, Some(-60));
704    }
705
706    #[test]
707    fn test_age_candidates() {
708        let manager = create_manager();
709        manager.start().unwrap();
710
711        let beacon = HiveBeacon {
712            node_id: NodeId::new(0x5678),
713            hierarchy_level: HierarchyLevel::Squad,
714            version: 1,
715            seq_num: 1,
716            capabilities: 0,
717            battery_percent: 100,
718            geohash: 0,
719        };
720
721        manager.process_beacon(&beacon, -50);
722
723        // Age past the threshold
724        manager.age_candidates(25_000);
725
726        // Should be removed
727        let best = manager.select_best_parent();
728        assert!(best.is_none());
729    }
730}
hive_btle/mesh/manager.rs

hive_btle/mesh/
manager.rs
