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

bacnet_objects/notification_forwarder/
mod.rs
