bacnet_ops/watch/
sample.rs

1//! Every value a point gets, as a stream per lease: the snapshot keeps
2//! only the latest values and the change signal folds changes together,
3//! so a front end that keeps a history or a log reads it here.
4
5use std::time::SystemTime;
6
7use tokio::sync::broadcast::{self, error::RecvError, error::TryRecvError};
8use tokio::time::Instant;
9
10use crate::format::RawRow;
11
12/// How a sample's values came.
13#[derive(Debug, Clone, Copy, PartialEq, Eq)]
14pub enum Via {
15    /// A start's read: the point's first, or the read before a retried
16    /// subscription.
17    First,
18    /// A COV notification.
19    Notification,
20    /// A poll.
21    Poll,
22}
23
24/// One set of values a point got.
25#[derive(Debug, Clone, PartialEq)]
26pub struct Sample {
27    /// The properties as read or notified, or why the read gave none.
28    pub values: Result<Vec<RawRow>, String>,
29    /// How they came.
30    pub via: Via,
31    /// When they arrived.
32    pub at: Instant,
33    /// When they arrived, by the wall clock.
34    pub time: SystemTime,
35}
36
37impl Sample {
38    pub(super) fn new(values: Result<Vec<RawRow>, String>, via: Via) -> Self {
39        Self::at(values, via, Instant::now(), SystemTime::now())
40    }
41
42    /// A sample of values that arrived at `at` (`time` by the wall clock).
43    pub(super) fn at(
44        values: Result<Vec<RawRow>, String>,
45        via: Via,
46        at: Instant,
47        time: SystemTime,
48    ) -> Self {
49        Self {
50            values,
51            via,
52            at,
53            time,
54        }
55    }
56}
57
58/// A lease's samples since it asked ([`Lease::samples`](super::Lease::samples)).
59/// It holds at least [`WatchOptions::samples`](super::WatchOptions::samples); a
60/// reader that falls behind loses the oldest, counted in
61/// [`dropped`](Self::dropped).
62#[derive(Debug)]
63pub struct Samples {
64    receiver: broadcast::Receiver<Sample>,
65    dropped: u64,
66}
67
68impl Samples {
69    pub(super) fn new(receiver: broadcast::Receiver<Sample>) -> Self {
70        Self {
71            receiver,
72            dropped: 0,
73        }
74    }
75
76    /// The next sample waiting, if one is.
77    pub fn try_next(&mut self) -> Option<Sample> {
78        loop {
79            match self.receiver.try_recv() {
80                Ok(sample) => return Some(sample),
81                Err(TryRecvError::Lagged(missed)) => self.dropped += missed,
82                Err(TryRecvError::Empty | TryRecvError::Closed) => return None,
83            }
84        }
85    }
86
87    /// Wait for the next sample; `None` once the point is no longer
88    /// watched and every sample was read. Cancel safe.
89    pub async fn next(&mut self) -> Option<Sample> {
90        loop {
91            match self.receiver.recv().await {
92                Ok(sample) => return Some(sample),
93                Err(RecvError::Lagged(missed)) => self.dropped += missed,
94                Err(RecvError::Closed) => return None,
95            }
96        }
97    }
98
99    /// Samples lost because this reader fell behind.
100    pub fn dropped(&self) -> u64 {
101        self.dropped
102    }
103}
bacnet_ops/watch/sample.rs

bacnet_ops/watch/
sample.rs
