bacnet_ops/engine/
read.rs

1//! The reading strategy, one confirmed request at a time per device.
2//!
3//! - Properties: one ReadPropertyMultiple, with ALL for an object or an
4//!   explicit list for the device card (Clause 15.7). When the device
5//!   refuses RPM, ReadProperty each property instead (Clause 15.5), and
6//!   remember the refusal for later reads of that device. For ALL, the
7//!   object's Property_List names the properties; it leaves out the four
8//!   every object has, which are added back (Clause 12.1.1.4.1).
9//! - Object_List: one ReadProperty of the whole array. When the device
10//!   can't return it in one answer (it would need segmentation one side
11//!   lacks, say), read element 0, which holds the length, and then each
12//!   element in turn (Clause 12.11.16).
13
14use std::sync::atomic::{AtomicBool, Ordering};
15
16use bacnet_client::client::BACnetClient;
17use bacnet_services::rpm::ReadPropertyMultipleACK;
18use bacnet_transport::port::TransportPort;
19use bacnet_types::constructed::{PropertyReference, ReadAccessSpecification};
20use bacnet_types::enums::{AbortReason, ErrorClass, ObjectType, PropertyIdentifier, RejectReason};
21use bacnet_types::error::{Error, RemoteError};
22use bacnet_types::primitives::{ObjectIdentifier, PropertyValue};
23
24use crate::format::{RawRow, RowError, decode_all};
25use crate::limits::{RequestLimits, Turn};
26
27/// How a property list was read.
28#[derive(Debug, Clone, Copy, PartialEq, Eq)]
29pub enum ReadVia {
30    /// One ReadPropertyMultiple.
31    Rpm,
32    /// ReadProperty per property, after the device refused RPM.
33    Rp,
34}
35
36/// Elements a device's Object_List may claim before the browser stops
37/// reading it; a device holds far fewer.
38pub const MAX_OBJECTS: u32 = 100_000;
39
40/// Properties read when an object's Property_List can't be: the ones the
41/// browser shows most.
42const FALLBACK_PROPERTIES: [PropertyIdentifier; 9] = [
43    PropertyIdentifier::PRESENT_VALUE,
44    PropertyIdentifier::DESCRIPTION,
45    PropertyIdentifier::STATUS_FLAGS,
46    PropertyIdentifier::EVENT_STATE,
47    PropertyIdentifier::RELIABILITY,
48    PropertyIdentifier::OUT_OF_SERVICE,
49    PropertyIdentifier::UNITS,
50    PropertyIdentifier::PRIORITY_ARRAY,
51    PropertyIdentifier::RELINQUISH_DEFAULT,
52];
53
54/// Why a read stopped.
55#[derive(Debug)]
56pub enum Halt {
57    /// The job was stopped between requests.
58    Cancelled,
59    /// The device, or the way to it, failed the request.
60    Device(Error),
61}
62
63/// What [`Reader::properties`] reads.
64pub enum Wanted<'a> {
65    /// Every property (ALL, or Property_List over ReadProperty).
66    All,
67    /// These properties.
68    List(&'a [PropertyIdentifier]),
69}
70
71/// What an RPM failure says about trying ReadProperty instead.
72#[derive(Debug, Clone, Copy, PartialEq, Eq)]
73pub enum RpmFailure {
74    /// The device doesn't take RPM: a reject (an unknown service, say) or a
75    /// services-class error. Remembered for the rest of the session.
76    Refused,
77    /// This answer was too large to send (an abort for segmentation or a
78    /// full buffer): ReadProperty for this request only.
79    TooLarge,
80    /// Anything else, a timeout included: the read fails.
81    Other,
82}
83
84/// Classify an RPM failure.
85pub fn rpm_failure(error: &Error) -> RpmFailure {
86    let too_large = [
87        AbortReason::SEGMENTATION_NOT_SUPPORTED,
88        AbortReason::BUFFER_OVERFLOW,
89        AbortReason::APDU_TOO_LONG,
90    ];
91    match error {
92        Error::Remote(RemoteError::Reject { reason })
93            if *reason == RejectReason::BUFFER_OVERFLOW =>
94        {
95            RpmFailure::TooLarge
96        }
97        Error::Remote(RemoteError::Reject { .. }) => RpmFailure::Refused,
98        Error::Remote(RemoteError::Error { class, .. }) if *class == ErrorClass::SERVICES => {
99            RpmFailure::Refused
100        }
101        Error::Remote(RemoteError::Abort { reason }) if too_large.contains(reason) => {
102            RpmFailure::TooLarge
103        }
104        _ => RpmFailure::Other,
105    }
106}
107
108/// True when `error` says the device, or the way to it, isn't answering: a
109/// timeout or a transport failure. Reading on would only time out again.
110pub fn unreachable(error: &Error) -> bool {
111    timed_out(error) || matches!(error, Error::Transport(_))
112}
113
114/// True when `error` says the request timed out: the client's own timeout,
115/// or the transaction state machine giving up on its retries. A transport
116/// failure is unreachable but not a timeout.
117pub fn timed_out(error: &Error) -> bool {
118    match error {
119        Error::Remote(RemoteError::Abort { reason }) => *reason == AbortReason::TSM_TIMEOUT,
120        Error::Timeout(_) => true,
121        _ => false,
122    }
123}
124
125/// The network `device` is on, for limits that pace networks: its routed
126/// network, or 0 for the local one. `None` when the limits don't pace
127/// networks, so the lookup is skipped, or the device isn't known.
128pub async fn network_of<T: TransportPort + 'static>(
129    client: &BACnetClient<T>,
130    limits: &RequestLimits,
131    device: u32,
132) -> Option<u16> {
133    if !limits.paces_networks() {
134        return None;
135    }
136    let found = client.get_device(device).await?;
137    Some(found.source_network.unwrap_or(0))
138}
139
140/// How an Object_List read went.
141#[derive(Debug)]
142pub struct ListRead {
143    /// The entries read, in device order.
144    pub objects: Vec<ObjectIdentifier>,
145    /// It took element-by-element reads.
146    pub indexed: bool,
147    /// An element-by-element read stopped early, and why: `objects` holds
148    /// the elements before it.
149    pub stopped: Option<String>,
150}
151
152/// Reads one device, one request at a time, each with its turn at the
153/// device's gate and the global rate cap.
154pub struct Reader<'a, T: TransportPort + 'static> {
155    /// The client the requests go through.
156    pub client: &'a BACnetClient<T>,
157    /// The device instance read, looked up in the client's discovery table.
158    pub device: u32,
159    /// The limits each request waits for.
160    pub rate: &'a RequestLimits,
161    /// Set to stop the read before its next request.
162    pub cancel: &'a AtomicBool,
163    /// False once the device has refused RPM.
164    pub rpm: bool,
165}
166
167impl<T: TransportPort + 'static> Reader<'_, T> {
168    /// Wait for this request's turn at the device, unless the job was
169    /// stopped. Hold it for the one request (see [`RequestLimits::turn`]).
170    /// A scheduling hook for the TUI's own request loops; the read methods
171    /// take their turns themselves.
172    pub async fn turn(&self) -> Result<Turn, Halt> {
173        if self.cancel.load(Ordering::Relaxed) {
174            return Err(Halt::Cancelled);
175        }
176        let network = network_of(self.client, self.rate, self.device).await;
177        let turn = self.rate.turn_on(self.device, network).await;
178        if self.cancel.load(Ordering::Relaxed) {
179            return Err(Halt::Cancelled);
180        }
181        Ok(turn)
182    }
183
184    /// Wait until a request could go at once, without taking the device's
185    /// gate or a slot, unless the job was stopped (see
186    /// [`RequestLimits::wait_free`]). Another worker can take the gate
187    /// before [`Self::turn`]; the TUI's name reads, the hook's one caller,
188    /// check again then which names are still wanted.
189    pub async fn free_turn(&self) -> Result<(), Halt> {
190        if self.cancel.load(Ordering::Relaxed) {
191            return Err(Halt::Cancelled);
192        }
193        self.rate.wait_device_free(self.device).await;
194        if self.cancel.load(Ordering::Relaxed) {
195            return Err(Halt::Cancelled);
196        }
197        Ok(())
198    }
199
200    /// The Device object.
201    pub fn device_object(&self) -> Result<ObjectIdentifier, Halt> {
202        ObjectIdentifier::new(ObjectType::DEVICE, self.device).map_err(Halt::Device)
203    }
204
205    /// ReadProperty, routed if the device is.
206    pub async fn rp(
207        &self,
208        object: ObjectIdentifier,
209        property: PropertyIdentifier,
210        index: Option<u32>,
211    ) -> Result<Vec<u8>, Halt> {
212        let _turn = self.turn().await?;
213        self.client
214            .read_property_from_device(self.device, object, property, index)
215            .await
216            .map(|ack| ack.property_value)
217            .map_err(Halt::Device)
218    }
219
220    async fn rpm_rows(
221        &self,
222        object: ObjectIdentifier,
223        wanted: &Wanted<'_>,
224    ) -> Result<Vec<RawRow>, Halt> {
225        let references = match wanted {
226            Wanted::All => vec![PropertyIdentifier::ALL],
227            Wanted::List(list) => list.to_vec(),
228        };
229        let spec = ReadAccessSpecification {
230            object_identifier: object,
231            list_of_property_references: references
232                .into_iter()
233                .map(|property_identifier| PropertyReference {
234                    property_identifier,
235                    property_array_index: None,
236                })
237                .collect(),
238        };
239        let _turn = self.turn().await?;
240        let ack = self
241            .client
242            .read_property_multiple_from_device(self.device, vec![spec])
243            .await
244            .map_err(Halt::Device)?;
245        Ok(rpm_rows(ack))
246    }
247
248    /// The properties `wanted` of `object`, by RPM or, once the device has
249    /// refused it, by ReadProperty each.
250    pub async fn properties(
251        &mut self,
252        object: ObjectIdentifier,
253        wanted: Wanted<'_>,
254    ) -> Result<(Vec<RawRow>, ReadVia), Halt> {
255        if self.rpm {
256            match self.rpm_rows(object, &wanted).await {
257                Ok(rows) => return Ok((rows, ReadVia::Rpm)),
258                Err(Halt::Device(error)) => match rpm_failure(&error) {
259                    RpmFailure::Refused => {
260                        tracing::info!(
261                            "device {} refused ReadPropertyMultiple ({error}); reading one property at a time",
262                            self.device
263                        );
264                        self.rpm = false;
265                    }
266                    RpmFailure::TooLarge => tracing::info!(
267                        "{object}: the RPM answer is too large ({error}); reading one property at a time"
268                    ),
269                    RpmFailure::Other => return Err(Halt::Device(error)),
270                },
271                Err(Halt::Cancelled) => return Err(Halt::Cancelled),
272            }
273        }
274        let properties = match wanted {
275            Wanted::All => self.property_list(object).await?,
276            Wanted::List(list) => list.to_vec(),
277        };
278        let mut rows = Vec::with_capacity(properties.len());
279        for property in properties {
280            let value = match self.rp(object, property, None).await {
281                Ok(value) => Ok(value),
282                // A device that stopped answering fails the read once.
283                Err(Halt::Device(error)) if unreachable(&error) => return Err(Halt::Device(error)),
284                Err(Halt::Device(error)) => Err(RowError::from(error)),
285                Err(Halt::Cancelled) => return Err(Halt::Cancelled),
286            };
287            rows.push((property, value));
288        }
289        Ok((rows, ReadVia::Rp))
290    }
291
292    /// The properties to read one by one for ALL.
293    async fn property_list(
294        &self,
295        object: ObjectIdentifier,
296    ) -> Result<Vec<PropertyIdentifier>, Halt> {
297        let mut list = vec![
298            PropertyIdentifier::OBJECT_IDENTIFIER,
299            PropertyIdentifier::OBJECT_NAME,
300            PropertyIdentifier::OBJECT_TYPE,
301        ];
302        let listed = match self
303            .rp(object, PropertyIdentifier::PROPERTY_LIST, None)
304            .await
305        {
306            Ok(data) => decode_all(&data)
307                .unwrap_or_default()
308                .into_iter()
309                .filter_map(|value| match value {
310                    PropertyValue::Enumerated(raw) => Some(PropertyIdentifier::from_raw(raw)),
311                    _ => None,
312                })
313                .collect(),
314            Err(Halt::Device(error)) if unreachable(&error) => return Err(Halt::Device(error)),
315            Err(Halt::Device(error)) => {
316                tracing::info!(
317                    "{object}: no Property_List ({error}); reading the usual properties"
318                );
319                FALLBACK_PROPERTIES.to_vec()
320            }
321            Err(Halt::Cancelled) => return Err(Halt::Cancelled),
322        };
323        for property in listed {
324            if !list.contains(&property) && property != PropertyIdentifier::PROPERTY_LIST {
325                list.push(property);
326            }
327        }
328        Ok(list)
329    }
330
331    /// The device's Object_List. `progress` hears `(done, total)` along an
332    /// element-by-element read.
333    pub async fn object_list(&self, mut progress: impl FnMut(u32, u32)) -> Result<ListRead, Halt> {
334        let device = self.device_object()?;
335        let list = PropertyIdentifier::OBJECT_LIST;
336        let whole = match self.rp(device, list, None).await {
337            Ok(data) => {
338                return Ok(ListRead {
339                    objects: identifiers(&data),
340                    indexed: false,
341                    stopped: None,
342                });
343            }
344            // Element 0 would only time out too.
345            Err(Halt::Device(error)) if unreachable(&error) => return Err(Halt::Device(error)),
346            Err(Halt::Device(error)) => error,
347            Err(Halt::Cancelled) => return Err(Halt::Cancelled),
348        };
349        tracing::info!(
350            "device {}: Object_List not readable whole ({whole}); reading it element by element",
351            self.device
352        );
353        let total = match self.rp(device, list, Some(0)).await {
354            Ok(data) => match decode_all(&data).as_deref() {
355                Some([PropertyValue::Unsigned(n)]) => u32::try_from(*n).unwrap_or(u32::MAX),
356                _ => return Err(Halt::Device(whole)),
357            },
358            Err(Halt::Device(_)) => return Err(Halt::Device(whole)),
359            Err(Halt::Cancelled) => return Err(Halt::Cancelled),
360        };
361        if total > MAX_OBJECTS {
362            return Err(Halt::Device(Error::Encoding(format!(
363                "Object_List claims {total} objects; the browser reads at most {MAX_OBJECTS}"
364            ))));
365        }
366        let mut objects = Vec::with_capacity(total as usize);
367        let mut stopped = None;
368        for index in 1..=total {
369            match self.rp(device, list, Some(index)).await {
370                Ok(data) => objects.extend(identifiers(&data)),
371                // Keep what was read rather than lose the whole list.
372                Err(Halt::Device(error)) => {
373                    stopped = Some(format!("element {index} of {total}: {error}"));
374                    break;
375                }
376                Err(Halt::Cancelled) => return Err(Halt::Cancelled),
377            }
378            progress(index, total);
379        }
380        Ok(ListRead {
381            objects,
382            indexed: true,
383            stopped,
384        })
385    }
386}
387
388/// The properties an RPM answer holds, each its value or the device's
389/// error for it.
390pub(crate) fn rpm_rows(ack: ReadPropertyMultipleACK) -> Vec<RawRow> {
391    ack.list_of_read_access_results
392        .into_iter()
393        .flat_map(|result| result.list_of_results)
394        .map(|element| {
395            let value = match (element.property_value, element.error) {
396                (Some(value), _) => Ok(value),
397                (None, Some((class, code))) => {
398                    Err(RowError::Remote(RemoteError::Error { class, code }))
399                }
400                (None, None) => Err(RowError::NoValue),
401            };
402            (element.property_identifier, value)
403        })
404        .collect()
405}
406
407/// The object identifiers in an application-tagged value.
408fn identifiers(data: &[u8]) -> Vec<ObjectIdentifier> {
409    decode_all(data)
410        .unwrap_or_default()
411        .into_iter()
412        .filter_map(|value| match value {
413            PropertyValue::ObjectIdentifier(object) => Some(object),
414            _ => None,
415        })
416        .collect()
417}
bacnet_ops/engine/read.rs

bacnet_ops/engine/
read.rs
