bacnet_objects/notification_forwarder/
mod.rs

1//! Notification Forwarder object (type 51, Clause 12.51).
2//!
3//! A forwarder originates no events. It takes the event notifications this
4//! device receives, and those its own objects address to the local Device,
5//! and sends each one on to the destinations it holds: the configured
6//! Recipient_List and the expiring Subscribed_Recipients. The server does the
7//! sending; see [`forwarding_targets`] for which notifications a forwarder
8//! takes and where it sends them.
9//!
10//! # Rows
11//!
12//! The object serves every required row of Table 12-58, plus Description and,
13//! when the application configures one, Port_Filter. Over the network:
14//!
15//! - Recipient_List takes the same writes as a Notification Class's, capped
16//!   at [`MAX_RECIPIENT_LIST_DESTINATIONS`] destinations.
17//! - Subscribed_Recipients is a [`SubscribedRecipients`] store: entries run
18//!   out after their Time Remaining and are edited by AddListElement and
19//!   RemoveListElement.
20//! - Process_Identifier_Filter takes an Unsigned32, or NULL to forward every
21//!   process identifier; Local_Forwarding_Only and Out_Of_Service take a
22//!   BOOLEAN.
23//! - Port_Filter takes writes of its Enabled members only (Clause 12.51.11):
24//!   the array keeps its size and Port_IDs.
25//! - Reliability is always NO_FAULT_DETECTED and is not writable.
26//!
27//! # Restarts
28//!
29//! Clauses 12.51.8 and 12.51.9 ask for both lists to survive a restart. A
30//! forwarder built with [`NotificationForwarderObject::with_persistence`]
31//! saves Recipient_List and Subscribed_Recipients together, each
32//! subscription with the minutes it has left, when a write changes either
33//! list, when an entry lapses, and at most once a minute while the minutes
34//! the entries serve fall; it restores both when built again. A restored
35//! entry counts down from its saved minutes: no fewer than it had left when
36//! the device stopped, about a minute more at most, and never more than its
37//! last subscription gave it, so restarts do not keep an entry alive.
38//!
39//! Saves run on the forwarder's own writer thread, never while the database
40//! guard is held ([`crate::durable`]), and a list write that cannot be saved
41//! is refused with DEVICE / OPERATIONAL_PROBLEM, leaving the old list. A
42//! failed save is logged and counted ([`ForwarderSaveCounters`]); one the
43//! operation task made is retried a minute later.
44//!
45//! A written Recipient_List wins over the one the application configures:
46//! once a write has set the list and it was saved, a rebuilt forwarder
47//! serves the saved list and ignores [`add_destination`] calls. Until then
48//! the configured destinations apply at every start; they are not saved. A
49//! forwarder built with [`NotificationForwarderObject::new`] keeps both lists
50//! in memory only.
51//!
52//! [`add_destination`]: NotificationForwarderObject::add_destination
53
54use crate::prelude::*;
55use alloc::borrow::Cow;
56use alloc::sync::Arc;
57use core::time::Duration;
58
59use bacnet_types::constructed::{
60    BACnetDestination, BACnetEventNotificationSubscription, BACnetPortPermission,
61};
62use bacnet_types::enums::{ObjectType, PropertyIdentifier, Reliability};
63use bacnet_types::error::Error;
64use bacnet_types::primitives::{ObjectIdentifier, PropertyValue, StatusFlags};
65use bytes::BytesMut;
66
67use crate::common::{self, read_common_properties};
68use crate::durable::staged::{self, Step};
69use crate::durable::{DurableWrites, PendingWrite, SaveWait, StageStep};
70use crate::notification_class::recipient_list;
71use crate::subscribed_recipients::SubscribedRecipients;
72use crate::traits::{BACnetObject, MonotonicClock};
73
74mod metadata;
75mod persistence;
76mod port_filter;
77mod saving;
78mod selection;
79
80pub use crate::notification_class::MAX_RECIPIENT_LIST_DESTINATIONS;
81#[cfg(feature = "std")]
82pub use persistence::FileNotificationForwarderPersistence;
83pub use persistence::{ForwarderSnapshot, NotificationForwarderPersistence};
84pub use saving::ForwarderSaveCounters;
85pub use selection::{ForwardingInput, ForwardingTargets, forwarding_targets};
86
87/// BACnet Notification Forwarder object (type 51). See the
88/// [module documentation](self).
89///
90/// # Unwind safety
91///
92/// The forwarder is neither `UnwindSafe` nor `RefUnwindSafe`, by decision
93/// (#1452). Its Subscribed_Recipients keep the monotonic clock the server
94/// binds, a closure the caller supplies and runs on its own thread. Nothing
95/// catches a panic from it, so wrapping it in `AssertUnwindSafe`, as the
96/// save writer's parts are, would claim more than the type can promise.
97/// Caller clocks get no unwind-safety bound either.
98pub struct NotificationForwarderObject {
99    oid: ObjectIdentifier,
100    name: String,
101    description: String,
102    status_flags: StatusFlags,
103    reliability: Reliability,
104    out_of_service: bool,
105    recipient_list: Vec<BACnetDestination>,
106    subscribed_recipients: SubscribedRecipients,
107    process_identifier_filter: Option<u32>,
108    local_forwarding_only: bool,
109    port_filter: Option<Vec<BACnetPortPermission>>,
110    storage: Option<saving::Storage>,
111    /// Writes either list has taken, so a staged write can tell whether
112    /// another came between.
113    list_writes: u64,
114    /// A write set Recipient_List, now or before a restart, so storage keeps
115    /// it and configured destinations no longer apply.
116    recipient_list_written: bool,
117    save_counters: ForwarderSaveCounters,
118}
119
120impl NotificationForwarderObject {
121    /// A forwarder with empty lists that forwards every process identifier
122    /// for any device. Its lists are kept in memory only.
123    pub fn new(instance: u32, name: impl Into<String>) -> Result<Self, Error> {
124        Ok(Self {
125            oid: ObjectIdentifier::new(ObjectType::NOTIFICATION_FORWARDER, instance)?,
126            name: name.into(),
127            description: String::new(),
128            status_flags: StatusFlags::empty(),
129            reliability: Reliability::NO_FAULT_DETECTED,
130            out_of_service: false,
131            recipient_list: Vec::new(),
132            subscribed_recipients: SubscribedRecipients::new(),
133            process_identifier_filter: None,
134            local_forwarding_only: false,
135            port_filter: None,
136            storage: None,
137            list_writes: 0,
138            recipient_list_written: false,
139            save_counters: ForwarderSaveCounters::default(),
140        })
141    }
142
143    /// A forwarder that keeps its Recipient_List and Subscribed_Recipients
144    /// in `persistence`, starting from the lists saved there for this object,
145    /// if any.
146    ///
147    /// Fails when the saved lists cannot be loaded, or hold an entry a write
148    /// would refuse: a destination a framed write could not carry, a
149    /// subscription with a Time Remaining of 0 or past a day, or more entries
150    /// than a list's cap.
151    pub fn with_persistence(
152        instance: u32,
153        name: impl Into<String>,
154        persistence: Arc<dyn NotificationForwarderPersistence>,
155    ) -> Result<Self, Error> {
156        let mut forwarder = Self::new(instance, name)?;
157        if let Some(saved) = persistence.load(forwarder.oid)? {
158            if let Some(list) = saved.recipient_list {
159                if list.len() > MAX_RECIPIENT_LIST_DESTINATIONS {
160                    return Err(recipient_list::no_space_error());
161                }
162                for destination in &list {
163                    recipient_list::check_added(destination)?;
164                }
165                forwarder.recipient_list = list;
166                forwarder.recipient_list_written = true;
167            }
168            if !saved.subscribed_recipients.is_empty() {
169                forwarder
170                    .subscribed_recipients
171                    .write(framed(&saved.subscribed_recipients)?)?;
172            }
173        }
174        forwarder.storage = Some(saving::Storage::new(
175            forwarder.oid,
176            persistence,
177            forwarder.snapshot(),
178            forwarder.save_counters.clone(),
179        ));
180        Ok(forwarder)
181    }
182
183    /// This forwarder's save counters, shared with the object, so they stay
184    /// readable after it joins a database. They stay zero without
185    /// persistence.
186    pub fn save_counters(&self) -> ForwarderSaveCounters {
187        self.save_counters.clone()
188    }
189
190    /// Set the description string.
191    pub fn set_description(&mut self, description: impl Into<String>) {
192        self.description = description.into();
193    }
194
195    /// The process identifier a notification must carry to be forwarded, or
196    /// `None` (Process_Identifier_Filter NULL) to forward every one.
197    pub fn process_identifier_filter(&self) -> Option<u32> {
198        self.process_identifier_filter
199    }
200
201    /// Set Process_Identifier_Filter; `None` is NULL.
202    pub fn set_process_identifier_filter(&mut self, filter: Option<u32>) {
203        self.process_identifier_filter = filter;
204    }
205
206    /// Whether only notifications this device's own objects generate are
207    /// forwarded.
208    pub fn local_forwarding_only(&self) -> bool {
209        self.local_forwarding_only
210    }
211
212    /// Set Local_Forwarding_Only.
213    pub fn set_local_forwarding_only(&mut self, local_only: bool) {
214        self.local_forwarding_only = local_only;
215    }
216
217    /// Add a destination to Recipient_List, with the checks and cap of a
218    /// Notification Class's
219    /// [`add_destination`](crate::notification_class::NotificationClass::add_destination).
220    ///
221    /// This configures the list the application starts with; it is not
222    /// saved. On a forwarder [`with_persistence`](Self::with_persistence)
223    /// whose storage holds a written Recipient_List
224    /// ([`recipient_list_saved`](Self::recipient_list_saved)), the saved list
225    /// wins: the destination is checked but not added.
226    pub fn add_destination(&mut self, destination: BACnetDestination) -> Result<(), Error> {
227        recipient_list::check_added(&destination)?;
228        if self.recipient_list_saved() {
229            log_debug!(
230                forwarder = %self.oid,
231                "Saved Recipient_List kept over a configured destination"
232            );
233            return Ok(());
234        }
235        if self.recipient_list.len() >= MAX_RECIPIENT_LIST_DESTINATIONS {
236            return Err(recipient_list::no_space_error());
237        }
238        self.recipient_list.push(destination);
239        Ok(())
240    }
241
242    /// The Recipient_List destinations, in list order.
243    pub fn recipient_list(&self) -> &[BACnetDestination] {
244        &self.recipient_list
245    }
246
247    /// The live Subscribed_Recipients entries, each with the whole minutes it
248    /// has left.
249    pub fn subscriptions(&self) -> Vec<BACnetEventNotificationSubscription> {
250        self.subscribed_recipients.subscriptions()
251    }
252
253    /// Serve Port_Filter with one element per network port, or `None` to
254    /// leave the property out, as a device that does not route may
255    /// (Clause 12.51.11). The server receives through one port, Port_ID 0.
256    pub fn set_port_filter(&mut self, ports: Option<Vec<BACnetPortPermission>>) {
257        self.port_filter = ports;
258    }
259
260    /// The Port_Filter elements, when the property is served.
261    pub fn port_filter(&self) -> Option<&[BACnetPortPermission]> {
262        self.port_filter.as_deref()
263    }
264
265    /// Whether storage holds a Recipient_List a write set, now or before a
266    /// restart. [`add_destination`](Self::add_destination) then leaves the
267    /// list alone. Always false without persistence.
268    pub fn recipient_list_saved(&self) -> bool {
269        self.storage.is_some() && self.recipient_list_written
270    }
271
272    /// Block until the saves queued so far have run. Dropping the forwarder
273    /// waits for them too.
274    pub fn wait_for_saves(&self) {
275        if let Some(storage) = &self.storage {
276            storage.wait_idle();
277        }
278    }
279
280    /// What storage holds for the lists the forwarder serves.
281    fn snapshot(&self) -> ForwarderSnapshot {
282        ForwarderSnapshot {
283            recipient_list: self
284                .recipient_list_written
285                .then(|| self.recipient_list.clone()),
286            subscribed_recipients: self.subscribed_recipients.subscriptions(),
287        }
288    }
289
290    /// The list a write of `value` to `property` would leave on top of the
291    /// `earlier` steps of its request, or the write's refusal.
292    fn next_list(
293        &self,
294        earlier: &[Step<saving::NextList>],
295        property: PropertyIdentifier,
296        value: PropertyValue,
297    ) -> Result<saving::NextList, Error> {
298        if property == PropertyIdentifier::RECIPIENT_LIST {
299            return recipient_list::decode_write(value).map(saving::NextList::RecipientList);
300        }
301        // A written entry keeps the deadline of a live one it renews, so the
302        // store builds on the one an earlier step left.
303        let current = earlier
304            .iter()
305            .rev()
306            .find_map(|step| match &step.next {
307                saving::NextList::SubscribedRecipients(store) => Some(store),
308                saving::NextList::RecipientList(_) => None,
309            })
310            .unwrap_or(&self.subscribed_recipients);
311        let mut next = current.clone();
312        next.write(value)?;
313        Ok(saving::NextList::SubscribedRecipients(next))
314    }
315
316    fn install(&mut self, next: saving::NextList) {
317        match next {
318            saving::NextList::RecipientList(list) => {
319                self.recipient_list = list;
320                self.recipient_list_written = true;
321            }
322            saving::NextList::SubscribedRecipients(store) => self.subscribed_recipients = store,
323        }
324        self.list_writes = self.list_writes.wrapping_add(1);
325    }
326
327    /// Replace Recipient_List or Subscribed_Recipients with a written list,
328    /// saving both lists before the forwarder serves it. A staged write takes
329    /// the lists already saved; any other saves now. A list that cannot be
330    /// saved is refused with DEVICE / OPERATIONAL_PROBLEM and the old list
331    /// stays.
332    fn write_list(
333        &mut self,
334        property: PropertyIdentifier,
335        value: PropertyValue,
336    ) -> Result<(), Error> {
337        let refused = |_| {
338            common::protocol_error(
339                bacnet_types::enums::ErrorClass::DEVICE,
340                bacnet_types::enums::ErrorCode::OPERATIONAL_PROBLEM,
341            )
342        };
343        if let Some(taken) = self.take_staged(property, &value) {
344            return taken.map_err(refused);
345        }
346        // A list the forwarder refuses leaves a write staged for another
347        // request alone (#1424).
348        let next = self.next_list(&[], property, value)?;
349        if self.storage.is_some() {
350            let mut snapshot = self.snapshot();
351            saving::apply(&mut snapshot, &next);
352            // This write will be made, so it supersedes a staged one: storage
353            // goes back to the served lists ahead of its save.
354            self.with_storage(|storage, _| storage.drop_staged());
355            let storage = self.storage.as_mut().expect("checked above");
356            storage.save_now(snapshot).map_err(refused)?;
357        }
358        self.install(next);
359        Ok(())
360    }
361
362    /// Take the staged step for a write of `value` to `property`, if it is
363    /// the next one: the forwarder then serves its list, or the save's error
364    /// comes back. `None` when nothing staged is this write.
365    fn take_staged(
366        &mut self,
367        property: PropertyIdentifier,
368        value: &PropertyValue,
369    ) -> Option<Result<(), Error>> {
370        let mut storage = self.storage.take()?;
371        let taken = storage
372            .claim(property, value, self.list_writes)
373            .map(|claimed| claimed.map(|next| self.install(next)));
374        // Steps the request has still to take keep the lists served now.
375        storage.correct(|| self.snapshot());
376        self.storage = Some(storage);
377        taken
378    }
379
380    /// After time passes, keep storage current (see the module docs).
381    fn after_time_passed(&mut self, lapsed: bool) -> bool {
382        let now = self.subscribed_recipients.current_time();
383        self.with_storage(|storage, served| storage.keep_current(now, lapsed, served));
384        lapsed
385    }
386
387    /// Run `f` on storage, handing it the lists the forwarder serves; then,
388    /// if `f` dropped a staged write, queue the save that puts storage back
389    /// to the served lists. `None` without persistence.
390    fn with_storage<R>(
391        &mut self,
392        f: impl FnOnce(&mut saving::Storage, &dyn Fn() -> ForwarderSnapshot) -> R,
393    ) -> Option<R> {
394        let mut storage = self.storage.take()?;
395        let served = || self.snapshot();
396        let result = f(&mut storage, &served);
397        storage.correct(served);
398        self.storage = Some(storage);
399        Some(result)
400    }
401}
402
403/// Whether `property` is one of the two lists a forwarder saves.
404fn listed(property: PropertyIdentifier) -> bool {
405    matches!(
406        property,
407        PropertyIdentifier::RECIPIENT_LIST | PropertyIdentifier::SUBSCRIBED_RECIPIENTS
408    )
409}
410
411impl DurableWrites for NotificationForwarderObject {
412    fn stage_write(
413        &mut self,
414        property: PropertyIdentifier,
415        array_index: Option<u32>,
416        value: &PropertyValue,
417    ) -> StageStep {
418        self.stage_writes(&[PendingWrite {
419            property,
420            array_index,
421            value: value.clone(),
422        }])
423    }
424
425    /// Fold the request's writes to both lists into one staged save
426    /// (#1423).
427    fn stage_writes(&mut self, writes: &[PendingWrite]) -> StageStep {
428        if self.storage.is_none() || !writes.iter().any(|write| listed(write.property)) {
429            return StageStep::Skip;
430        }
431        if let Some(wait) = self.with_storage(|storage, _| storage.busy()).flatten() {
432            return StageStep::Busy(wait);
433        }
434        let steps = staged::steps(writes, |earlier, write| {
435            if !listed(write.property) {
436                return Ok(None);
437            }
438            if write.array_index.is_some() {
439                return Err(common::property_is_not_an_array_error());
440            }
441            self.next_list(earlier, write.property, write.value.clone())
442                .map(Some)
443        });
444        let served = self.snapshot();
445        let base = self.list_writes;
446        let storage = self.storage.as_mut().expect("checked above");
447        storage.stage(steps, base, served)
448    }
449
450    fn release_staged_write(&mut self, staged: &SaveWait) {
451        self.with_storage(|storage, _| storage.release(staged));
452    }
453
454    fn has_staged_write(&self) -> bool {
455        self.storage
456            .as_ref()
457            .is_some_and(saving::Storage::is_staged)
458    }
459
460    fn settle_forgotten_writes(&mut self) -> Option<SaveWait> {
461        self.with_storage(|storage, served| storage.drop_forgotten(served))
462    }
463}
464
465/// A list of subscriptions in the framed form the store writes. Fails for a
466/// recipient MAC past `BACnetAddress::MAX_MAC_LEN`, which a loaded list cannot
467/// hold since its decoder refuses one.
468fn framed(subscriptions: &[BACnetEventNotificationSubscription]) -> Result<PropertyValue, Error> {
469    let mut buf = BytesMut::new();
470    bacnet_encoding::constructed::encode_event_notification_subscription_list(
471        &mut buf,
472        subscriptions,
473    )?;
474    Ok(PropertyValue::ApplicationData(buf.to_vec()))
475}
476
477fn read_bool(value: &PropertyValue) -> Result<bool, Error> {
478    match value {
479        PropertyValue::Boolean(value) => Ok(*value),
480        _ => Err(common::invalid_data_type_error()),
481    }
482}
483
484impl BACnetObject for NotificationForwarderObject {
485    fn object_identifier(&self) -> ObjectIdentifier {
486        self.oid
487    }
488
489    fn object_name(&self) -> &str {
490        &self.name
491    }
492
493    fn read_property(
494        &self,
495        property: PropertyIdentifier,
496        array_index: Option<u32>,
497    ) -> Result<PropertyValue, Error> {
498        if let Some(result) = read_common_properties!(self, property, array_index) {
499            return result;
500        }
501        match property {
502            PropertyIdentifier::OBJECT_TYPE => Ok(PropertyValue::Enumerated(
503                ObjectType::NOTIFICATION_FORWARDER.to_raw(),
504            )),
505            PropertyIdentifier::RECIPIENT_LIST => {
506                let mut buf = BytesMut::new();
507                bacnet_encoding::constructed::encode_destination_list(
508                    &mut buf,
509                    &self.recipient_list,
510                )?;
511                Ok(PropertyValue::ApplicationData(buf.to_vec()))
512            }
513            PropertyIdentifier::SUBSCRIBED_RECIPIENTS => Ok(self.subscribed_recipients.read()),
514            PropertyIdentifier::PROCESS_IDENTIFIER_FILTER => Ok(self
515                .process_identifier_filter
516                .map_or(PropertyValue::Null, |id| PropertyValue::Unsigned(id.into()))),
517            PropertyIdentifier::LOCAL_FORWARDING_ONLY => {
518                Ok(PropertyValue::Boolean(self.local_forwarding_only))
519            }
520            PropertyIdentifier::PORT_FILTER => match &self.port_filter {
521                Some(ports) => port_filter::read(ports, array_index),
522                None => Err(common::unknown_property_error()),
523            },
524            _ => Err(common::unknown_property_error()),
525        }
526    }
527
528    fn write_property(
529        &mut self,
530        property: PropertyIdentifier,
531        array_index: Option<u32>,
532        value: PropertyValue,
533        _priority: Option<u8>,
534    ) -> Result<(), Error> {
535        if listed(property) && array_index.is_some() {
536            return Err(common::property_is_not_an_array_error());
537        }
538        match property {
539            PropertyIdentifier::RECIPIENT_LIST | PropertyIdentifier::SUBSCRIBED_RECIPIENTS => {
540                return self.write_list(property, value);
541            }
542            PropertyIdentifier::PROCESS_IDENTIFIER_FILTER if array_index.is_none() => {
543                self.process_identifier_filter = match value {
544                    PropertyValue::Null => None,
545                    PropertyValue::Unsigned(id) => {
546                        Some(u32::try_from(id).map_err(|_| common::value_out_of_range_error())?)
547                    }
548                    _ => return Err(common::invalid_data_type_error()),
549                };
550                return Ok(());
551            }
552            PropertyIdentifier::LOCAL_FORWARDING_ONLY if array_index.is_none() => {
553                self.local_forwarding_only = read_bool(&value)?;
554                return Ok(());
555            }
556            PropertyIdentifier::PORT_FILTER => {
557                if let Some(ports) = &mut self.port_filter {
558                    return port_filter::write(ports, array_index, value);
559                }
560            }
561            _ => {}
562        }
563        if array_index.is_none() {
564            if let Some(result) =
565                common::write_out_of_service(&mut self.out_of_service, property, &value)
566            {
567                return result;
568            }
569            if let Some(result) = common::write_description(&mut self.description, property, &value)
570            {
571                return result;
572            }
573        }
574        Err(common::unhandled_write_error(
575            self.property_metadata().as_ref(),
576            property,
577            array_index,
578        ))
579    }
580
581    fn property_metadata(&self) -> Cow<'_, [crate::property_metadata::PropertyMetadata]> {
582        metadata::for_object(self)
583    }
584
585    fn property_list(&self) -> Cow<'static, [PropertyIdentifier]> {
586        crate::property_metadata::property_list_from_metadata(self.property_metadata().as_ref())
587    }
588
589    fn advance_time_internal(&mut self, elapsed: Duration) -> bool {
590        let lapsed = self.subscribed_recipients.advance_by(elapsed);
591        self.after_time_passed(lapsed)
592    }
593
594    fn bind_monotonic_clock_internal(&mut self, clock: Option<Arc<MonotonicClock>>) {
595        self.subscribed_recipients.bind_monotonic_clock(clock);
596        self.with_storage(|storage, _| storage.clock_changed());
597    }
598
599    fn advance_monotonic_time_internal(&mut self, now: Duration) -> bool {
600        let lapsed = self.subscribed_recipients.advance_to(now);
601        self.after_time_passed(lapsed)
602    }
603
604    fn next_monotonic_deadline_internal(&self) -> Option<Duration> {
605        self.subscribed_recipients.next_deadline()
606    }
607
608    fn durable_writes_internal(&mut self) -> Option<&mut dyn DurableWrites> {
609        Some(self)
610    }
611}
612
613#[cfg(test)]
614mod tests;
bacnet_objects/notification_forwarder/mod.rs

bacnet_objects/notification_forwarder/
mod.rs
