bacnet_client/client/
i_have.rs

1//! I-Have delivery (Clause 16.9, #960): every decoded I-Have goes to the
2//! subscribers of [`BACnetClient::i_have_events`], and
3//! [`BACnetClient::who_has_replies`] collects the answers to one Who-Has.
4
5use super::*;
6use bacnet_services::who_has::{IHaveRequest, WhoHasObject};
7use bacnet_services::who_is::DeviceInstanceRange;
8use tokio::sync::broadcast::error::RecvError;
9
10/// I-Have broadcast channel capacity. A subscriber this far behind loses
11/// the oldest and is told how many with `RecvError::Lagged`; the dispatch
12/// task never waits for a subscriber.
13pub const I_HAVE_CHANNEL_CAPACITY: usize = 256;
14
15/// An I-Have this client received.
16#[derive(Debug, Clone, PartialEq, Eq)]
17pub struct ReceivedIHave {
18    /// The device, the object it has, and the object's name.
19    pub i_have: IHaveRequest,
20    /// The link-layer peer it came from: the device, or the router that
21    /// passed it on.
22    pub source_mac: MacAddr,
23    /// SNET/SADR, when a router passed it on from another network.
24    pub source_network: Option<NpduAddress>,
25}
26
27impl ReceivedIHave {
28    /// Whether it answers a Who-Has for `object` in `range`: the same
29    /// identifier, or the same name, compared exactly, from a device in
30    /// the range (Clause 16.9.2).
31    pub fn answers(&self, object: &WhoHasObject, range: Option<DeviceInstanceRange>) -> bool {
32        let in_range = range
33            .is_none_or(|range| range.contains(self.i_have.device_identifier.instance_number()));
34        in_range
35            && match object {
36                WhoHasObject::Identifier(id) => self.i_have.object_identifier == *id,
37                WhoHasObject::Name(name) => self.i_have.object_name == *name,
38            }
39    }
40}
41
42/// Decode an I-Have and hand it to the subscribers. A malformed one is
43/// logged through `decode_warning` and dropped; with no subscriber nothing
44/// is decoded.
45pub(super) fn deliver(
46    i_have_tx: &broadcast::Sender<ReceivedIHave>,
47    source_mac: &[u8],
48    source_network: &Option<NpduAddress>,
49    service_request: &[u8],
50    decode_warning: &WarnThrottle,
51) {
52    if i_have_tx.receiver_count() == 0 {
53        return;
54    }
55    match IHaveRequest::decode(service_request) {
56        Ok(i_have) => {
57            let _ = i_have_tx.send(ReceivedIHave {
58                i_have,
59                source_mac: MacAddr::from_slice(source_mac),
60                source_network: source_network.clone(),
61            });
62        }
63        Err(e) => {
64            warn_throttled!(decode_warning, error = %e, "Failed to decode IHave");
65        }
66    }
67}
68
69/// Receive from `rx` until `deadline`, handing each item to `each`. A lag
70/// is logged with the count lost, and collection goes on.
71pub(super) async fn collect_until<T: Clone>(
72    rx: &mut broadcast::Receiver<T>,
73    deadline: tokio::time::Instant,
74    what: &'static str,
75    mut each: impl FnMut(T),
76) {
77    loop {
78        match tokio::time::timeout_at(deadline, rx.recv()).await {
79            Ok(Ok(item)) => each(item),
80            Ok(Err(RecvError::Lagged(missed))) => {
81                warn!(missed, what, "collector fell behind; replies were lost");
82            }
83            Ok(Err(RecvError::Closed)) | Err(_) => return,
84        }
85    }
86}
87
88impl<T: TransportPort + 'static> BACnetClient<T> {
89    /// Get a receiver for every I-Have this client receives, an answer to
90    /// its own Who-Has or not (a device may send one at any time). Each call
91    /// returns a new independent receiver. A receiver that falls
92    /// [`I_HAVE_CHANNEL_CAPACITY`] behind loses the oldest.
93    pub fn i_have_events(&self) -> broadcast::Receiver<ReceivedIHave> {
94        self.i_have_tx.subscribe()
95    }
96
97    /// Send a Who-Has as [`Self::who_has`] does, then collect the I-Haves
98    /// that answer it ([`ReceivedIHave::answers`]) for `wait`, one per
99    /// device and object, in the order they arrived. No answer is an empty
100    /// list, not an error.
101    pub async fn who_has_replies(
102        &self,
103        object: WhoHasObject,
104        range: Option<DeviceInstanceRange>,
105        wait: Duration,
106    ) -> Result<Vec<ReceivedIHave>, Error> {
107        // Subscribe first, so no answer can arrive before the receiver.
108        let mut rx = self.i_have_events();
109        self.who_has(object.clone(), range).await?;
110        let deadline = tokio::time::Instant::now() + wait;
111        let mut replies: Vec<ReceivedIHave> = Vec::new();
112        collect_until(&mut rx, deadline, "I-Have", |reply| {
113            let seen = replies.iter().any(|kept| {
114                kept.i_have.device_identifier == reply.i_have.device_identifier
115                    && kept.i_have.object_identifier == reply.i_have.object_identifier
116            });
117            if !seen && reply.answers(&object, range) {
118                replies.push(reply);
119            }
120        })
121        .await;
122        Ok(replies)
123    }
124}
125
126#[cfg(test)]
127#[path = "i_have_tests.rs"]
128mod tests;
bacnet_client/client/i_have.rs

bacnet_client/client/
i_have.rs
