bacnet_server/
cov.rs

1//! COV subscription engine — tracks SubscribeCOV subscriptions and their lifetimes.
2
3use std::collections::HashMap;
4use std::sync::atomic::Ordering;
5use std::sync::Arc;
6use std::time::Instant;
7
8use bacnet_encoding::npdu::NpduAddress;
9use bacnet_types::enums::{ErrorClass, ErrorCode, PropertyIdentifier};
10use bacnet_types::error::Error;
11use bacnet_types::primitives::ObjectIdentifier;
12use bacnet_types::MacAddr;
13
14mod identity;
15pub use identity::*;
16pub(crate) mod active;
17mod admission;
18pub use admission::MultipleRefusal;
19mod confirmed;
20pub(crate) use confirmed::{BeginRefusal, CovRevisits};
21mod sample;
22pub use sample::CovSample;
23mod observation;
24mod observation_order;
25pub use observation::CovObservation;
26use observation_order::ObservationOwner;
27pub(crate) use observation_order::PreparedCovCompletion;
28pub(crate) mod flags;
29mod lifetime;
30pub(crate) mod multiple_reads;
31pub(crate) mod prepare;
32pub(crate) mod reported;
33pub(crate) mod timed;
34mod timed_capture;
35pub(crate) use timed_capture::TimedWriteCapture;
36pub(crate) mod value_source;
37pub use lifetime::CovTimeRemaining;
38
39mod policy;
40pub use policy::*;
41
42#[cfg(test)]
43mod identity_tests;
44#[cfg(test)]
45mod policy_tests;
46#[cfg(test)]
47mod tests;
48
49/// Largest B/IP APDU; the default history bound until a server sets its own.
50const DEFAULT_TIMED_APDU_LENGTH: usize = 1476;
51
52/// Proposed COV subscription data. Table acceptance validates its canonical
53/// identity and returns an immutable [`CovSubscriptionSnapshot`] for delivery.
54#[derive(Debug, Clone)]
55pub struct CovSubscription {
56    /// Immediate delivery MAC: the local subscriber or its current router.
57    pub subscriber_mac: MacAddr,
58    /// Routed source address when the subscriber is behind a BACnet router.
59    pub subscriber_network: Option<NpduAddress>,
60    /// Process identifier chosen by the subscriber.
61    pub subscriber_process_identifier: u32,
62    /// The object being monitored.
63    pub monitored_object_identifier: ObjectIdentifier,
64    /// Notification form. Mutable renewal data for ordinary/Single subscriptions;
65    /// part of the canonical identity for Multiple references.
66    pub issue_confirmed_notifications: bool,
67    /// When this subscription expires (None = infinite lifetime).
68    pub expires_at: Option<Instant>,
69    /// Last delivered bounded observation, including specialized command fields.
70    pub last_notified_observation: Option<CovObservation>,
71    /// Monitored property for Single-property and Multiple-reference subscriptions.
72    pub monitored_property: Option<PropertyIdentifier>,
73    /// Accepted property index; absent, zero and element indexes are independent.
74    pub monitored_property_array_index: Option<u32>,
75    /// COV increment override (SubscribeCOVProperty only).
76    pub cov_increment: Option<f32>,
77    /// Notification service family used for this subscription.
78    pub notification_kind: CovNotificationKind,
79    /// Whether COVNotificationMultiple values should include timeOfChange.
80    pub timestamped: bool,
81}
82
83impl CovSubscription {
84    /// Original client address shared by subscription identity and accounting.
85    /// Process, family and monitored coordinates additionally distinguish subscriptions.
86    pub fn recipient(&self) -> CovRecipient {
87        CovRecipient::from_endpoint(&self.subscriber_mac, self.subscriber_network.as_ref())
88    }
89}
90
91/// COV notification service family for a stored subscription.
92#[derive(Debug, Clone, Copy, PartialEq, Eq)]
93pub enum CovNotificationKind {
94    /// COVNotification / ConfirmedCOVNotification.
95    Single,
96    /// COVNotificationMultiple / ConfirmedCOVNotificationMultiple.
97    Multiple,
98}
99
100/// Table of active COV subscriptions.
101#[derive(Debug)]
102pub struct CovSubscriptionTable {
103    subs: HashMap<CovSubscriptionKey, CovSubscriptionSnapshot>,
104    generation: u64,
105    owner: Arc<ObservationOwner>,
106    peer_counts: HashMap<CovRecipient, usize>,
107    peer_indefinite_counts: HashMap<CovRecipient, usize>,
108    policy: CovPolicy,
109    counters: Arc<AtomicCovCounters>,
110    in_flight: Arc<CovInFlightTracker>,
111    revisits: Arc<CovRevisits>,
112    dispatch_turn: usize,
113    timed: timed::TimedStore,
114    /// Live list samples taken for read requests, so tests can see which
115    /// requests snapshot the table (#1213).
116    #[cfg(test)]
117    live_samples: std::sync::atomic::AtomicUsize,
118}
119
120impl Default for CovSubscriptionTable {
121    fn default() -> Self {
122        Self::new()
123    }
124}
125
126impl CovSubscriptionTable {
127    /// Create a new COV subscription table with default policy and counters.
128    pub fn new() -> Self {
129        Self::with_policy(CovPolicy::default(), Arc::new(AtomicCovCounters::default()))
130    }
131
132    /// Create a new COV subscription table with a custom policy and counters.
133    pub fn with_policy(policy: CovPolicy, counters: Arc<AtomicCovCounters>) -> Self {
134        let timed = timed::TimedStore::new(DEFAULT_TIMED_APDU_LENGTH, Arc::clone(&counters));
135        Self {
136            subs: HashMap::new(),
137            generation: 0,
138            owner: Arc::new(ObservationOwner::default()),
139            peer_counts: HashMap::new(),
140            peer_indefinite_counts: HashMap::new(),
141            policy: policy.sanitized(),
142            counters,
143            in_flight: Arc::new(CovInFlightTracker::default()),
144            revisits: Arc::default(),
145            dispatch_turn: 0,
146            timed,
147            #[cfg(test)]
148            live_samples: std::sync::atomic::AtomicUsize::new(0),
149        }
150    }
151
152    /// Bound each Multiple context's pending timestamped changes by what one
153    /// notification of this maximum APDU length can carry.
154    pub fn with_max_apdu_length(mut self, max_apdu_length: usize) -> Self {
155        self.timed = timed::TimedStore::new(max_apdu_length, Arc::clone(&self.counters));
156        self
157    }
158
159    /// Pending timestamped COV-multiple changes of this table's references.
160    pub(crate) fn timed(&self) -> &timed::TimedStore {
161        &self.timed
162    }
163
164    /// Return the next notification dispatch turn counter (wrapping).
165    pub(crate) fn next_dispatch_turn(&mut self) -> usize {
166        let turn = self.dispatch_turn;
167        self.dispatch_turn = self.dispatch_turn.wrapping_add(1);
168        turn
169    }
170
171    /// Get a reference to the active COV policy.
172    pub fn policy(&self) -> &CovPolicy {
173        &self.policy
174    }
175
176    /// Get a reference to the atomic counters.
177    pub fn counters(&self) -> &Arc<AtomicCovCounters> {
178        &self.counters
179    }
180
181    /// Get a reference to the in-flight confirmed notification tracker.
182    pub fn in_flight_tracker(&self) -> &Arc<CovInFlightTracker> {
183        &self.in_flight
184    }
185
186    /// Get the number of active subscriptions for a peer.
187    pub fn peer_subscription_count(&self, peer: &CovRecipient) -> usize {
188        self.peer_counts.get(peer).copied().unwrap_or(0)
189    }
190
191    /// Get the number of active indefinite subscriptions for a peer.
192    pub fn peer_indefinite_count(&self, peer: &CovRecipient) -> usize {
193        self.peer_indefinite_counts.get(peer).copied().unwrap_or(0)
194    }
195
196    /// Get an accepted entry by its complete typed identity.
197    pub fn get_subscription(&self, key: &CovSubscriptionKey) -> Option<&CovSubscriptionSnapshot> {
198        self.subs.get(key)
199    }
200
201    /// Every accepted reference of one Multiple context.
202    pub(crate) fn multiple_context_references<'a>(
203        &'a self,
204        context: &'a MultipleContextKey,
205    ) -> impl Iterator<Item = &'a CovSubscriptionSnapshot> + 'a {
206        self.subs
207            .iter()
208            .filter(move |(key, _)| key.multiple_context() == Some(context))
209            .map(|(_, sub)| sub)
210    }
211
212    /// Whether an exact subscription identity is present.
213    pub fn contains(&self, key: &CovSubscriptionKey) -> bool {
214        self.subs.contains_key(key)
215    }
216
217    fn remove_internal(&mut self, key: &CovSubscriptionKey, was_cancelled: bool) -> bool {
218        if let Some(sub) = self.subs.remove(key) {
219            self.timed.lock().remove(key);
220            self.revisits.forget(key);
221            let peer = sub.recipient();
222            if let Some(count) = self.peer_counts.get_mut(&peer) {
223                *count = count.saturating_sub(1);
224                if *count == 0 {
225                    self.peer_counts.remove(&peer);
226                }
227            }
228            if sub.expires_at.is_none() {
229                if let Some(count) = self.peer_indefinite_counts.get_mut(&peer) {
230                    *count = count.saturating_sub(1);
231                    if *count == 0 {
232                        self.peer_indefinite_counts.remove(&peer);
233                    }
234                }
235            }
236            if was_cancelled {
237                self.counters
238                    .subscriptions_cancelled
239                    .fetch_add(1, Ordering::Relaxed);
240            }
241            self.counters
242                .subscriptions_active
243                .store(self.subs.len() as u64, Ordering::Relaxed);
244            true
245        } else {
246            false
247        }
248    }
249
250    /// Cancel only the exact family/form/property/index identity.
251    pub fn unsubscribe(&mut self, key: &CovSubscriptionKey) -> bool {
252        self.remove_internal(key, true)
253    }
254
255    /// Cancel every reference in the exact Multiple context.
256    pub fn unsubscribe_cov_multiple_context(&mut self, context: &MultipleContextKey) {
257        let keys: Vec<_> = self
258            .subs
259            .keys()
260            .filter(|key| key.multiple_context() == Some(context))
261            .cloned()
262            .collect();
263        for key in keys {
264            self.remove_internal(&key, true);
265        }
266    }
267
268    /// Remove all subscriptions for a given object (used on DeleteObject).
269    pub fn remove_for_object(&mut self, oid: ObjectIdentifier) {
270        let to_remove: Vec<_> = self
271            .subs
272            .iter()
273            .filter(|(k, _)| k.object() == oid)
274            .map(|(k, _)| k.clone())
275            .collect();
276        for key in to_remove {
277            self.remove_internal(&key, false);
278        }
279    }
280
281    /// Remove subscriptions using this current route and release shared peer quotas.
282    /// An obsolete router cannot remove a subscription migrated elsewhere.
283    pub fn remove_peer_subscriptions(
284        &mut self,
285        mac: &[u8],
286        network: Option<&NpduAddress>,
287    ) -> usize {
288        let target = SubscriberEndpoint::new(mac, network);
289        let to_remove: Vec<_> = self
290            .subs
291            .iter()
292            .filter(|(_, entry)| entry.endpoint() == target)
293            .map(|(k, _)| k.clone())
294            .collect();
295        let count = to_remove.len();
296        for key in to_remove {
297            self.remove_internal(&key, true);
298        }
299        count
300    }
301
302    /// Remove all expired subscriptions. Returns the number removed.
303    pub fn purge_expired(&mut self) -> usize {
304        let now = Instant::now();
305        let mut purged_count = 0;
306        let mut to_remove = Vec::new();
307        for (k, sub) in &self.subs {
308            if sub.expires_at.is_some_and(|exp| exp <= now) {
309                to_remove.push(k.clone());
310            }
311        }
312        for key in to_remove {
313            purged_count += usize::from(self.remove_internal(&key, false));
314        }
315        if purged_count > 0 {
316            self.counters
317                .subscriptions_purged
318                .fetch_add(purged_count as u64, Ordering::Relaxed);
319            self.counters
320                .subscriptions_active
321                .store(self.subs.len() as u64, Ordering::Relaxed);
322        }
323        purged_count
324    }
325
326    /// Test fixture: let every subscription's lifetime run out now, without
327    /// waiting on the wall clock that lifetimes use.
328    #[cfg(test)]
329    pub(crate) fn expire_all_for_test(&mut self) {
330        let now = Instant::now();
331        for entry in self.subs.values_mut() {
332            entry.subscription.expires_at = Some(now);
333        }
334    }
335
336    /// Get all active (non-expired) subscriptions for a given object.
337    pub fn subscriptions_for(&mut self, oid: &ObjectIdentifier) -> Vec<&CovSubscriptionSnapshot> {
338        self.purge_expired();
339        self.subs
340            .values()
341            .filter(|sub| sub.monitored_object_identifier == *oid)
342            .collect()
343    }
344
345    /// Whether a snapshot still owns a live entry in this table.
346    pub fn is_current(&self, snapshot: &CovSubscriptionSnapshot) -> bool {
347        self.remaining_lifetime(snapshot, Instant::now())
348            .and_then(CovTimeRemaining::wire_seconds)
349            .is_some()
350    }
351
352    /// Ordinary whole-object trigger policy: a numeric Present_Value must move
353    /// by the increment, any other value must change, and the first report
354    /// always fires. Status_Flags and other Table 13-1 trigger values (such as
355    /// Staging's Present_Stage) are checked separately by the caller.
356    /// An unchanged object therefore reports nothing, however often it is
357    /// fanned out. Property subscriptions compare their prepared sample instead.
358    pub fn should_notify(
359        sub: &CovSubscription,
360        current_value: Option<&CovSample>,
361        cov_increment: Option<f64>,
362    ) -> bool {
363        let Some(current) = current_value else {
364            return true;
365        };
366        current.reports(
367            sub.last_notified_observation
368                .as_ref()
369                .map(|observation| observation.sample()),
370            cov_increment,
371            true,
372        )
373    }
374
375    /// Number of active subscriptions.
376    pub fn len(&self) -> usize {
377        self.subs.len()
378    }
379
380    /// Whether the table is empty.
381    pub fn is_empty(&self) -> bool {
382        self.subs.is_empty()
383    }
384}
bacnet_server/cov.rs

bacnet_server/
cov.rs
