bacnet_ops/watch/
poll.rs

1//! Batched polls of one device's watched points (from the TUI's watch
2//! list, #2115).
3//!
4//! The rows due on a device are read together: by ReadPropertyMultiple, up
5//! to [`MAX_BATCH`] rows a request, each row its own access specification,
6//! or by ReadProperty once the device has refused RPM (Clause 15.5). A
7//! batch whose answer would be too large is asked again a row at a time;
8//! any other batch the device fails as a whole is read by ReadProperty for
9//! this poll only, so one bad object can't put errors on its neighbours. A device
10//! that stops answering ends the poll, and the rows left get its error at
11//! once. Every request takes its own turn at the device's gate under the
12//! shared [`RequestLimits`].
13
14use std::collections::VecDeque;
15
16use bacnet_client::client::BACnetClient;
17use bacnet_transport::port::TransportPort;
18use bacnet_types::constructed::{PropertyReference, ReadAccessSpecification};
19use bacnet_types::enums::PropertyIdentifier;
20use bacnet_types::error::{Error, RemoteError};
21use bacnet_types::primitives::ObjectIdentifier;
22
23use crate::engine::{RpmFailure, network_of, rpm_failure, unreachable};
24use crate::format::{RawRow, RowError};
25use crate::limits::{RequestLimits, Turn};
26
27/// Rows read by one ReadPropertyMultiple when polling.
28pub const MAX_BATCH: usize = 16;
29
30/// Properties one ReadPropertyMultiple of a poll asks for at most, all its
31/// rows together: a batch ends at [`MAX_BATCH`] rows or here, whichever
32/// comes first, so rows that read many properties make smaller batches.
33pub const MAX_BATCH_PROPERTIES: usize = 32;
34
35/// Why a row of an answer has no reading.
36pub const MISSING: &str = "missing from the device's answer";
37
38/// One row of a poll: what it reads of which object.
39#[derive(Debug, Clone, PartialEq, Eq)]
40pub struct PollRow<K> {
41    /// The caller's name for the row.
42    pub key: K,
43    /// The object read.
44    pub object: ObjectIdentifier,
45    /// The properties a ReadPropertyMultiple asks for, each with its array
46    /// element if one; the first is the watched one.
47    pub wanted: Vec<(PropertyIdentifier, Option<u32>)>,
48    /// How many of `wanted`, from the first, are read one by one by
49    /// ReadProperty when RPM isn't used; at least one.
50    pub one_by_one: usize,
51}
52
53/// One poll of a device's rows.
54#[derive(Debug)]
55pub struct Polled<K> {
56    /// Every row asked for: its properties as read, or why it has none.
57    pub rows: Vec<(K, Result<Vec<RawRow>, String>)>,
58    /// The device refused RPM during this poll.
59    pub refused_rpm: bool,
60    /// The device stopped answering: the poll gave up on the rows left.
61    pub unreachable: bool,
62}
63
64/// `device`'s turn, on its network when the limits pace networks.
65async fn turn<T: TransportPort + 'static>(
66    client: &BACnetClient<T>,
67    limits: &RequestLimits,
68    device: u32,
69) -> Turn {
70    let network = network_of(client, limits, device).await;
71    limits.turn_on(device, network).await
72}
73
74/// Read every row in `rows` of `device`, by RPM while `rpm` holds.
75pub async fn poll<K: Copy, T: TransportPort + 'static>(
76    client: &BACnetClient<T>,
77    limits: &RequestLimits,
78    device: u32,
79    rows: &[PollRow<K>],
80    mut rpm: bool,
81) -> Polled<K> {
82    let started_rpm = rpm;
83    let mut read = Vec::with_capacity(rows.len());
84    let mut silent: Option<Error> = None;
85    let mut batches: VecDeque<&[PollRow<K>]> = batches(rows).collect();
86    while let Some(batch) = batches.pop_front() {
87        if let Some(error) = &silent {
88            read.extend(batch.iter().map(|row| (row.key, Err(error.to_string()))));
89            continue;
90        }
91        if rpm {
92            match rpm_batch(client, limits, device, batch).await {
93                Ok(rows) => {
94                    read.extend(rows);
95                    continue;
96                }
97                Err(error) if unreachable(&error) => {
98                    read.extend(batch.iter().map(|row| (row.key, Err(error.to_string()))));
99                    silent = Some(error);
100                    continue;
101                }
102                Err(error) => match rpm_failure(&error) {
103                    RpmFailure::Refused => {
104                        tracing::info!(
105                            "device {device} refused ReadPropertyMultiple ({error}); polling one property at a time"
106                        );
107                        rpm = false;
108                    }
109                    RpmFailure::TooLarge if batch.len() > 1 => {
110                        tracing::debug!(
111                            "device {device}: batch answer too large ({error}); reading it a row at a time"
112                        );
113                        for at in (0..batch.len()).rev() {
114                            batches.push_front(&batch[at..=at]);
115                        }
116                        continue;
117                    }
118                    RpmFailure::TooLarge | RpmFailure::Other => tracing::debug!(
119                        "device {device}: batch read failed ({error}); reading it one property at a time"
120                    ),
121                },
122            }
123        }
124        for row in batch {
125            if let Some(error) = &silent {
126                read.push((row.key, Err(error.to_string())));
127                continue;
128            }
129            match one_by_one(client, limits, device, row).await {
130                Ok(rows) => read.push((row.key, Ok(rows))),
131                Err(error) => {
132                    read.push((row.key, Err(error.to_string())));
133                    silent = Some(error);
134                }
135            }
136        }
137    }
138    Polled {
139        rows: read,
140        refused_rpm: started_rpm && !rpm,
141        unreachable: silent.is_some(),
142    }
143}
144
145/// `rows` in batches of at most [`MAX_BATCH`] rows and
146/// [`MAX_BATCH_PROPERTIES`] properties; a row asking for more is a batch of
147/// its own.
148fn batches<K>(rows: &[PollRow<K>]) -> impl Iterator<Item = &[PollRow<K>]> {
149    let mut start = 0;
150    std::iter::from_fn(move || {
151        if start == rows.len() {
152            return None;
153        }
154        let mut end = start;
155        let mut properties = 0;
156        while end < rows.len() && end - start < MAX_BATCH {
157            let more = rows[end].wanted.len();
158            if end > start && properties + more > MAX_BATCH_PROPERTIES {
159                break;
160            }
161            properties += more;
162            end += 1;
163        }
164        let batch = &rows[start..end];
165        start = end;
166        Some(batch)
167    })
168}
169
170/// `row`'s first [`PollRow::one_by_one`] properties by ReadProperty each;
171/// the error of a device that stopped answering.
172async fn one_by_one<K, T: TransportPort + 'static>(
173    client: &BACnetClient<T>,
174    limits: &RequestLimits,
175    device: u32,
176    row: &PollRow<K>,
177) -> Result<Vec<RawRow>, Error> {
178    let mut rows = Vec::with_capacity(row.one_by_one);
179    for &(property, index) in row.wanted.iter().take(row.one_by_one.max(1)) {
180        let turn = turn(client, limits, device).await;
181        let read = client
182            .read_property_from_device(device, row.object, property, index)
183            .await;
184        drop(turn);
185        let value = match read {
186            Ok(ack) => Ok(ack.property_value),
187            Err(error) if unreachable(&error) => return Err(error),
188            Err(error) => Err(RowError::from(error)),
189        };
190        rows.push((property, value));
191    }
192    Ok(rows)
193}
194
195/// One ReadPropertyMultiple for `batch`, one access specification per row.
196async fn rpm_batch<K: Copy, T: TransportPort + 'static>(
197    client: &BACnetClient<T>,
198    limits: &RequestLimits,
199    device: u32,
200    batch: &[PollRow<K>],
201) -> Result<Vec<(K, Result<Vec<RawRow>, String>)>, Error> {
202    let specs = batch
203        .iter()
204        .map(|row| ReadAccessSpecification {
205            object_identifier: row.object,
206            list_of_property_references: row
207                .wanted
208                .iter()
209                .map(
210                    |&(property_identifier, property_array_index)| PropertyReference {
211                        property_identifier,
212                        property_array_index,
213                    },
214                )
215                .collect(),
216        })
217        .collect();
218    let turn = turn(client, limits, device).await;
219    let ack = client
220        .read_property_multiple_from_device(device, specs)
221        .await;
222    drop(turn);
223    let ack = ack?;
224    // The answer lists the objects in the order they were asked for; a row
225    // it leaves out gets an error.
226    let mut results = ack.list_of_read_access_results.into_iter();
227    Ok(batch
228        .iter()
229        .map(|row| {
230            let Some(result) = results.next() else {
231                return (row.key, Err(MISSING.to_string()));
232            };
233            let rows: Vec<RawRow> = result
234                .list_of_results
235                .into_iter()
236                .map(|element| {
237                    let value = match (element.property_value, element.error) {
238                        (Some(value), _) => Ok(value),
239                        (None, Some((class, code))) => {
240                            Err(RowError::Remote(RemoteError::Error { class, code }))
241                        }
242                        (None, None) => Err(RowError::NoValue),
243                    };
244                    (element.property_identifier, value)
245                })
246                .collect();
247            (row.key, Ok(rows))
248        })
249        .collect())
250}
251
252#[cfg(test)]
253mod tests {
254    use bacnet_types::enums::ObjectType;
255
256    use super::*;
257
258    fn row(key: usize, properties: usize) -> PollRow<usize> {
259        PollRow {
260            key,
261            object: ObjectIdentifier::new(ObjectType::ANALOG_INPUT, 1).unwrap(),
262            wanted: vec![(PropertyIdentifier::PRESENT_VALUE, None); properties],
263            one_by_one: 1,
264        }
265    }
266
267    fn sizes(rows: &[PollRow<usize>]) -> Vec<usize> {
268        batches(rows).map(<[_]>::len).collect()
269    }
270
271    /// Two-property rows go sixteen to a batch; ten-property rows three;
272    /// a row past the property cap goes alone.
273    #[test]
274    fn batches_end_at_sixteen_rows_or_thirty_two_properties() {
275        let small: Vec<_> = (0..40).map(|k| row(k, 2)).collect();
276        assert_eq!(sizes(&small), [16, 16, 8]);
277        let points: Vec<_> = (0..7).map(|k| row(k, 10)).collect();
278        assert_eq!(sizes(&points), [3, 3, 1]);
279        let wide = [row(0, 40), row(1, 2), row(2, 40)];
280        assert_eq!(sizes(&wide), [1, 1, 1]);
281        assert!(sizes(&[]).is_empty());
282    }
283}
bacnet_ops/watch/poll.rs

bacnet_ops/watch/
poll.rs
