bacnet_client/client/
mod.rs

1//! BACnetClient: high-level and low-level request APIs.
2//!
3//! The client owns a NetworkLayer, spawns an APDU dispatch task, and provides
4//! methods for sending confirmed and unconfirmed BACnet requests.
5//! Replies to inbound accepted-direct confirmed requests use the original sealed
6//! response capability before any prompt channel, without address fallback on
7//! retirement or invalid authority. This does not change outgoing transaction
8//! correlation, retries or their segmented controls.
9
10use std::collections::HashMap;
11use std::net::Ipv4Addr;
12#[cfg(feature = "ipv6")]
13use std::net::Ipv6Addr;
14use std::sync::Arc;
15#[cfg(test)]
16use std::time::Instant;
17
18use bytes::{Bytes, BytesMut};
19use tokio::sync::{Mutex, broadcast, mpsc, oneshot};
20use tokio::task::JoinHandle;
21use tokio::time::Duration;
22#[cfg(test)]
23use tokio::time::timeout;
24use tracing::{debug, warn};
25
26use bacnet_encoding::apdu::{
27    self, AbortPdu, Apdu, ConfirmedRequest as ConfirmedRequestPdu, RejectPdu,
28    SegmentAck as SegmentAckPdu, SimpleAck, encode_apdu, validate_max_apdu_length,
29    validate_max_segments,
30};
31use bacnet_encoding::npdu::NpduAddress;
32#[cfg(test)]
33use bacnet_encoding::npdu::{Npdu, encode_npdu};
34use bacnet_endpoint_core::coordinator::{CanonicalPeer, OutboundTransactionCoordinator};
35use bacnet_network::layer::NetworkLayer;
36use bacnet_npdu::NetworkPriority;
37use bacnet_services::cov::COVNotificationRequest;
38use bacnet_transport::any::AnyTransport;
39use bacnet_transport::bip::{BipMode, BipMulticast, BipTransport};
40#[cfg(feature = "ipv6")]
41use bacnet_transport::bip6::{Bip6ForeignDeviceConfig, Bip6Transport};
42use bacnet_transport::port::{TransportPort, TransportProvenance};
43use bacnet_transport::warn_throttle::WarnThrottle;
44use bacnet_transport::warn_throttled;
45use bacnet_types::MacAddr;
46use bacnet_types::enums::{ConfirmedServiceChoice, RejectReason, UnconfirmedServiceChoice};
47use bacnet_types::error::{Error, RemoteError};
48
49use crate::discovery::{DeviceTable, DeviceUpsertResult, DiscoveredDevice, RoutedDeviceConfig};
50use crate::segmentation::{
51    SegmentReceiver, SegmentedPduType, duplicate_in_window, max_segment_payload, split_payload,
52};
53use crate::tsm::{
54    ProgressReceiver, RequestTimerExpiration, SegmentAckPhase, SegmentTimerExpiration,
55    SegmentedResponseAdmission, TransactionOwner, TransactionProgress, Tsm, TsmConfig, TsmResponse,
56};
57#[cfg(test)]
58use transaction_cleanup::{SegmentedCleanupHook, SegmentedPostWaitCleanupHook};
59use transaction_cleanup::{TransactionCleanup, TransactionGuard};
60
61mod sent_services;
62pub use sent_services::SENT_SERVICES;
63
64/// Default COV notification broadcast channel capacity.
65pub const DEFAULT_COV_CHANNEL_CAPACITY: usize = 64;
66
67/// Maximum COV notification broadcast channel capacity accepted at startup.
68pub const MAX_COV_CHANNEL_CAPACITY: usize = 65_536;
69
70/// Device discovery event broadcast channel capacity.
71pub const DEVICE_EVENT_CHANNEL_CAPACITY: usize = 64;
72
73/// Type of change observed in the device discovery table.
74#[derive(Debug, Clone, Copy, PartialEq, Eq)]
75pub enum DeviceEventKind {
76    /// A new device instance was inserted into the discovery table.
77    Discovered,
78    /// An existing device instance was refreshed by a later I-Am.
79    Updated,
80    /// A previously discovered device was purged as stale.
81    Lost,
82}
83
84/// Notification emitted when the client observes a device discovery change.
85#[derive(Debug, Clone)]
86pub struct DeviceEvent {
87    /// The kind of discovery-table change.
88    pub kind: DeviceEventKind,
89    /// Snapshot of the device entry after discovery/update or before removal.
90    pub device: DiscoveredDevice,
91}
92
93/// Notification emitted when two endpoints claim the same Device instance.
94#[derive(Debug, Clone)]
95pub struct DeviceCollisionEvent {
96    /// Snapshot of the authoritative discovery-table row that was retained.
97    pub retained: DiscoveredDevice,
98    /// Snapshot constructed from the conflicting incoming I-Am.
99    pub incoming: DiscoveredDevice,
100}
101
102/// Client configuration.
103#[derive(Debug, Clone)]
104pub struct ClientConfig {
105    /// Local interface to bind.
106    pub interface: Ipv4Addr,
107    /// UDP port (0 for ephemeral).
108    pub port: u16,
109    /// Directed broadcast address.
110    pub broadcast_address: Ipv4Addr,
111    /// B/IP only: bind the interface address itself, so transports on other
112    /// addresses of this host can share the port (#1538). Off by default.
113    /// Needs an explicit interface and a nonzero port; see
114    /// `BipTransport::set_share_port_by_address` for what changes. Only the
115    /// B/IP builder reads it; a transport passed to `start` has its own.
116    pub share_port_by_address: bool,
117    /// APDU timeout in milliseconds.
118    pub apdu_timeout_ms: u64,
119    /// Number of APDU retries.
120    pub apdu_retries: u8,
121    /// Maximum APDU length this client accepts. Each confirmed request
122    /// advertises the largest valid bucket at or below both this and the
123    /// transport's limit at that moment, so it follows a limit that changes
124    /// after start (#2086).
125    pub max_apdu_length: u16,
126    /// Maximum segments this client accepts (`None` = unspecified).
127    ///
128    /// Values 0 and 1 are invalid. Finite values between the BACnet wire
129    /// encodings are conservatively rounded down (for example, 5 advertises 4).
130    pub max_segments: Option<u8>,
131    /// Whether this client accepts segmented responses.
132    pub segmented_response_accepted: bool,
133    /// Proposed window size for segmented transfers (1-127, default 1).
134    pub proposed_window_size: u8,
135    /// Least time, in milliseconds, from the latest confirmed request sent
136    /// to a destination finishing (reply, error or cancellation) to the next
137    /// being sent; while it is still outstanding, the next waits that long
138    /// after it was sent. A waiting request checks again when it wakes;
139    /// requests waiting together go one at a time, in no promised order. A
140    /// request given up before it went leaves no trace. 0 (the default)
141    /// sends at once. A slow
142    /// device then serves its other clients between this client's requests,
143    /// paging a log or polling alike (#1535). Behind a router, whose path
144    /// lease every device on that network shares, the pause after a reply
145    /// holds for requests made one after another only. At most
146    /// [`MAX_MIN_REQUEST_INTERVAL_MS`], an hour; more fails the build. The
147    /// endpoint client takes the same setting from its session (#1542).
148    pub min_request_interval_ms: u64,
149}
150
151/// Additional client startup options.
152#[derive(Clone)]
153pub struct ClientOptions {
154    /// Capacity of the COV notification broadcast channel.
155    ///
156    /// Slow receivers lag once more than this many notifications arrive before
157    /// they call `recv()`. The default preserves the historical fixed capacity
158    /// of 64.
159    pub cov_channel_capacity: usize,
160    /// Capacity of the independent event notification broadcast channel (default 64).
161    /// Slow receivers observe lag; valid confirmed notifications are still acknowledged.
162    pub event_channel_capacity: usize,
163    confirmed_cov_notification_ack_policy: ConfirmedCOVNotificationAckPolicy,
164}
165
166impl Default for ClientConfig {
167    fn default() -> Self {
168        Self {
169            interface: Ipv4Addr::UNSPECIFIED,
170            port: 0xBAC0,
171            broadcast_address: Ipv4Addr::BROADCAST,
172            share_port_by_address: false,
173            apdu_timeout_ms: 6000,
174            apdu_retries: 3,
175            max_apdu_length: 1476,
176            max_segments: None,
177            segmented_response_accepted: true,
178            proposed_window_size: 1,
179            min_request_interval_ms: 0,
180        }
181    }
182}
183
184impl Default for ClientOptions {
185    fn default() -> Self {
186        Self {
187            cov_channel_capacity: DEFAULT_COV_CHANNEL_CAPACITY,
188            event_channel_capacity: DEFAULT_EVENT_CHANNEL_CAPACITY,
189            confirmed_cov_notification_ack_policy:
190                cov_notifications::default_confirmed_cov_notification_ack_policy(),
191        }
192    }
193}
194
195pub(crate) fn new_coordinated_tsm(
196    config: &ClientConfig,
197    coordinator: Arc<OutboundTransactionCoordinator>,
198) -> Tsm {
199    Tsm::new_coordinated(
200        TsmConfig {
201            apdu_timeout_ms: config.apdu_timeout_ms,
202            apdu_segment_timeout_ms: config.apdu_timeout_ms,
203            apdu_retries: config.apdu_retries,
204        },
205        coordinator,
206    )
207}
208
209pub(crate) fn confirmed_response_result(response: TsmResponse) -> Result<Bytes, Error> {
210    match response {
211        TsmResponse::SimpleAck => Ok(Bytes::new()),
212        TsmResponse::ComplexAck { service_data } => Ok(service_data),
213        TsmResponse::Remote {
214            error: RemoteError::Error { class, code },
215            detail,
216        } => Err(Error::with_detail(class, code, detail)),
217        TsmResponse::Remote { error, .. } => Err(Error::Remote(error)),
218        TsmResponse::NetworkPathTooLong { dnet } => Err(Error::RoutedPathTooLong { dnet }),
219    }
220}
221
222impl ClientOptions {
223    /// Set the COV notification broadcast channel capacity.
224    pub fn with_cov_channel_capacity(mut self, capacity: usize) -> Self {
225        self.cov_channel_capacity = capacity;
226        self
227    }
228
229    fn validate(&self) -> Result<(), Error> {
230        if !(1..=MAX_COV_CHANNEL_CAPACITY).contains(&self.cov_channel_capacity) {
231            return Err(Error::Encoding(format!(
232                "invalid cov-channel-capacity {}; expected 1..={}",
233                self.cov_channel_capacity, MAX_COV_CHANNEL_CAPACITY
234            )));
235        }
236        if !(1..=MAX_EVENT_CHANNEL_CAPACITY).contains(&self.event_channel_capacity) {
237            return Err(Error::Encoding(format!(
238                "invalid event-channel-capacity {}; expected 1..={}",
239                self.event_channel_capacity, MAX_EVENT_CHANNEL_CAPACITY
240            )));
241        }
242        Ok(())
243    }
244}
245
246/// Generic builder for BACnetClient with a pre-built transport. The
247/// transport type defaults to [`AnyTransport`], as [`BACnetClient`]'s does.
248pub struct ClientBuilder<T: TransportPort = AnyTransport> {
249    config: ClientConfig,
250    options: ClientOptions,
251    transport: Option<T>,
252}
253
254impl<T: TransportPort + 'static> ClientBuilder<T> {
255    /// Set the pre-built transport.
256    pub fn transport(mut self, transport: T) -> Self {
257        self.transport = Some(transport);
258        self
259    }
260
261    /// Set APDU timeout in milliseconds.
262    pub fn apdu_timeout_ms(mut self, ms: u64) -> Self {
263        self.config.apdu_timeout_ms = ms;
264        self
265    }
266
267    /// Set the maximum APDU length this client accepts.
268    pub fn max_apdu_length(mut self, len: u16) -> Self {
269        self.config.max_apdu_length = len;
270        self
271    }
272
273    /// Set the COV notification broadcast channel capacity.
274    pub fn cov_channel_capacity(mut self, capacity: usize) -> Self {
275        self.options.cov_channel_capacity = capacity;
276        self
277    }
278
279    /// Build and start the client.
280    pub async fn build(self) -> Result<BACnetClient<T>, Error> {
281        let transport = self
282            .transport
283            .ok_or_else(|| Error::Encoding("transport not set on ClientBuilder".into()))?;
284        BACnetClient::start_with_options(self.config, transport, self.options).await
285    }
286}
287
288/// BIP-specific builder that constructs `BipTransport` from interface/port/broadcast fields.
289pub struct BipClientBuilder {
290    config: ClientConfig,
291    options: ClientOptions,
292    mode: BipMode,
293    multicast: Option<BipMulticast>,
294}
295
296impl BipClientBuilder {
297    /// Make the client a B/IP-M node in `multicast`'s group (Annex J.8), or,
298    /// with `None` (the default), not: its broadcasts, Who-Is among them,
299    /// go to the group, and it takes broadcasts only from the group. Needs
300    /// an explicit [`interface`](Self::interface), and a nonzero
301    /// [`port`](Self::port) unless the group has its own. A multicast TTL of
302    /// 1, the default, keeps the group's traffic on the local network. See
303    /// [`BipTransport::set_multicast`]; `build()` fails on what it refuses.
304    ///
305    /// ```no_run
306    /// use bacnet_client::client::BACnetClient;
307    /// use bacnet_transport::bip::BipMulticast;
308    /// use std::net::Ipv4Addr;
309    /// # async fn example() -> Result<(), bacnet_types::error::Error> {
310    /// let group = BipMulticast::new(Ipv4Addr::new(239, 255, 66, 1))?;
311    /// let mut client = BACnetClient::bip_builder()
312    ///     .interface(Ipv4Addr::new(192, 0, 2, 20))
313    ///     .multicast(Some(group))
314    ///     .build()
315    ///     .await?;
316    /// client.stop().await?;
317    /// # Ok(()) }
318    /// ```
319    pub fn multicast(mut self, multicast: Option<BipMulticast>) -> Self {
320        self.multicast = multicast;
321        self
322    }
323
324    /// Set the B/IP mode: a normal node (the default), a foreign device or
325    /// a BBMD (#938). `build()` starts the client locally; a foreign device
326    /// doesn't wait for its BBMD to accept the registration. Poll
327    /// [`BACnetClient::foreign_registration`] for how it went: `registered`,
328    /// and the last attempt's outcome, a refusal included. Invalid TTL or
329    /// renewal settings fail `build()` before any I/O.
330    ///
331    /// ```no_run
332    /// use bacnet_client::client::BACnetClient;
333    /// use bacnet_transport::bip::{BipMode, ForeignDeviceConfig};
334    /// use std::{net::Ipv4Addr, time::Duration};
335    /// # async fn example() -> Result<(), bacnet_types::error::Error> {
336    /// let mut client = BACnetClient::bip_builder()
337    ///     .port(0)
338    ///     .mode(BipMode::Foreign(ForeignDeviceConfig {
339    ///         bbmd_ip: Ipv4Addr::new(192, 0, 2, 10), bbmd_port: 47808,
340    ///         ttl: 60, renewal_interval: Some(Duration::from_secs(20)),
341    ///     }))
342    ///     .build().await?;
343    /// // A successful local start alone does not establish registration.
344    /// let registration = client.foreign_registration()?;
345    /// let (registered, last_attempt) = (registration.registered, registration.last_outcome);
346    /// client.stop().await?;
347    /// # Ok(()) }
348    /// ```
349    pub fn mode(mut self, mode: BipMode) -> Self {
350        self.mode = mode;
351        self
352    }
353
354    /// Set the local interface IP.
355    pub fn interface(mut self, ip: Ipv4Addr) -> Self {
356        self.config.interface = ip;
357        self
358    }
359
360    /// Set the UDP port (0 for ephemeral).
361    pub fn port(mut self, port: u16) -> Self {
362        self.config.port = port;
363        self
364    }
365
366    /// Set the directed broadcast address.
367    pub fn broadcast_address(mut self, addr: Ipv4Addr) -> Self {
368        self.config.broadcast_address = addr;
369        self
370    }
371
372    /// Bind the interface address itself, so transports on other addresses of
373    /// this host can share the port, each getting only its own unicast
374    /// (#1538). Off by default, which binds `0.0.0.0`. Needs an explicit
375    /// [`interface`](Self::interface) and a nonzero port, or `build` fails.
376    /// See [`BipTransport::set_share_port_by_address`] for what changes.
377    pub fn share_port_by_address(mut self, enabled: bool) -> Self {
378        self.config.share_port_by_address = enabled;
379        self
380    }
381
382    /// Set APDU timeout in milliseconds.
383    pub fn apdu_timeout_ms(mut self, ms: u64) -> Self {
384        self.config.apdu_timeout_ms = ms;
385        self
386    }
387
388    /// Set the maximum APDU length this client accepts.
389    pub fn max_apdu_length(mut self, len: u16) -> Self {
390        self.config.max_apdu_length = len;
391        self
392    }
393
394    /// Set the COV notification broadcast channel capacity.
395    pub fn cov_channel_capacity(mut self, capacity: usize) -> Self {
396        self.options.cov_channel_capacity = capacity;
397        self
398    }
399}
400
401// ---------------------------------------------------------------------------
402// Multi-device batch operation types
403// ---------------------------------------------------------------------------
404
405/// Default concurrency limit for multi-device batch operations.
406const DEFAULT_BATCH_CONCURRENCY: usize = 32;
407
408mod batch_debug;
409
410/// A request to read a single property from a discovered device.
411#[derive(Debug, Clone)]
412pub struct DeviceReadRequest {
413    /// Device instance number (must be in the device table).
414    pub device_instance: u32,
415    /// Object to read from.
416    pub object_identifier: bacnet_types::primitives::ObjectIdentifier,
417    /// Property to read.
418    pub property_identifier: bacnet_types::enums::PropertyIdentifier,
419    /// Optional array index.
420    pub property_array_index: Option<u32>,
421}
422
423/// Result of a single-property read from a device within a batch.
424pub struct DeviceReadResult {
425    /// Zero-based position of this occurrence in the original request vector.
426    pub request_index: usize,
427    /// The device instance this result corresponds to.
428    pub device_instance: u32,
429    /// The read result (Ok = decoded ACK, Err = protocol/timeout error).
430    pub result: Result<bacnet_services::read_property::ReadPropertyACK, Error>,
431}
432
433/// A request to read multiple properties from a discovered device (RPM).
434#[derive(Debug, Clone)]
435pub struct DeviceRpmRequest {
436    /// Device instance number (must be in the device table).
437    pub device_instance: u32,
438    /// ReadAccessSpecifications to send in a single RPM.
439    pub specs: Vec<bacnet_types::constructed::ReadAccessSpecification>,
440}
441
442/// Result of an RPM to a single device within a batch.
443pub struct DeviceRpmResult {
444    /// Zero-based position of this occurrence in the original request vector.
445    pub request_index: usize,
446    /// The device instance this result corresponds to.
447    pub device_instance: u32,
448    /// The RPM result.
449    pub result: Result<bacnet_services::rpm::ReadPropertyMultipleACK, Error>,
450}
451
452/// A request to write a single property on a discovered device.
453#[derive(Clone)]
454pub struct DeviceWriteRequest {
455    /// Device instance number (must be in the device table).
456    pub device_instance: u32,
457    /// Object to write to.
458    pub object_identifier: bacnet_types::primitives::ObjectIdentifier,
459    /// Property to write.
460    pub property_identifier: bacnet_types::enums::PropertyIdentifier,
461    /// Optional array index.
462    pub property_array_index: Option<u32>,
463    /// Encoded property value bytes.
464    pub property_value: Vec<u8>,
465    /// Optional write priority (1-16).
466    pub priority: Option<u8>,
467}
468
469/// Result of a single-property write to a device within a batch.
470#[derive(Debug)]
471pub struct DeviceWriteResult {
472    /// Zero-based position of this occurrence in the original request vector.
473    pub request_index: usize,
474    /// The device instance this result corresponds to.
475    pub device_instance: u32,
476    /// The write result (Ok = success, Err = protocol/timeout error).
477    pub result: Result<(), Error>,
478}
479
480/// The receive-side promises this client puts in every confirmed request.
481///
482/// Clause 20.1.2.3 and Clause 20.1.2.4 let the responder choose a response
483/// form the requester can receive, so the same values must bound what
484/// the dispatch loop is willing to take back. Keeping the pair together stops
485/// the two halves from drifting as they are threaded through dispatch.
486#[derive(Debug, Clone, Copy)]
487struct ResponseLimits {
488    /// Clause 20.1.2.3 'segmented-response-accepted'.
489    segmented_response_accepted: bool,
490    /// Most segments this client will hold for one reassembly.
491    max_reassembly_segments: usize,
492}
493
494/// In-progress segmented receive state.
495struct SegmentedReceiveState {
496    receiver: SegmentReceiver,
497    owner: TransactionOwner,
498    /// Provenance snapshot at session open (RB-07). Compared by value on
499    /// every later segment; a conflicting context fails closed (abort).
500    /// Expires with the session; SC disconnect drops the transport queue so
501    /// no snapshot outlives its connection.
502    provenance: TransportProvenance,
503    /// Immediate MAC used to send SegmentAck/Abort PDUs.
504    reply_mac: MacAddr,
505    /// The peer's SNET/SADR when the segments arrive through a router; the
506    /// reply's DNET/DADR, or the router takes the reply for itself instead
507    /// of forwarding it (Clause 6.5.2.1).
508    reply_network: Option<NpduAddress>,
509    /// Next expected sequence number (for gap detection).
510    expected_next_seq: u8,
511    /// Last sequence number in the previously completed receive window.
512    initial_sequence_number: u8,
513    /// Last segment accepted in order.
514    last_sequence_number: u8,
515    /// Duplicates silently discarded in the current receive window.
516    duplicate_count: u8,
517    /// Window position counter for per-window SegmentAck (Clause 5.2.2).
518    window_position: u8,
519    /// Window size accepted for this receive session.
520    actual_window_size: u8,
521    /// How many distinct segments have been stored.
522    ///
523    /// Monotonic, and deliberately not derived from `expected_next_seq`, which
524    /// is a `u8` and wraps: Clause 20.1.5.4 makes sequence numbers modulo 256,
525    /// so after 256 segments the counter returns to a slot already occupied.
526    /// This is the only value that can tell 257 segments from 1.
527    accepted_segments: usize,
528}
529
530/// Key for tracking in-progress segmented receives:
531/// (correlation_mac, invoke_id, provenance).
532///
533/// Including the immutable provenance snapshot gives cross-peer isolation:
534/// the same MAC via different trust contexts never shares a reassembly
535/// session. A conflicting provenance for an otherwise identical key is a
536/// fail-closed abort, not a merge (RB-07 compat mode: no policy change).
537type SegKey = (MacAddr, u8, TransportProvenance);
538
539/// Key for routing inbound SegmentACKs to in-flight segmented sends:
540/// (correlation_mac, invoke_id).
541///
542/// Compat mode: provenance is threaded to the dispatch point but does not
543/// gate SegmentACK delivery (RB-09 consumes it later).
544type SegAckKey = (MacAddr, u8);
545
546struct SegmentAckRoute {
547    owner: TransactionOwner,
548    sender: mpsc::Sender<SegmentAckPdu>,
549}
550
551/// BACnet client with low-level and high-level request APIs.
552///
553/// `T` is the transport the client runs over. It defaults to
554/// [`AnyTransport`], the type the builders' `build_any()` methods return,
555/// so `BACnetClient` alone names a client over any built-in transport;
556/// `BACnetClient<BipTransport>` and the like name one over a concrete
557/// transport.
558pub struct BACnetClient<T: TransportPort = AnyTransport> {
559    config: ClientConfig,
560    network: Arc<NetworkLayer<T>>,
561    tsm: Arc<Mutex<Tsm>>,
562    device_table: Arc<Mutex<DeviceTable>>,
563    cov_tx: broadcast::Sender<ReceivedCOVNotification>,
564    event_tx: broadcast::Sender<ReceivedEventNotification>,
565    device_tx: broadcast::Sender<DeviceEvent>,
566    device_collision_tx: broadcast::Sender<DeviceCollisionEvent>,
567    i_have_tx: broadcast::Sender<ReceivedIHave>,
568    network_report_tx: broadcast::Sender<bacnet_network::layer::ReceivedNetworkReport>,
569    dispatch_task: Option<JoinHandle<()>>,
570    network_number_task: Option<JoinHandle<()>>,
571    /// Owner-qualified channels feeding SegmentACKs to in-flight segmented sends.
572    ///
573    /// Dispatch may hold [`Self::tsm`] while acquiring this lock so phase
574    /// validation and delivery cannot race terminal completion. No path may
575    /// acquire the locks in the opposite order.
576    seg_ack_senders: Arc<Mutex<HashMap<SegAckKey, SegmentAckRoute>>>,
577    cleanup_tx: mpsc::UnboundedSender<TransactionCleanup>,
578    #[cfg(test)]
579    segmented_post_wait_cleanup: Arc<SegmentedPostWaitCleanupHook>,
580    #[cfg(test)]
581    segmented_cleanup: Arc<SegmentedCleanupHook>,
582    /// The dispatch task's decode-warning throttle, for tests to read.
583    #[cfg(test)]
584    decode_warning: Arc<WarnThrottle>,
585    /// The dispatch task's response-warning throttle, for tests to read.
586    #[cfg(test)]
587    response_warning: Arc<WarnThrottle>,
588    local_mac: MacAddr,
589    routed_path_limits: Arc<RoutedPathLimits>,
590    /// See [`Self::group_source_request_drops`].
591    group_source_request_drops: Arc<std::sync::atomic::AtomicU64>,
592    pacer: Arc<pacing::RequestPacer>,
593}
594
595impl BACnetClient<BipTransport> {
596    /// Create a BIP-specific builder with interface/port/broadcast fields.
597    pub fn bip_builder() -> BipClientBuilder {
598        BipClientBuilder {
599            config: ClientConfig::default(),
600            options: ClientOptions::default(),
601            mode: BipMode::Normal,
602            multicast: None,
603        }
604    }
605}
606
607#[cfg(feature = "sc-tls")]
608impl BACnetClient<bacnet_transport::sc::ScTransport<bacnet_transport::sc_tls::TlsWebSocket>> {
609    /// Create an SC-specific builder that connects to a BACnet/SC hub.
610    #[cfg_attr(docsrs, doc(cfg(feature = "sc-tls")))]
611    pub fn sc_builder() -> ScClientBuilder {
612        ScClientBuilder {
613            config: ClientConfig::default(),
614            options: ClientOptions::default(),
615            hub_url: String::new(),
616            failover_hub_url: None,
617            tls_config: None,
618            vmac: [0; 6],
619            device_uuid: [0; 16],
620            heartbeat_interval_ms: 30_000,
621            heartbeat_timeout_ms: 60_000,
622            connect_timeout_ms: 10_000,
623            reconnect: None,
624            npdu_admission_policy: bacnet_transport::sc::ScNpduAdmissionPolicy::default(),
625        }
626    }
627}
628
629/// SC-specific client builder.
630///
631/// Created by [`BACnetClient::sc_builder()`].  Requires the `sc-tls` feature.
632#[cfg(feature = "sc-tls")]
633#[cfg_attr(docsrs, doc(cfg(feature = "sc-tls")))]
634pub struct ScClientBuilder {
635    config: ClientConfig,
636    options: ClientOptions,
637    hub_url: String,
638    failover_hub_url: Option<String>,
639    tls_config: Option<bacnet_transport::sc_tls::ScNodeTlsConfig>,
640    vmac: bacnet_transport::sc_frame::Vmac,
641    device_uuid: [u8; 16],
642    heartbeat_interval_ms: u64,
643    heartbeat_timeout_ms: u64,
644    connect_timeout_ms: u64,
645    reconnect: Option<bacnet_transport::sc::ScReconnectConfig>,
646    npdu_admission_policy: bacnet_transport::sc::ScNpduAdmissionPolicy,
647}
648
649#[cfg(feature = "sc-tls")]
650impl ScClientBuilder {
651    /// Set the hub WebSocket URL (e.g. `wss://hub.example.com/bacnet`).
652    pub fn hub_url(mut self, url: &str) -> Self {
653        self.hub_url = url.to_string();
654        self
655    }
656
657    /// Set the validated local node TLS policy, shared across initial and
658    /// reconnect attempts (including normal TLS resumption).
659    ///
660    /// ```
661    /// use bacnet_client::client::{BACnetClient, ScClientBuilder};
662    /// use bacnet_transport::sc_tls::ScNodeTlsConfig;
663    /// fn configured(tls: ScNodeTlsConfig) -> ScClientBuilder {
664    ///     BACnetClient::sc_builder().tls_config(tls)
665    /// }
666    /// ```
667    ///
668    /// ```compile_fail,E0308
669    /// use bacnet_client::client::BACnetClient;
670    /// fn raw(config: std::sync::Arc<tokio_rustls::rustls::ClientConfig>) {
671    ///     let _ = BACnetClient::sc_builder().tls_config(config);
672    /// }
673    /// ```
674    pub fn tls_config(mut self, config: bacnet_transport::sc_tls::ScNodeTlsConfig) -> Self {
675        self.tls_config = Some(config);
676        self
677    }
678
679    /// Set the local VMAC address.
680    pub fn vmac(mut self, vmac: [u8; 6]) -> Self {
681        self.vmac = vmac;
682        self
683    }
684
685    /// Set the persistent BACnet/SC device UUID.
686    pub fn device_uuid(mut self, uuid: [u8; 16]) -> Self {
687        self.device_uuid = uuid;
688        self
689    }
690
691    /// Set the APDU timeout in milliseconds.
692    pub fn apdu_timeout_ms(mut self, ms: u64) -> Self {
693        self.config.apdu_timeout_ms = ms;
694        self
695    }
696
697    /// Set the COV notification broadcast channel capacity.
698    pub fn cov_channel_capacity(mut self, capacity: usize) -> Self {
699        self.options.cov_channel_capacity = capacity;
700        self
701    }
702
703    /// Set the heartbeat interval in milliseconds (default 30 000).
704    pub fn heartbeat_interval_ms(mut self, ms: u64) -> Self {
705        self.heartbeat_interval_ms = ms;
706        self
707    }
708
709    /// Set the heartbeat timeout in milliseconds (default 60 000).
710    pub fn heartbeat_timeout_ms(mut self, ms: u64) -> Self {
711        self.heartbeat_timeout_ms = ms;
712        self
713    }
714
715    /// Set the connect timeout in milliseconds (default 10 000). It bounds
716    /// each hub dial (host lookup, TCP, TLS and WebSocket upgrade), including the ones
717    /// `build()` makes, and each Connect-Request handshake.
718    pub fn connect_timeout_ms(mut self, ms: u64) -> Self {
719        self.connect_timeout_ms = ms;
720        self
721    }
722
723    /// Set the maximum queued, unconsumed NPDUs per ingress path and source.
724    ///
725    /// Defaults to 32. The accepted range is 1 through 256; checked before
726    /// TLS lookup or dialing. The aggregate queue remains fixed at 256.
727    pub fn npdu_per_origin_limit(mut self, limit: usize) -> Self {
728        self.npdu_admission_policy.per_origin_limit = limit;
729        self
730    }
731
732    /// Set the per-source limit on inbound network-layer messages (default
733    /// [`ScControlRateLimit::default`](bacnet_transport::sc::ScControlRateLimit::default),
734    /// #518, #1969); see
735    /// [`ScTransport::with_control_rate_limit`](bacnet_transport::sc::ScTransport::with_control_rate_limit).
736    /// Checked with the NPDU quota before TLS lookup or dialing.
737    pub fn control_rate_limit(mut self, limit: bacnet_transport::sc::ScControlRateLimit) -> Self {
738        self.npdu_admission_policy.control_rate = limit;
739        self
740    }
741
742    /// Set the failover hub WebSocket URL, dialled with the same TLS policy
743    /// when the primary hub cannot be reached (Annex AB.5.2).
744    ///
745    /// Requires [`reconnect`](Self::reconnect): building with a failover
746    /// hub and no reconnect configuration fails before any dial. Both URLs
747    /// are checked (scheme, host) before dialling.
748    ///
749    /// [`build`](Self::build) starts on the failover hub when the primary
750    /// cannot be dialled, and the connector also turns to it when the primary
751    /// refuses the first Connect-Request or the reconnect attempts to the
752    /// primary are used up. While on the failover hub it keeps trying the
753    /// primary and returns to it when it can. Follow it through
754    /// `connection_events()` and `connection_status()` on
755    /// [`BACnetClient::transport`].
756    pub fn failover_hub_url(mut self, url: &str) -> Self {
757        self.failover_hub_url = Some(url.to_string());
758        self
759    }
760
761    /// Enable automatic reconnection with the given configuration.
762    ///
763    /// Reconnects redial the hub that was in use, with exponential backoff
764    /// from `initial_delay_ms` up to `max_delay_ms`, for `max_retries`
765    /// attempts, before turning to the failover hub, if one is set. Required
766    /// by [`failover_hub_url`](Self::failover_hub_url).
767    pub fn reconnect(mut self, config: bacnet_transport::sc::ScReconnectConfig) -> Self {
768        self.reconnect = Some(config);
769        self
770    }
771
772    fn validate_identity(&self) -> Result<(), Error> {
773        match self.vmac {
774            bacnet_transport::sc_frame::UNKNOWN_VMAC => Err(Error::Encoding(
775                "SC client builder: vmac must not be the reserved unknown VMAC".into(),
776            )),
777            bacnet_transport::sc_frame::BROADCAST_VMAC => Err(Error::Encoding(
778                "SC client builder: vmac must not be the reserved broadcast VMAC".into(),
779            )),
780            _ if self.device_uuid == [0; 16] => Err(Error::Encoding(
781                "SC client builder: device_uuid is required and must not be all zero".into(),
782            )),
783            _ => Ok(()),
784        }
785    }
786
787    fn sc_transport<W: bacnet_transport::sc::WebSocketPort>(
788        &self,
789        ws: W,
790    ) -> bacnet_transport::sc::ScTransport<W> {
791        bacnet_transport::sc::ScTransport::new(ws, self.vmac)
792            .with_device_uuid(self.device_uuid)
793            .with_npdu_per_origin_limit(self.npdu_admission_policy.per_origin_limit)
794            .with_control_rate_limit(self.npdu_admission_policy.control_rate)
795            .with_heartbeat_interval_ms(self.heartbeat_interval_ms)
796            .with_heartbeat_timeout_ms(self.heartbeat_timeout_ms)
797            .with_connect_timeout_ms(self.connect_timeout_ms)
798    }
799
800    #[cfg(test)]
801    pub(crate) async fn build_with_websocket_for_test<
802        W: bacnet_transport::sc::WebSocketPort + 'static,
803    >(
804        self,
805        ws: W,
806    ) -> Result<BACnetClient<bacnet_transport::sc::ScTransport<W>>, Error> {
807        self.validate()?;
808        let transport = self.sc_transport(ws);
809        BACnetClient::start_with_options(self.config, transport, self.options).await
810    }
811}
812
813/// Routing target for confirmed requests.
814#[derive(Clone, Copy)]
815enum ConfirmedTarget<'a> {
816    Local {
817        mac: &'a [u8],
818    },
819    Routed {
820        router_mac: &'a [u8],
821        dest_network: u16,
822        dest_mac: &'a [u8],
823    },
824}
825
826impl<'a> ConfirmedTarget<'a> {
827    fn additional_npdu_header_len(&self) -> u16 {
828        match self {
829            Self::Local { .. } => 0,
830            Self::Routed { dest_mac, .. } => u16::try_from(dest_mac.len())
831                .unwrap_or(u16::MAX)
832                .saturating_add(4),
833        }
834    }
835}
836
837fn max_apdu_bucket_at_or_below(limit: u16) -> Option<u16> {
838    bacnet_encoding::apdu::max_apdu_header_at_or_below(u32::from(limit)).ok()
839}
840
841fn cap_max_apdu_to_transport(configured: u16, transport_limit: u16) -> Result<u16, Error> {
842    let limit = configured.min(transport_limit);
843    max_apdu_bucket_at_or_below(limit).ok_or_else(|| {
844        Error::Encoding(format!(
845            "transport max-APDU-length {transport_limit} leaves no valid BACnet APDU bucket"
846        ))
847    })
848}
849
850mod audit;
851#[cfg(feature = "ipv6")]
852mod builder_bip6;
853#[cfg(feature = "ipv6")]
854pub use builder_bip6::Bip6ClientBuilder;
855mod builder_build;
856mod builder_options;
857mod cov;
858mod cov_notifications;
859mod cov_renewal;
860mod cov_target;
861mod device_events;
862mod device_mgmt;
863mod discovery;
864mod dispatch;
865mod dispatch_context;
866mod inbound_replies;
867use inbound_replies::InboundReply;
868mod event_notifications;
869mod file_list;
870mod i_have;
871mod lifecycle;
872mod network_reports;
873mod object_mgmt;
874pub(crate) mod pacing;
875mod property;
876mod read_range;
877mod requests;
878mod response_admission;
879mod routed_path_limits;
880mod segmentation;
881mod segmentation_abort;
882mod segmentation_context;
883mod segmented_request;
884mod summaries;
885mod transaction_cleanup;
886mod transaction_peer;
887mod transport_access;
888mod write_group;
889pub(crate) use routed_path_limits::check_routed_unicast;
890use routed_path_limits::{
891    RoutedPathLease, RoutedPathLimits, forwarded_npci_len, routed_path_quarantine_horizon,
892};
893pub(crate) use transaction_peer::TransactionPeer;
894
895pub use cov::CovPropertySubscription;
896pub use cov_notifications::{
897    COVNotificationDelivery, ConfirmedCOVNotificationAckPolicy, ConfirmedCOVNotificationResponse,
898    ReceivedCOVNotification,
899};
900pub use cov_renewal::{
901    ManagedCOVSubscription, ManagedCOVSubscriptionEvent, ManagedCOVSubscriptionOptions,
902};
903pub use cov_target::CovTarget;
904pub use event_notifications::{
905    DEFAULT_EVENT_CHANNEL_CAPACITY, EventNotificationDelivery, MAX_EVENT_CHANNEL_CAPACITY,
906    ReceivedEventNotification,
907};
908pub use i_have::{I_HAVE_CHANNEL_CAPACITY, ReceivedIHave};
909pub use network_reports::{DiscoveredRouter, NETWORK_REPORT_CHANNEL_CAPACITY};
910pub use pacing::MAX_MIN_REQUEST_INTERVAL_MS;
911pub use write_group::WriteGroupDestination;
912
913#[cfg(test)]
914mod acknowledge_alarm_tests;
915#[cfg(test)]
916mod audit_tests;
917#[cfg(test)]
918mod batch_tests;
919#[cfg(test)]
920mod broadcast_mac_tests;
921#[cfg(test)]
922mod builder_options_tests;
923#[cfg(test)]
924mod confirmed_request_dispatch_tests;
925#[cfg(test)]
926mod coordinator_tests;
927#[cfg(test)]
928mod cov_notification_tests;
929#[cfg(test)]
930mod cov_renewal_gate_tests;
931#[cfg(test)]
932mod cov_renewal_routed_tests;
933#[cfg(test)]
934mod cov_renewal_tests;
935#[cfg(test)]
936mod cov_tests;
937#[cfg(test)]
938mod device_events_tests;
939#[cfg(test)]
940mod event_notification_tests;
941#[cfg(test)]
942pub(crate) mod fake_device;
943#[cfg(test)]
944mod group_source_request_tests;
945#[cfg(test)]
946mod list_error_tests;
947#[cfg(test)]
948mod list_validation_tests;
949#[cfg(test)]
950mod pacing_tests;
951#[cfg(test)]
952mod peer_max_apdu_tests;
953#[cfg(test)]
954mod peer_segmentation_tests;
955#[cfg(test)]
956mod rb07_provenance_tests;
957#[cfg(test)]
958mod read_range_tests;
959#[cfg(test)]
960mod request_timer_tests;
961#[cfg(test)]
962mod response_correlation_tests;
963#[cfg(test)]
964mod routed_max_apdu_tests;
965#[cfg(test)]
966mod routed_path_limit_tests;
967#[cfg(test)]
968mod routed_reply_tests;
969#[cfg(all(test, feature = "sc-tls"))]
970mod sc_builder_tests;
971#[cfg(test)]
972mod sc_max_apdu_tests;
973#[cfg(test)]
974mod segmentation_retransmit_tests;
975#[cfg(test)]
976mod segmented_receive_duplicate_tests;
977#[cfg(test)]
978mod segmented_receive_lifecycle_tests;
979#[cfg(test)]
980mod segmented_request_ordering_tests;
981#[cfg(test)]
982mod segmented_request_state_tests;
983#[cfg(test)]
984mod segmented_response_admission_tests;
985#[cfg(test)]
986mod segmented_response_capacity_tests;
987#[cfg(test)]
988mod segmented_timeout_tests;
989#[cfg(test)]
990mod sent_services_tests;
991#[cfg(test)]
992mod structured_error_tests;
993#[cfg(test)]
994mod tests;
995#[cfg(test)]
996mod transport_access_tests;
997#[cfg(test)]
998mod typed_error_wire_tests;
999#[cfg(test)]
1000mod wpm_validation_tests;
1001#[cfg(test)]
1002mod write_priority_tests;
1003
1004impl<T: TransportPort + 'static> BACnetClient<T> {
1005    /// Create a generic builder that accepts a pre-built transport, such as
1006    /// an MS/TP port. A serial port opened with `TokioSerialPort::open` (the
1007    /// `serial` feature) takes the place of the in-memory line here; give
1008    /// the [`MstpConfig`](bacnet_transport::mstp::MstpConfig) the port's
1009    /// baud rate, which its timers follow.
1010    ///
1011    /// ```
1012    /// use bacnet_client::client::BACnetClient;
1013    /// use bacnet_transport::mstp::{LoopbackSerial, MstpConfig, MstpTransport};
1014    ///
1015    /// # #[tokio::main(flavor = "current_thread")]
1016    /// # async fn main() -> Result<(), bacnet_types::error::Error> {
1017    /// let (serial, _line) = LoopbackSerial::pair();
1018    /// let config = MstpConfig {
1019    ///     this_station: 1,
1020    ///     max_master: 127,
1021    ///     baud_rate: 76_800,
1022    ///     ..MstpConfig::default()
1023    /// };
1024    /// let mut client = BACnetClient::generic_builder()
1025    ///     .transport(MstpTransport::new(serial, config))
1026    ///     .build()
1027    ///     .await?;
1028    /// client.stop().await?;
1029    /// # Ok(())
1030    /// # }
1031    /// ```
1032    pub fn generic_builder() -> ClientBuilder<T> {
1033        ClientBuilder {
1034            config: ClientConfig::default(),
1035            options: ClientOptions::default(),
1036            transport: None,
1037        }
1038    }
1039}
1040
1041#[cfg(test)]
1042mod number_tests;
1043
1044mod network_number;
bacnet_client/client/mod.rs

bacnet_client/client/
mod.rs
