bacnet_ops/engine/
mod.rs

1//! The read operations every front end shares, over one client and one set
2//! of request limits.
3//!
4//! [`Reader`] is the reading strategy for one device: each request waits
5//! for its turn at the device's gate and the global rate cap
6//! ([`RequestLimits`]), a property list goes by ReadPropertyMultiple with a
7//! ReadProperty fallback, and the Object_List falls back to element-by-element
8//! reads. The TUI's browser, watch list and write previews use it directly,
9//! with their own stop flags.
10//!
11//! [`Engine`] is the facade the MCP server builds on: it owns the client,
12//! the limits and the per-device memory of RPM refusals, and offers the
13//! read operations as plain async calls on a device instance. A device is
14//! looked up in the client's discovery table, so routed devices work as
15//! the `*_from_device` client calls route them. Cancelling a call means
16//! dropping its future: a turn's wait is cancel-safe, and a request
17//! already sent is abandoned to the client's transaction state machine.
18
19mod bbmd;
20mod discover;
21mod log;
22mod read;
23
24use std::collections::HashSet;
25use std::sync::atomic::AtomicBool;
26use std::sync::{Arc, Mutex, PoisonError};
27
28use bacnet_client::client::BACnetClient;
29use bacnet_client::log_reader::{LogCursor, LogPage, read_log_page};
30use bacnet_services::alarm_event::GetEventInformationAck;
31use bacnet_services::alarm_summary::GetAlarmSummaryAck;
32use bacnet_services::enrollment_summary::{GetEnrollmentSummaryAck, GetEnrollmentSummaryRequest};
33use bacnet_transport::port::TransportPort;
34use bacnet_types::enums::{ObjectType, PropertyIdentifier};
35use bacnet_types::error::Error;
36use bacnet_types::primitives::ObjectIdentifier;
37
38pub use self::discover::WhoIsScope;
39pub use self::log::Naming;
40pub(crate) use self::read::rpm_rows;
41pub use self::read::{
42    Halt, ListRead, MAX_OBJECTS, ReadVia, Reader, RpmFailure, Wanted, network_of, rpm_failure,
43    timed_out, unreachable,
44};
45use crate::format::{RawRow, object_name};
46use crate::limits::{RequestLimits, Turn};
47
48/// The Device properties a device summary reads (Clause 12.11): who made
49/// it, what it runs, and what it can do. The TUI's device card shows them.
50pub const DEVICE_PROPERTIES: [PropertyIdentifier; 13] = [
51    PropertyIdentifier::OBJECT_NAME,
52    PropertyIdentifier::VENDOR_NAME,
53    PropertyIdentifier::VENDOR_IDENTIFIER,
54    PropertyIdentifier::MODEL_NAME,
55    PropertyIdentifier::FIRMWARE_REVISION,
56    PropertyIdentifier::APPLICATION_SOFTWARE_VERSION,
57    PropertyIdentifier::PROTOCOL_VERSION,
58    PropertyIdentifier::PROTOCOL_REVISION,
59    PropertyIdentifier::PROTOCOL_SERVICES_SUPPORTED,
60    PropertyIdentifier::PROTOCOL_OBJECT_TYPES_SUPPORTED,
61    PropertyIdentifier::MAX_APDU_LENGTH_ACCEPTED,
62    PropertyIdentifier::SEGMENTATION_SUPPORTED,
63    PropertyIdentifier::DATABASE_REVISION,
64];
65
66/// The stop flag of the engine's readers. Nothing sets it: an engine call
67/// is cancelled by dropping its future.
68static NEVER_STOPPED: AtomicBool = AtomicBool::new(false);
69
70/// One object's properties as read.
71#[derive(Debug, Clone, PartialEq, Eq)]
72pub struct PropertyRead {
73    /// The object read.
74    pub object: ObjectIdentifier,
75    /// Each property's raw application-tagged value, or why the device
76    /// gave none.
77    pub rows: Vec<RawRow>,
78    /// Whether the read went by ReadPropertyMultiple or ReadProperty.
79    pub via: ReadVia,
80}
81
82/// A device's [`DEVICE_PROPERTIES`] as read.
83#[derive(Debug, Clone, PartialEq, Eq)]
84pub struct DeviceSummary {
85    /// The device instance.
86    pub device: u32,
87    /// Each property's raw value, or why the device gave none.
88    pub rows: Vec<RawRow>,
89    /// Whether the read went by ReadPropertyMultiple or ReadProperty.
90    pub via: ReadVia,
91}
92
93impl DeviceSummary {
94    /// The device's Object_Name, cleaned for display
95    /// ([`clean_name`](crate::format::clean_name)); `None` when the device
96    /// gave none. Device-supplied text: a front end shows it as data.
97    pub fn name(&self) -> Option<String> {
98        object_name(&self.rows)
99    }
100}
101
102/// The read operations over one client, shared by every caller.
103///
104/// Calls to one device queue at its gate, one request outstanding at a
105/// time, and every request takes a slot under the global rate cap, so many
106/// callers can share an engine without overrunning a device.
107pub struct Engine<T: TransportPort + 'static> {
108    pub(crate) client: Arc<BACnetClient<T>>,
109    limits: Arc<RequestLimits>,
110    /// Devices that refused ReadPropertyMultiple, read by ReadProperty
111    /// from then on. The lock is held only to look up or record a device,
112    /// never across an await.
113    no_rpm: Mutex<HashSet<u32>>,
114}
115
116impl<T: TransportPort + 'static> Engine<T> {
117    /// An engine over `client`, under `limits` (shared with any other
118    /// worker on the same client).
119    pub fn new(client: Arc<BACnetClient<T>>, limits: Arc<RequestLimits>) -> Self {
120        Self {
121            client,
122            limits,
123            no_rpm: Mutex::new(HashSet::new()),
124        }
125    }
126
127    /// The limits every request waits for.
128    pub fn limits(&self) -> &RequestLimits {
129        &self.limits
130    }
131
132    /// The client, for the watch hub's subscriptions and the write desk.
133    pub fn client(&self) -> &Arc<BACnetClient<T>> {
134        &self.client
135    }
136
137    /// True once `device` has refused ReadPropertyMultiple.
138    pub fn refused_rpm(&self, device: u32) -> bool {
139        self.no_rpm().contains(&device)
140    }
141
142    fn no_rpm(&self) -> std::sync::MutexGuard<'_, HashSet<u32>> {
143        self.no_rpm.lock().unwrap_or_else(PoisonError::into_inner)
144    }
145
146    /// A reader for `device`, by RPM unless the device refused it before.
147    pub(crate) fn reader(&self, device: u32) -> Reader<'_, T> {
148        Reader {
149            client: &self.client,
150            device,
151            rate: &self.limits,
152            cancel: &NEVER_STOPPED,
153            rpm: !self.refused_rpm(device),
154        }
155    }
156
157    /// Remember a refusal `reader` met.
158    pub(crate) fn settle(&self, reader: &Reader<'_, T>) {
159        if !reader.rpm {
160            self.no_rpm().insert(reader.device);
161        }
162    }
163
164    /// The properties `wanted` of `object` on `device`: one
165    /// ReadPropertyMultiple, or ReadProperty each once the device has
166    /// refused RPM. A property the device can't give is an error row,
167    /// not a failed read.
168    pub async fn read(
169        &self,
170        device: u32,
171        object: ObjectIdentifier,
172        wanted: Wanted<'_>,
173    ) -> Result<PropertyRead, Error> {
174        let mut reader = self.reader(device);
175        let read = reader.properties(object, wanted).await;
176        self.settle(&reader);
177        let (rows, via) = read.map_err(device_error)?;
178        Ok(PropertyRead { object, rows, via })
179    }
180
181    /// One property of `object` on `device`, or one element of it
182    /// (`index` 0 is an array's length), as its raw application-tagged
183    /// value.
184    pub async fn read_property(
185        &self,
186        device: u32,
187        object: ObjectIdentifier,
188        property: PropertyIdentifier,
189        index: Option<u32>,
190    ) -> Result<Vec<u8>, Error> {
191        self.reader(device)
192            .rp(object, property, index)
193            .await
194            .map_err(device_error)
195    }
196
197    /// `device`'s Object_List: whole, or element by element when the
198    /// device can't send it in one answer (at most [`MAX_OBJECTS`]).
199    pub async fn list_objects(&self, device: u32) -> Result<ListRead, Error> {
200        self.reader(device)
201            .object_list(|_, _| {})
202            .await
203            .map_err(device_error)
204    }
205
206    /// `device`'s [`DEVICE_PROPERTIES`].
207    pub async fn describe_device(&self, device: u32) -> Result<DeviceSummary, Error> {
208        let object = ObjectIdentifier::new(ObjectType::DEVICE, device)?;
209        let PropertyRead { rows, via, .. } = self
210            .read(device, object, Wanted::List(&DEVICE_PROPERTIES))
211            .await?;
212        Ok(DeviceSummary { device, rows, via })
213    }
214
215    /// One page of at most `page_size` records of `log` (a Trend Log,
216    /// Trend Log Multiple, Event Log or Audit Log) on `device`, from
217    /// `cursor`; the page says where to read next.
218    ///
219    /// A page can take more than one request (the log's record counts,
220    /// then the ReadRange). It holds the device's gate throughout, so they
221    /// still go one at a time, and each takes a rate slot of its own.
222    pub async fn read_trend(
223        &self,
224        device: u32,
225        log: ObjectIdentifier,
226        cursor: LogCursor,
227        page_size: u16,
228    ) -> Result<LogPage, Error> {
229        let network = network_of(&self.client, &self.limits, device).await;
230        let _turn = self.limits.turn_on(device, network).await;
231        let requester = log::PagedLog {
232            client: &self.client,
233            limits: &self.limits,
234            device,
235            network,
236            first: AtomicBool::new(true),
237        };
238        // The requester's calls route by device, so the link MAC is unused.
239        read_log_page(&requester, &[], log, cursor, page_size).await
240    }
241
242    /// `device`'s turn under the limits, on its network.
243    async fn turn(&self, device: u32) -> Turn {
244        let network = network_of(&self.client, &self.limits, device).await;
245        self.limits.turn_on(device, network).await
246    }
247
248    /// One GetEventInformation answer from `device`: the objects with an
249    /// event state other than normal or with transitions not yet
250    /// acknowledged. `after` continues from the last object of the
251    /// previous answer while that answer said more events remain.
252    pub async fn alarms(
253        &self,
254        device: u32,
255        after: Option<ObjectIdentifier>,
256    ) -> Result<GetEventInformationAck, Error> {
257        let _turn = self.turn(device).await;
258        self.client
259            .get_event_information_from_device(device, after)
260            .await
261    }
262
263    /// `device`'s GetAlarmSummary answer: the objects whose alarms it
264    /// reports this way.
265    pub async fn alarm_summary(&self, device: u32) -> Result<GetAlarmSummaryAck, Error> {
266        let _turn = self.turn(device).await;
267        self.client.get_alarm_summary_from_device(device).await
268    }
269
270    /// `device`'s GetEnrollmentSummary answer for `request`'s filters.
271    pub async fn enrollment_summary(
272        &self,
273        device: u32,
274        request: &GetEnrollmentSummaryRequest,
275    ) -> Result<GetEnrollmentSummaryAck, Error> {
276        let _turn = self.turn(device).await;
277        self.client
278            .get_enrollment_summary_from_device(device, request)
279            .await
280    }
281}
282
283/// The device's error behind a halted read.
284fn device_error(halt: Halt) -> Error {
285    match halt {
286        Halt::Device(error) => error,
287        // The engine's readers share NEVER_STOPPED, which nothing sets, and
288        // a reader halts as cancelled only when its flag is set.
289        Halt::Cancelled => unreachable!("an engine read is never stopped by its flag"),
290    }
291}
292
293#[cfg(test)]
294mod tests;
bacnet_ops/engine/mod.rs

bacnet_ops/engine/
mod.rs
