bacnet_objects/notification_forwarder/
persistence.rs

1//! Keeping a forwarder's Recipient_List and Subscribed_Recipients across a
2//! restart (Clauses 12.51.8 and 12.51.9).
3//!
4//! Like [`AuditLogPersistence`](crate::audit::AuditLogPersistence), the
5//! storage belongs to the application. The forwarder loads it once, when it is
6//! built, and saves both lists whenever either changes. Saves run on the
7//! forwarder's own writer thread. The bundled server waits for them with the
8//! database guard dropped; a write nobody staged waits where it is
9//! ([`crate::durable`] lists those paths). Each saved Subscribed_Recipients
10//! entry carries the whole minutes it had left at the save.
11
12use crate::prelude::*;
13#[cfg(feature = "std")]
14use std::path::Path;
15
16#[cfg(feature = "std")]
17use bacnet_encoding::constructed::{
18    decode_destination, decode_event_notification_subscription, encode_destination_list,
19    encode_event_notification_subscription_list,
20};
21use bacnet_types::constructed::{BACnetDestination, BACnetEventNotificationSubscription};
22use bacnet_types::error::Error;
23use bacnet_types::primitives::ObjectIdentifier;
24#[cfg(feature = "std")]
25use bytes::BytesMut;
26
27#[cfg(feature = "std")]
28use super::MAX_RECIPIENT_LIST_DESTINATIONS;
29#[cfg(feature = "std")]
30use crate::durable::file::ObjectFile;
31#[cfg(feature = "std")]
32use crate::subscribed_recipients::MAX_SUBSCRIBED_RECIPIENTS;
33
34/// What a forwarder keeps across a restart: both of its lists.
35#[derive(Clone, Debug, Default, PartialEq)]
36pub struct ForwarderSnapshot {
37    /// Recipient_List, in list order, once a write has set it. `None` until
38    /// then: the destinations the application configures apply at each
39    /// start.
40    pub recipient_list: Option<Vec<BACnetDestination>>,
41    /// The live Subscribed_Recipients entries, in list order, each with the
42    /// whole minutes it had left when saved.
43    pub subscribed_recipients: Vec<BACnetEventNotificationSubscription>,
44}
45
46/// Application-owned storage for one Notification Forwarder's lists.
47///
48/// The forwarder calls it from its writer thread, one call at a time. A
49/// written list lands in storage a moment before the forwarder serves it: see
50/// [storage leads the served
51/// state](crate::durable#storage-leads-the-served-state).
52pub trait NotificationForwarderPersistence: Send + Sync {
53    /// The lists last saved for `forwarder`, or `None` when nothing was
54    /// saved.
55    fn load(&self, forwarder: ObjectIdentifier) -> Result<Option<ForwarderSnapshot>, Error>;
56
57    /// Durably replace what is saved for `forwarder` with `snapshot`.
58    fn save(&self, forwarder: ObjectIdentifier, snapshot: &ForwarderSnapshot) -> Result<(), Error>;
59}
60
61#[cfg(feature = "std")]
62const MAGIC: &[u8; 8] = b"RBNFWD01";
63/// After the shared header (the magic tag and the object identifier): one
64/// octet, 1 when the file holds a Recipient_List and 0 when not, then the
65/// Recipient_List's length.
66#[cfg(feature = "std")]
67const BODY_HEADER_LEN: usize = 1 + 4;
68/// A full Recipient_List of the longest destinations is 1,504 octets and a
69/// full Subscribed_Recipients 1,184; anything much larger is not a file this
70/// backend wrote.
71#[cfg(feature = "std")]
72pub(super) const MAX_FILE_BYTES: u64 = 64 * 1024;
73
74/// A file holding one forwarder's lists, replaced whole on each save.
75///
76/// A save writes a sibling `.tmp` file, synchronizes it, renames it over the
77/// old one and then synchronizes the directory, so a failed save leaves the
78/// previous lists in place and a completed one survives a power loss. Once
79/// the rename succeeds the save has landed: a directory sync that fails is
80/// logged, not returned. (On Windows the directory is not synchronized: see
81/// the [`durable`](crate::durable) module.) The file holds a magic tag, the
82/// forwarder's object identifier, whether a Recipient_List follows, the
83/// encoded Recipient_List with its length, and the encoded
84/// Subscribed_Recipients. It does not coordinate between processes.
85#[derive(Clone, Debug)]
86#[cfg(feature = "std")]
87pub struct FileNotificationForwarderPersistence {
88    file: ObjectFile,
89}
90
91#[cfg(feature = "std")]
92impl FileNotificationForwarderPersistence {
93    /// Keep the lists in the file at `path`.
94    pub fn new(path: impl AsRef<Path>) -> Result<Self, Error> {
95        Ok(Self {
96            file: ObjectFile::new(
97                path.as_ref(),
98                MAGIC,
99                MAX_FILE_BYTES,
100                "Notification Forwarder",
101            )?,
102        })
103    }
104
105    /// The file the lists are kept in.
106    pub fn path(&self) -> &Path {
107        self.file.path()
108    }
109}
110
111#[cfg(feature = "std")]
112impl NotificationForwarderPersistence for FileNotificationForwarderPersistence {
113    fn load(&self, forwarder: ObjectIdentifier) -> Result<Option<ForwarderSnapshot>, Error> {
114        let Some(body) = self.file.load(forwarder)? else {
115            return Ok(None);
116        };
117        if body.len() < BODY_HEADER_LEN {
118            return Err(self.file.corrupt("has no valid header"));
119        }
120        let length = u32::from_be_bytes(body[1..BODY_HEADER_LEN].try_into().unwrap());
121        let split = usize::try_from(length)
122            .ok()
123            .and_then(|length| BODY_HEADER_LEN.checked_add(length))
124            .filter(|split| *split <= body.len())
125            .ok_or_else(|| self.file.corrupt("has a Recipient_List past its end"))?;
126        let recipient_list = &body[BODY_HEADER_LEN..split];
127        let recipient_list = match body[0] {
128            0 if recipient_list.is_empty() => None,
129            1 => Some(self.file.decode_capped(
130                recipient_list,
131                MAX_RECIPIENT_LIST_DESTINATIONS,
132                decode_destination,
133            )?),
134            _ => return Err(self.file.corrupt("has no valid header")),
135        };
136        Ok(Some(ForwarderSnapshot {
137            recipient_list,
138            subscribed_recipients: self.file.decode_capped(
139                &body[split..],
140                MAX_SUBSCRIBED_RECIPIENTS,
141                decode_event_notification_subscription,
142            )?,
143        }))
144    }
145
146    fn save(&self, forwarder: ObjectIdentifier, snapshot: &ForwarderSnapshot) -> Result<(), Error> {
147        let mut recipient_list = BytesMut::new();
148        if let Some(list) = &snapshot.recipient_list {
149            encode_destination_list(&mut recipient_list, list)?;
150        }
151        let mut body = BytesMut::with_capacity(BODY_HEADER_LEN + recipient_list.len() + 1024);
152        body.extend_from_slice(&[u8::from(snapshot.recipient_list.is_some())]);
153        let length = u32::try_from(recipient_list.len())
154            .map_err(|_| Error::OutOfRange("Recipient_List too long to save".into()))?;
155        body.extend_from_slice(&length.to_be_bytes());
156        body.extend_from_slice(&recipient_list);
157        encode_event_notification_subscription_list(&mut body, &snapshot.subscribed_recipients)?;
158        self.file.save(forwarder, &body)
159    }
160}
bacnet_objects/notification_forwarder/persistence.rs

bacnet_objects/notification_forwarder/
persistence.rs
