bacnet_client/client/
cov_renewal.rs

1use super::*;
2use std::sync::Mutex as StdMutex;
3use tokio::sync::oneshot;
4use tokio::time::{sleep_until, Instant as TokioInstant};
5
6const DEFAULT_COV_RENEWAL_MARGIN: Duration = Duration::from_secs(30);
7const DEFAULT_COV_RENEWAL_EVENT_CAPACITY: usize = 16;
8
9/// Options for a managed finite COV subscription.
10#[derive(Debug, Clone)]
11#[non_exhaustive]
12pub struct ManagedCOVSubscriptionOptions {
13    /// How long before the known remaining lifetime the helper should renew.
14    ///
15    /// The helper schedules renewal at `time_remaining - renewal_margin`, using
16    /// the requested lifetime until inbound notifications provide a fresher
17    /// `time_remaining` value.
18    pub renewal_margin: Duration,
19    /// Capacity of the renewal event broadcast channel.
20    ///
21    /// Accepted range is `1..=MAX_COV_CHANNEL_CAPACITY`.
22    pub event_channel_capacity: usize,
23}
24
25impl Default for ManagedCOVSubscriptionOptions {
26    fn default() -> Self {
27        Self {
28            renewal_margin: DEFAULT_COV_RENEWAL_MARGIN,
29            event_channel_capacity: DEFAULT_COV_RENEWAL_EVENT_CAPACITY,
30        }
31    }
32}
33
34impl ManagedCOVSubscriptionOptions {
35    /// Set the renewal margin.
36    pub fn with_renewal_margin(mut self, margin: Duration) -> Self {
37        self.renewal_margin = margin;
38        self
39    }
40
41    /// Set the renewal event broadcast channel capacity.
42    pub fn with_event_channel_capacity(mut self, capacity: usize) -> Self {
43        self.event_channel_capacity = capacity;
44        self
45    }
46
47    fn validate(&self, lifetime: u32) -> Result<(), Error> {
48        if !(1..=MAX_COV_CHANNEL_CAPACITY).contains(&self.event_channel_capacity) {
49            return Err(Error::Encoding(format!(
50                "invalid managed COV subscription event channel capacity {}; expected 1..={}",
51                self.event_channel_capacity, MAX_COV_CHANNEL_CAPACITY
52            )));
53        }
54        if self.renewal_margin >= Duration::from_secs(u64::from(lifetime)) {
55            return Err(Error::Encoding(
56                "managed COV subscription renewal margin must be shorter than lifetime".into(),
57            ));
58        }
59        Ok(())
60    }
61
62    fn renewal_delay(&self, time_remaining: u32) -> Duration {
63        Duration::from_secs(u64::from(time_remaining)).saturating_sub(self.renewal_margin)
64    }
65}
66
67/// Events emitted by a managed finite COV subscription.
68#[derive(Debug, Clone, PartialEq, Eq)]
69#[non_exhaustive]
70pub enum ManagedCOVSubscriptionEvent {
71    /// A matching COV notification supplied a new lifetime observation.
72    NotificationObserved {
73        /// The `time_remaining` field from the inbound notification.
74        time_remaining: u32,
75        /// Delay until the next renewal after applying the configured margin.
76        renew_after: Duration,
77    },
78    /// The observed lifetime is inside the configured renewal margin.
79    ImpendingExpiry {
80        /// The `time_remaining` field from the inbound notification.
81        time_remaining: u32,
82    },
83    /// The helper successfully renewed the subscription.
84    Renewed {
85        /// Lifetime requested on the renewed SubscribeCOV request.
86        requested_lifetime: u32,
87        /// Delay until the next scheduled renewal after applying the margin.
88        renew_after: Duration,
89    },
90    /// A renewal SubscribeCOV request failed and the helper stopped.
91    RenewalFailed {
92        /// Display form of the protocol, reject, abort, transport, or timeout error.
93        error: String,
94    },
95}
96
97/// Handle for a managed finite COV subscription renewal task.
98pub struct ManagedCOVSubscription {
99    events_tx: broadcast::Sender<ManagedCOVSubscriptionEvent>,
100    last_event: Arc<StdMutex<Option<ManagedCOVSubscriptionEvent>>>,
101    stop_tx: Option<oneshot::Sender<()>>,
102    task: Option<JoinHandle<()>>,
103}
104
105impl ManagedCOVSubscription {
106    /// Subscribe to lifetime observations and renewal outcomes.
107    pub fn events(&self) -> broadcast::Receiver<ManagedCOVSubscriptionEvent> {
108        self.events_tx.subscribe()
109    }
110
111    /// Return the latest observed renewal event, if any.
112    pub fn last_event(&self) -> Option<ManagedCOVSubscriptionEvent> {
113        self.last_event.lock().ok().and_then(|event| event.clone())
114    }
115
116    /// Stop renewing and wait for the renewal task to exit.
117    pub async fn stop(mut self) {
118        if let Some(stop_tx) = self.stop_tx.take() {
119            let _ = stop_tx.send(());
120        }
121        if let Some(task) = self.task.take() {
122            let _ = task.await;
123        }
124    }
125
126    /// Return whether the renewal task has already finished.
127    pub fn is_finished(&self) -> bool {
128        match &self.task {
129            Some(task) => task.is_finished(),
130            None => true,
131        }
132    }
133}
134
135impl Drop for ManagedCOVSubscription {
136    fn drop(&mut self) {
137        if let Some(stop_tx) = self.stop_tx.take() {
138            let _ = stop_tx.send(());
139        }
140        if let Some(task) = self.task.take() {
141            task.abort();
142        }
143    }
144}
145
146impl<T: TransportPort + 'static> BACnetClient<T> {
147    /// Start a managed finite SubscribeCOV renewal task for a directly reachable device.
148    ///
149    /// The helper sends the initial SubscribeCOV request, then renews the same
150    /// subscription before expiry. It also listens to [`Self::cov_notifications`]
151    /// and moves the renewal earlier when matching notifications report a lower
152    /// `time_remaining` than the requested lifetime-based schedule.
153    ///
154    /// This method takes `Arc<Self>` so the renewal task can keep the client alive.
155    /// Call it as `Arc::clone(&client).manage_cov_subscription(...)`.
156    pub async fn manage_cov_subscription(
157        self: Arc<Self>,
158        destination_mac: &[u8],
159        subscriber_process_identifier: u32,
160        monitored_object_identifier: bacnet_types::primitives::ObjectIdentifier,
161        confirmed: bool,
162        lifetime: u32,
163        options: ManagedCOVSubscriptionOptions,
164    ) -> Result<ManagedCOVSubscription, Error> {
165        if lifetime == 0 {
166            return Err(Error::Encoding(
167                "managed COV subscriptions require a finite non-zero lifetime".into(),
168            ));
169        }
170        options.validate(lifetime)?;
171
172        let destination_mac = MacAddr::from_slice(destination_mac);
173        let mut notifications = self.cov_notifications();
174
175        self.subscribe_cov(
176            &destination_mac,
177            subscriber_process_identifier,
178            monitored_object_identifier,
179            confirmed,
180            Some(lifetime),
181        )
182        .await?;
183
184        let (events_tx, _) = broadcast::channel(options.event_channel_capacity);
185        let events_tx_for_task = events_tx.clone();
186        let last_event = Arc::new(StdMutex::new(None));
187        let last_event_for_task = Arc::clone(&last_event);
188        let (stop_tx, mut stop_rx) = oneshot::channel();
189        let client = Arc::clone(&self);
190        let mut renew_at = TokioInstant::now() + options.renewal_delay(lifetime);
191
192        let task = tokio::spawn(async move {
193            loop {
194                let renew_sleep = sleep_until(renew_at);
195                tokio::pin!(renew_sleep);
196                tokio::select! {
197                    _ = &mut stop_rx => break,
198                    _ = &mut renew_sleep => {
199                        let renewal = client.subscribe_cov(
200                            &destination_mac,
201                            subscriber_process_identifier,
202                            monitored_object_identifier,
203                            confirmed,
204                            Some(lifetime),
205                        );
206                        tokio::pin!(renewal);
207                        tokio::select! {
208                            _ = &mut stop_rx => break,
209                            result = &mut renewal => match result {
210                                Ok(()) => {
211                                    let renew_after = options.renewal_delay(lifetime);
212                                    publish_managed_cov_event(
213                                        &events_tx_for_task,
214                                        &last_event_for_task,
215                                        ManagedCOVSubscriptionEvent::Renewed {
216                                            requested_lifetime: lifetime,
217                                            renew_after,
218                                        },
219                                    );
220                                    renew_at = TokioInstant::now() + renew_after;
221                                }
222                                Err(error) => {
223                                    publish_managed_cov_event(
224                                        &events_tx_for_task,
225                                        &last_event_for_task,
226                                        ManagedCOVSubscriptionEvent::RenewalFailed {
227                                            error: error.to_string(),
228                                        },
229                                    );
230                                    break;
231                                }
232                            },
233                        }
234                    }
235                    received = notifications.recv() => {
236                        let received = match received {
237                            Ok(received) => received,
238                            Err(broadcast::error::RecvError::Lagged(_)) => continue,
239                            Err(broadcast::error::RecvError::Closed) => break,
240                        };
241                        if !matches_managed_cov_notification(
242                            &received,
243                            &destination_mac,
244                            subscriber_process_identifier,
245                            monitored_object_identifier,
246                        ) {
247                            continue;
248                        }
249
250                        let time_remaining = received.notification.time_remaining;
251                        let renew_after = options.renewal_delay(time_remaining);
252                        publish_managed_cov_event(
253                            &events_tx_for_task,
254                            &last_event_for_task,
255                            ManagedCOVSubscriptionEvent::NotificationObserved {
256                                time_remaining,
257                                renew_after,
258                            },
259                        );
260
261                        let observed_renew_at = TokioInstant::now() + renew_after;
262                        if renew_after.is_zero() {
263                            publish_managed_cov_event(
264                                &events_tx_for_task,
265                                &last_event_for_task,
266                                ManagedCOVSubscriptionEvent::ImpendingExpiry {
267                                    time_remaining,
268                                },
269                            );
270                            renew_at = observed_renew_at;
271                        } else if observed_renew_at < renew_at {
272                            renew_at = observed_renew_at;
273                        }
274                    }
275                }
276            }
277        });
278
279        Ok(ManagedCOVSubscription {
280            events_tx,
281            last_event,
282            stop_tx: Some(stop_tx),
283            task: Some(task),
284        })
285    }
286}
287
288fn publish_managed_cov_event(
289    events_tx: &broadcast::Sender<ManagedCOVSubscriptionEvent>,
290    last_event: &StdMutex<Option<ManagedCOVSubscriptionEvent>>,
291    event: ManagedCOVSubscriptionEvent,
292) {
293    if let Ok(mut last_event) = last_event.lock() {
294        *last_event = Some(event.clone());
295    }
296    let _ = events_tx.send(event);
297}
298
299fn matches_managed_cov_notification(
300    received: &ReceivedCOVNotification,
301    destination_mac: &[u8],
302    subscriber_process_identifier: u32,
303    monitored_object_identifier: bacnet_types::primitives::ObjectIdentifier,
304) -> bool {
305    received.notification.subscriber_process_identifier == subscriber_process_identifier
306        && received.notification.monitored_object_identifier == monitored_object_identifier
307        && received.source_network.is_none()
308        && received.source_address.is_none()
309        && &received.source_mac[..] == destination_mac
310}
bacnet_client/client/cov_renewal.rs

bacnet_client/client/
cov_renewal.rs
