bacnet_server/server/
broadcaster.rs

1//! Local sends share the server's admission seal and joined request lifetime.
2use super::*;
3use std::sync::{Mutex as SyncMutex, Weak};
4
5pub(super) struct BroadcasterState<T: TransportPort> {
6    network: SyncMutex<Option<Arc<NetworkLayer<T>>>>,
7    requests: Weak<request_tasks::RequestTasks>,
8    config: Arc<ServerConfig>,
9    /// Weak, so a server dropped without `stop()` is left holding the only
10    /// handle it hands off the runtime (#1409).
11    db: Weak<RwLock<ObjectDatabase>>,
12    /// The server's DeviceCommunicationControl state, read before each send.
13    comm_state: Arc<CommState>,
14    capacity: Arc<Semaphore>,
15}
16
17pub(super) fn stopped() -> Error {
18    Error::Encoding("server is stopping or stopped".into())
19}
20
21impl<T: TransportPort> BroadcasterState<T> {
22    pub(super) fn new(
23        network: &Arc<NetworkLayer<T>>,
24        requests: &Arc<request_tasks::RequestTasks>,
25        config: &Arc<ServerConfig>,
26        db: &Arc<RwLock<ObjectDatabase>>,
27        comm_state: &Arc<CommState>,
28    ) -> Arc<Self> {
29        Arc::new(Self {
30            network: SyncMutex::new(Some(Arc::clone(network))),
31            requests: Arc::downgrade(requests),
32            config: Arc::clone(config),
33            db: Arc::downgrade(db),
34            comm_state: Arc::clone(comm_state),
35            capacity: Arc::new(Semaphore::new(32)),
36        })
37    }
38
39    pub(super) fn seal(&self) {
40        // Serializes with clone + registration, including callers that already
41        // upgraded a weak handle on another thread. Never held across await.
42        self.network.lock().unwrap().take();
43        self.capacity.close();
44    }
45
46    pub(super) fn is_open(&self) -> bool {
47        self.network.lock().unwrap().is_some()
48    }
49}
50
51impl<T: TransportPort + 'static> BroadcasterState<T> {
52    fn admit(
53        &self,
54        limiter: Option<Arc<DiscoveryLimiter>>,
55    ) -> Result<oneshot::Receiver<Result<(), Error>>, Error> {
56        let slot = self.network.lock().unwrap();
57        let network = Arc::clone(slot.as_ref().ok_or_else(stopped)?);
58        let permit = Arc::clone(&self.capacity)
59            .try_acquire_owned()
60            .map_err(|_| Error::Encoding("server I-Am broadcast capacity exhausted".into()))?;
61        let requests = self.requests.upgrade().ok_or_else(stopped)?;
62        let config = Arc::clone(&self.config);
63        let db = self.db.upgrade().ok_or_else(stopped)?;
64        let comm_state = Arc::clone(&self.comm_state);
65        let (tx, rx) = oneshot::channel();
66        requests.spawn(async move {
67            let _permit = permit;
68            let result = discovery::broadcast_i_am_from(
69                &config,
70                &db,
71                &network,
72                &comm_state,
73                limiter.as_ref(),
74            )
75            .await;
76            let _ = tx.send(result);
77        });
78        Ok(rx)
79    }
80
81    pub(super) async fn send(&self, limiter: Option<Arc<DiscoveryLimiter>>) -> Result<(), Error> {
82        self.admit(limiter)?.await.map_err(|_| stopped())?
83    }
84}
85
86impl<T: TransportPort + 'static> IAmBroadcaster<T> {
87    /// Send through the running server; overload or shutdown returns an error,
88    /// and so does DeviceCommunicationControl restricting initiation (see
89    /// [`BACnetServer::broadcast_i_am`]). Cancelling this waiter does not
90    /// cancel an already admitted send.
91    pub async fn broadcast_i_am(&self) -> Result<(), Error> {
92        let completion = self.state.upgrade().ok_or_else(stopped)?.admit(None)?;
93        completion.await.map_err(|_| stopped())?
94    }
95}
96
97impl<T: TransportPort + 'static> BACnetServer<T> {
98    #[cfg(test)]
99    pub(super) fn test_network(&self) -> &Arc<NetworkLayer<T>> {
100        self.network
101            .as_ref()
102            .expect("test requires retained network")
103    }
104
105    pub(super) fn active_network(&self) -> Result<&Arc<NetworkLayer<T>>, Error> {
106        if !self.broadcaster.is_open() {
107            return Err(stopped());
108        }
109        self.network.as_ref().ok_or_else(stopped)
110    }
111}
bacnet_server/server/broadcaster.rs

bacnet_server/server/
broadcaster.rs
