bacnet_objects/subscribed_recipients.rs
1//! A Notification Forwarder's Subscribed_Recipients (Clause 12.51.9, #1049):
2//! a bounded BACnetLIST of BACnetEventNotificationSubscription whose entries
3//! lapse when their time runs out.
4//!
5//! The bundled [`NotificationForwarderObject`] keeps its list in a
6//! [`SubscribedRecipients`]. An application that implements its own type-51
7//! object can do the same, routing to the store the property's reads and
8//! writes ([`read`](SubscribedRecipients::read) and
9//! [`write`](SubscribedRecipients::write)) and the object's monotonic clock
10//! hooks ([`bind_monotonic_clock`](SubscribedRecipients::bind_monotonic_clock),
11//! [`advance_to`](SubscribedRecipients::advance_to) and
12//! [`next_deadline`](SubscribedRecipients::next_deadline)). The server then
13//! serves the list to ReadProperty, ReadPropertyMultiple and ReadRange, and
14//! its AddListElement and RemoveListElement handlers edit it the way the
15//! clause asks: an entry is named by its recipient and process identifier, and
16//! adding one already present renews it with the other members given.
17//!
18//! [`NotificationForwarderObject`]: crate::notification_forwarder::NotificationForwarderObject
19//!
20//! # Lifetimes
21//!
22//! Time Remaining travels in minutes. The store keeps each entry's deadline on
23//! the database's monotonic clock, the one the server's operation task uses
24//! for the Access Door's pulse relock, and serves the whole minutes left,
25//! rounded up, so a live entry never reads as zero. An entry leaves the list
26//! once its deadline passes: reads and writes skip it from then on, and the
27//! server's operation task drops it at its deadline (through
28//! `advance_monotonic_time_internal`). With no clock bound the store counts
29//! the time [`advance_by`](SubscribedRecipients::advance_by) gives it, and
30//! binding a clock later keeps the time each entry has left.
31//!
32//! Clause 12.51.9 asks for the list to survive a restart. The store itself
33//! lives in memory; the forwarder object saves it through a
34//! [`NotificationForwarderPersistence`] and restores it with
35//! [`write`](SubscribedRecipients::write).
36//!
37//! [`NotificationForwarderPersistence`]: crate::notification_forwarder::NotificationForwarderPersistence
38
39use crate::prelude::*;
40use alloc::sync::Arc;
41use core::fmt;
42use core::time::Duration;
43
44use bacnet_encoding::constructed::{
45 decode_event_notification_subscription, encode_event_notification_subscription_list,
46};
47use bacnet_types::constructed::{BACnetEventNotificationSubscription, BACnetRecipient};
48use bacnet_types::enums::{ErrorClass, ErrorCode};
49use bacnet_types::error::Error;
50use bacnet_types::primitives::PropertyValue;
51use bytes::BytesMut;
52
53use crate::common;
54use crate::traits::MonotonicClock;
55
56/// The most entries a [`SubscribedRecipients`] holds.
57///
58/// An entry naming a device encodes in at most 17 octets, and one naming an
59/// address with the longest MAC this stack carries
60/// ([`BACnetAddress::MAX_MAC_LEN`], 18 octets) in at most 37. A full list
61/// therefore takes at most 1,184 octets, and a ReadProperty of it fits B/IP's
62/// 1476-octet APDU unsegmented. It matches the Notification Class
63/// Recipient_List cap ([`MAX_RECIPIENT_LIST_DESTINATIONS`]).
64///
65/// A write that would leave more entries fails with RESOURCES /
66/// NO_SPACE_TO_WRITE_PROPERTY naming the first entry past the cap, which
67/// AddListElement reports as NO_SPACE_TO_ADD_LIST_ELEMENT at the request
68/// element that brought it (Clause 15.1.1.3). A renewal never counts against
69/// the cap.
70///
71/// [`BACnetAddress::MAX_MAC_LEN`]: bacnet_types::constructed::BACnetAddress::MAX_MAC_LEN
72/// [`MAX_RECIPIENT_LIST_DESTINATIONS`]: crate::notification_class::MAX_RECIPIENT_LIST_DESTINATIONS
73pub const MAX_SUBSCRIBED_RECIPIENTS: usize = 32;
74
75/// The longest Time Remaining, in minutes, the store accepts: 1,440 (a day).
76///
77/// Clause 12.51.9 requires a forwarder to take 1 through 1,440 and leaves
78/// longer subscriptions to the implementation. This one refuses them, and
79/// refuses 0, which no entry may hold, with PROPERTY / VALUE_OUT_OF_RANGE
80/// naming the entry.
81pub const MAX_SUBSCRIPTION_MINUTES: u32 = 1440;
82
83const NANOS_PER_MINUTE: u128 = 60_000_000_000;
84
85/// One held entry, its lifetime as a monotonic deadline.
86#[derive(Clone)]
87struct Entry {
88 recipient: BACnetRecipient,
89 process_identifier: u32,
90 issue_confirmed_notifications: bool,
91 expires_at: Duration,
92}
93
94impl Entry {
95 /// Whether `other` is the same entry: the same recipient and process
96 /// identifier, whatever its other members.
97 fn names_the_same(&self, other: &Entry) -> bool {
98 self.recipient == other.recipient && self.process_identifier == other.process_identifier
99 }
100
101 fn is_live(&self, now: Duration) -> bool {
102 self.expires_at > now
103 }
104
105 /// The entry as served at `now`, with the whole minutes left rounded up.
106 fn served(&self, now: Duration) -> BACnetEventNotificationSubscription {
107 let minutes = self
108 .expires_at
109 .saturating_sub(now)
110 .as_nanos()
111 .div_ceil(NANOS_PER_MINUTE);
112 BACnetEventNotificationSubscription {
113 recipient: self.recipient.clone(),
114 process_identifier: self.process_identifier,
115 issue_confirmed_notifications: self.issue_confirmed_notifications,
116 time_remaining: u32::try_from(minutes).unwrap_or(u32::MAX),
117 }
118 }
119}
120
121/// A Notification Forwarder's Subscribed_Recipients: at most
122/// [`MAX_SUBSCRIBED_RECIPIENTS`] entries, each lapsing when its Time
123/// Remaining runs out. See the [module documentation](self).
124#[derive(Clone, Default)]
125pub struct SubscribedRecipients {
126 entries: Vec<Entry>,
127 monotonic_clock: Option<Arc<MonotonicClock>>,
128 /// Elapsed time for a store with no bound clock, advanced by
129 /// [`Self::advance_by`].
130 logical_now: Duration,
131}
132
133impl fmt::Debug for SubscribedRecipients {
134 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
135 f.debug_struct("SubscribedRecipients")
136 .field("subscriptions", &self.subscriptions())
137 .finish_non_exhaustive()
138 }
139}
140
141impl SubscribedRecipients {
142 /// An empty list.
143 pub fn new() -> Self {
144 Self::default()
145 }
146
147 fn now(&self) -> Duration {
148 self.monotonic_clock
149 .as_ref()
150 .map_or(self.logical_now, |clock| clock())
151 }
152
153 /// The store's current time: the bound clock's reading, or the time
154 /// [`advance_by`](Self::advance_by) has counted.
155 pub(crate) fn current_time(&self) -> Duration {
156 self.now()
157 }
158
159 /// The live entries in list order, each with the whole minutes it has
160 /// left, rounded up.
161 pub fn subscriptions(&self) -> Vec<BACnetEventNotificationSubscription> {
162 let now = self.now();
163 self.entries
164 .iter()
165 .filter(|entry| entry.is_live(now))
166 .map(|entry| entry.served(now))
167 .collect()
168 }
169
170 /// The property value: the live entries, encoded back to back in
171 /// `PropertyValue::ApplicationData`.
172 pub fn read(&self) -> PropertyValue {
173 let mut buf = BytesMut::new();
174 // Every entry arrived through `write`, whose decoder holds recipients
175 // to BACnetAddress::MAX_MAC_LEN (#1156), so each one encodes.
176 encode_event_notification_subscription_list(&mut buf, &self.subscriptions())
177 .expect("stored recipients fit BACnetAddress::MAX_MAC_LEN");
178 PropertyValue::ApplicationData(buf.to_vec())
179 }
180
181 /// Replace the list with a written one, all or nothing.
182 ///
183 /// Only the framed form in `PropertyValue::ApplicationData` is a value;
184 /// anything else, or an entry that does not decode, fails with PROPERTY /
185 /// INVALID_DATA_TYPE. An entry whose Time Remaining is 0 or past
186 /// [`MAX_SUBSCRIPTION_MINUTES`] fails with PROPERTY / VALUE_OUT_OF_RANGE,
187 /// and the first entry past [`MAX_SUBSCRIBED_RECIPIENTS`] with RESOURCES /
188 /// NO_SPACE_TO_WRITE_PROPERTY; both name the entry's position in the
189 /// written list.
190 ///
191 /// Each entry's Time Remaining starts its lifetime over, except that an
192 /// entry written as it reads now keeps its deadline, and so does one
193 /// written with a minute more, which is how it read if a minute boundary
194 /// fell between the read and this write. A list read and written back, or
195 /// edited by the list services, so leaves the entries it doesn't change
196 /// alone, rather than stretching each to its rounded-up minute. Two
197 /// entries naming the same recipient and process are one entry, which
198 /// takes the later one's members.
199 pub fn write(&mut self, value: PropertyValue) -> Result<(), Error> {
200 let PropertyValue::ApplicationData(bytes) = value else {
201 return Err(common::invalid_data_type_error());
202 };
203 let now = self.now();
204 let mut written: Vec<Entry> = Vec::new();
205 let mut offset = 0;
206 let mut index = 0;
207 while offset < bytes.len() {
208 let (subscription, next) = decode_event_notification_subscription(&bytes, offset)
209 .map_err(|_| common::invalid_data_type_error())?;
210 offset = next;
211 if !(1..=MAX_SUBSCRIPTION_MINUTES).contains(&subscription.time_remaining) {
212 return Err(common::at_list_element(
213 common::value_out_of_range_error(),
214 index,
215 ));
216 }
217 let entry = self.entry_for(subscription, now);
218 match written.iter().position(|held| held.names_the_same(&entry)) {
219 Some(held) => written[held] = entry,
220 None if written.len() == MAX_SUBSCRIBED_RECIPIENTS => {
221 return Err(common::at_list_element(
222 common::protocol_error(
223 ErrorClass::RESOURCES,
224 ErrorCode::NO_SPACE_TO_WRITE_PROPERTY,
225 ),
226 index,
227 ));
228 }
229 None => written.push(entry),
230 }
231 index += 1;
232 }
233 self.entries = written;
234 Ok(())
235 }
236
237 /// The entry a written `subscription` becomes at `now`. A live entry with
238 /// the same recipient, process and confirmation flag keeps its deadline
239 /// when the written minutes equal what it serves now or one more.
240 fn entry_for(&self, subscription: BACnetEventNotificationSubscription, now: Duration) -> Entry {
241 let lifetime = Duration::from_secs(u64::from(subscription.time_remaining) * 60);
242 let expires_at = self
243 .entries
244 .iter()
245 .find(|entry| {
246 if !entry.is_live(now) {
247 return false;
248 }
249 let served = entry.served(now);
250 served.recipient == subscription.recipient
251 && served.process_identifier == subscription.process_identifier
252 && served.issue_confirmed_notifications
253 == subscription.issue_confirmed_notifications
254 && (served.time_remaining..=served.time_remaining.saturating_add(1))
255 .contains(&subscription.time_remaining)
256 })
257 .map_or_else(|| now.saturating_add(lifetime), |entry| entry.expires_at);
258 Entry {
259 recipient: subscription.recipient,
260 process_identifier: subscription.process_identifier,
261 issue_confirmed_notifications: subscription.issue_confirmed_notifications,
262 expires_at,
263 }
264 }
265
266 /// Bind the database's monotonic clock, from the object's
267 /// `bind_monotonic_clock_internal`. Each held entry keeps the time it has
268 /// left: its deadline moves from the old time base to the new one, so a
269 /// list written or restored before the object joins a database lapses on
270 /// time.
271 pub fn bind_monotonic_clock(&mut self, clock: Option<Arc<MonotonicClock>>) {
272 let before = self.now();
273 self.monotonic_clock = clock;
274 let after = self.now();
275 for entry in &mut self.entries {
276 let left = entry.expires_at.saturating_sub(before);
277 entry.expires_at = after.saturating_add(left);
278 }
279 }
280
281 /// Drop every entry whose deadline is at or before `now`, a monotonic
282 /// instant, from the object's `advance_monotonic_time_internal`. `true`
283 /// when one was dropped.
284 pub fn advance_to(&mut self, now: Duration) -> bool {
285 let held = self.entries.len();
286 self.entries.retain(|entry| entry.is_live(now));
287 self.entries.len() != held
288 }
289
290 /// Count `elapsed` on a store with no bound clock, then drop what has
291 /// lapsed, as [`Self::advance_to`] does. `true` when an entry was dropped.
292 pub fn advance_by(&mut self, elapsed: Duration) -> bool {
293 self.logical_now = self.logical_now.saturating_add(elapsed);
294 self.advance_to(self.now())
295 }
296
297 /// The earliest deadline among the held entries, for the object's
298 /// `next_monotonic_deadline_internal`.
299 pub fn next_deadline(&self) -> Option<Duration> {
300 self.entries.iter().map(|entry| entry.expires_at).min()
301 }
302}
303
304#[cfg(test)]
305#[path = "subscribed_recipients_tests.rs"]
306mod tests;