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
6use std::collections::HashMap;
7use std::net::Ipv4Addr;
8#[cfg(feature = "ipv6")]
9use std::net::Ipv6Addr;
10use std::sync::Arc;
11#[cfg(test)]
12use std::time::Instant;
13
14use bytes::{Bytes, BytesMut};
15use tokio::sync::{broadcast, mpsc, oneshot, Mutex};
16use tokio::task::JoinHandle;
17#[cfg(test)]
18use tokio::time::timeout;
19use tokio::time::Duration;
20use tracing::{debug, warn};
21
22use bacnet_encoding::apdu::{
23    self, encode_apdu, validate_max_apdu_length, validate_max_segments, AbortPdu, Apdu,
24    ConfirmedRequest as ConfirmedRequestPdu, RejectPdu, SegmentAck as SegmentAckPdu, SimpleAck,
25};
26use bacnet_encoding::npdu::{encode_npdu, Npdu, NpduAddress};
27use bacnet_endpoint_core::coordinator::{CanonicalPeer, OutboundTransactionCoordinator};
28use bacnet_network::layer::NetworkLayer;
29use bacnet_services::cov::COVNotificationRequest;
30use bacnet_transport::bip::BipTransport;
31#[cfg(feature = "ipv6")]
32use bacnet_transport::bip6::Bip6Transport;
33use bacnet_transport::port::TransportPort;
34use bacnet_types::enums::{
35    ConfirmedServiceChoice, NetworkPriority, RejectReason, UnconfirmedServiceChoice,
36};
37use bacnet_types::error::Error;
38use bacnet_types::MacAddr;
39
40use crate::discovery::{DeviceTable, DeviceUpsertResult, DiscoveredDevice, RoutedDeviceConfig};
41use crate::segmentation::{
42    duplicate_in_window, max_segment_payload, split_payload, SegmentReceiver, SegmentedPduType,
43};
44use crate::tsm::{
45    RequestTimerExpiration, SegmentAckPhase, SegmentTimerExpiration, SegmentedResponseAdmission,
46    TransactionOwner, TransactionProgress, Tsm, TsmConfig, TsmResponse,
47};
48#[cfg(test)]
49use transaction_cleanup::{SegmentedCleanupHook, SegmentedPostWaitCleanupHook};
50use transaction_cleanup::{TransactionCleanup, TransactionGuard};
51
52/// Default COV notification broadcast channel capacity.
53pub const DEFAULT_COV_CHANNEL_CAPACITY: usize = 64;
54
55/// Maximum COV notification broadcast channel capacity accepted at startup.
56pub const MAX_COV_CHANNEL_CAPACITY: usize = 65_536;
57
58/// Device discovery event broadcast channel capacity.
59pub const DEVICE_EVENT_CHANNEL_CAPACITY: usize = 64;
60
61const VALID_MAX_APDU_LENGTHS: [u16; 6] = [50, 128, 206, 480, 1024, 1476];
62
63/// Type of change observed in the device discovery table.
64#[derive(Debug, Clone, Copy, PartialEq, Eq)]
65pub enum DeviceEventKind {
66    /// A new device instance was inserted into the discovery table.
67    Discovered,
68    /// An existing device instance was refreshed by a later I-Am.
69    Updated,
70    /// A previously discovered device was purged as stale.
71    Lost,
72}
73
74/// Notification emitted when the client observes a device discovery change.
75#[derive(Debug, Clone)]
76pub struct DeviceEvent {
77    /// The kind of discovery-table change.
78    pub kind: DeviceEventKind,
79    /// Snapshot of the device entry after discovery/update or before removal.
80    pub device: DiscoveredDevice,
81}
82
83/// Notification emitted when two endpoints claim the same Device instance.
84#[derive(Debug, Clone)]
85pub struct DeviceCollisionEvent {
86    /// Snapshot of the authoritative discovery-table row that was retained.
87    pub retained: DiscoveredDevice,
88    /// Snapshot constructed from the conflicting incoming I-Am.
89    pub incoming: DiscoveredDevice,
90}
91
92/// Client configuration.
93#[derive(Debug, Clone)]
94pub struct ClientConfig {
95    /// Local interface to bind.
96    pub interface: Ipv4Addr,
97    /// UDP port (0 for ephemeral).
98    pub port: u16,
99    /// Directed broadcast address.
100    pub broadcast_address: Ipv4Addr,
101    /// APDU timeout in milliseconds.
102    pub apdu_timeout_ms: u64,
103    /// Number of APDU retries.
104    pub apdu_retries: u8,
105    /// Maximum APDU length this client accepts.
106    pub max_apdu_length: u16,
107    /// Maximum segments this client accepts (`None` = unspecified).
108    ///
109    /// Values 0 and 1 are invalid. Finite values between the BACnet wire
110    /// encodings are conservatively rounded down (for example, 5 advertises 4).
111    pub max_segments: Option<u8>,
112    /// Whether this client accepts segmented responses.
113    pub segmented_response_accepted: bool,
114    /// Proposed window size for segmented transfers (1-127, default 1).
115    pub proposed_window_size: u8,
116}
117
118/// Additional client startup options.
119#[derive(Clone)]
120pub struct ClientOptions {
121    /// Capacity of the COV notification broadcast channel.
122    ///
123    /// Slow receivers lag once more than this many notifications arrive before
124    /// they call `recv()`. The default preserves the historical fixed capacity
125    /// of 64.
126    pub cov_channel_capacity: usize,
127    confirmed_cov_notification_ack_policy: ConfirmedCOVNotificationAckPolicy,
128}
129
130impl Default for ClientConfig {
131    fn default() -> Self {
132        Self {
133            interface: Ipv4Addr::UNSPECIFIED,
134            port: 0xBAC0,
135            broadcast_address: Ipv4Addr::BROADCAST,
136            apdu_timeout_ms: 6000,
137            apdu_retries: 3,
138            max_apdu_length: 1476,
139            max_segments: None,
140            segmented_response_accepted: true,
141            proposed_window_size: 1,
142        }
143    }
144}
145
146impl Default for ClientOptions {
147    fn default() -> Self {
148        Self {
149            cov_channel_capacity: DEFAULT_COV_CHANNEL_CAPACITY,
150            confirmed_cov_notification_ack_policy:
151                cov_notifications::default_confirmed_cov_notification_ack_policy(),
152        }
153    }
154}
155
156pub(crate) fn new_coordinated_tsm(
157    config: &ClientConfig,
158    coordinator: Arc<OutboundTransactionCoordinator>,
159) -> Tsm {
160    Tsm::new_coordinated(
161        TsmConfig {
162            apdu_timeout_ms: config.apdu_timeout_ms,
163            apdu_segment_timeout_ms: config.apdu_timeout_ms,
164            apdu_retries: config.apdu_retries,
165        },
166        coordinator,
167    )
168}
169
170pub(crate) fn confirmed_response_result(response: TsmResponse) -> Result<Bytes, Error> {
171    match response {
172        TsmResponse::SimpleAck => Ok(Bytes::new()),
173        TsmResponse::ComplexAck { service_data } => Ok(service_data),
174        TsmResponse::Error { class, code } => Err(Error::Protocol { class, code }),
175        TsmResponse::Reject { reason } => Err(Error::Reject { reason }),
176        TsmResponse::Abort { reason } => Err(Error::Abort { reason }),
177        TsmResponse::NetworkPathTooLong { dnet } => Err(Error::RoutedPathTooLong { dnet }),
178    }
179}
180
181impl ClientOptions {
182    /// Set the COV notification broadcast channel capacity.
183    pub fn with_cov_channel_capacity(mut self, capacity: usize) -> Self {
184        self.cov_channel_capacity = capacity;
185        self
186    }
187
188    fn validate(&self) -> Result<(), Error> {
189        if !(1..=MAX_COV_CHANNEL_CAPACITY).contains(&self.cov_channel_capacity) {
190            return Err(Error::Encoding(format!(
191                "invalid cov-channel-capacity {}; expected 1..={}",
192                self.cov_channel_capacity, MAX_COV_CHANNEL_CAPACITY
193            )));
194        }
195        Ok(())
196    }
197}
198
199/// Generic builder for BACnetClient with a pre-built transport.
200pub struct ClientBuilder<T: TransportPort> {
201    config: ClientConfig,
202    options: ClientOptions,
203    transport: Option<T>,
204}
205
206impl<T: TransportPort + 'static> ClientBuilder<T> {
207    /// Set the pre-built transport.
208    pub fn transport(mut self, transport: T) -> Self {
209        self.transport = Some(transport);
210        self
211    }
212
213    /// Set APDU timeout in milliseconds.
214    pub fn apdu_timeout_ms(mut self, ms: u64) -> Self {
215        self.config.apdu_timeout_ms = ms;
216        self
217    }
218
219    /// Set the maximum APDU length this client accepts.
220    pub fn max_apdu_length(mut self, len: u16) -> Self {
221        self.config.max_apdu_length = len;
222        self
223    }
224
225    /// Set the COV notification broadcast channel capacity.
226    pub fn cov_channel_capacity(mut self, capacity: usize) -> Self {
227        self.options.cov_channel_capacity = capacity;
228        self
229    }
230
231    /// Build and start the client.
232    pub async fn build(self) -> Result<BACnetClient<T>, Error> {
233        let transport = self
234            .transport
235            .ok_or_else(|| Error::Encoding("transport not set on ClientBuilder".into()))?;
236        BACnetClient::start_with_options(self.config, transport, self.options).await
237    }
238}
239
240/// BIP-specific builder that constructs `BipTransport` from interface/port/broadcast fields.
241pub struct BipClientBuilder {
242    config: ClientConfig,
243    options: ClientOptions,
244}
245
246impl BipClientBuilder {
247    /// Set the local interface IP.
248    pub fn interface(mut self, ip: Ipv4Addr) -> Self {
249        self.config.interface = ip;
250        self
251    }
252
253    /// Set the UDP port (0 for ephemeral).
254    pub fn port(mut self, port: u16) -> Self {
255        self.config.port = port;
256        self
257    }
258
259    /// Set the directed broadcast address.
260    pub fn broadcast_address(mut self, addr: Ipv4Addr) -> Self {
261        self.config.broadcast_address = addr;
262        self
263    }
264
265    /// Set APDU timeout in milliseconds.
266    pub fn apdu_timeout_ms(mut self, ms: u64) -> Self {
267        self.config.apdu_timeout_ms = ms;
268        self
269    }
270
271    /// Set the maximum APDU length this client accepts.
272    pub fn max_apdu_length(mut self, len: u16) -> Self {
273        self.config.max_apdu_length = len;
274        self
275    }
276
277    /// Set the COV notification broadcast channel capacity.
278    pub fn cov_channel_capacity(mut self, capacity: usize) -> Self {
279        self.options.cov_channel_capacity = capacity;
280        self
281    }
282
283    /// Build and start the client, constructing a BipTransport from the config.
284    pub async fn build(self) -> Result<BACnetClient<BipTransport>, Error> {
285        let transport = BipTransport::new(
286            self.config.interface,
287            self.config.port,
288            self.config.broadcast_address,
289        );
290        BACnetClient::start_with_options(self.config, transport, self.options).await
291    }
292}
293
294// ---------------------------------------------------------------------------
295// Multi-device batch operation types
296// ---------------------------------------------------------------------------
297
298/// Default concurrency limit for multi-device batch operations.
299const DEFAULT_BATCH_CONCURRENCY: usize = 32;
300
301/// A request to read a single property from a discovered device.
302#[derive(Debug, Clone)]
303pub struct DeviceReadRequest {
304    /// Device instance number (must be in the device table).
305    pub device_instance: u32,
306    /// Object to read from.
307    pub object_identifier: bacnet_types::primitives::ObjectIdentifier,
308    /// Property to read.
309    pub property_identifier: bacnet_types::enums::PropertyIdentifier,
310    /// Optional array index.
311    pub property_array_index: Option<u32>,
312}
313
314/// Result of a single-property read from a device within a batch.
315#[derive(Debug)]
316pub struct DeviceReadResult {
317    /// The device instance this result corresponds to.
318    pub device_instance: u32,
319    /// The read result (Ok = decoded ACK, Err = protocol/timeout error).
320    pub result: Result<bacnet_services::read_property::ReadPropertyACK, Error>,
321}
322
323/// A request to read multiple properties from a discovered device (RPM).
324#[derive(Debug, Clone)]
325pub struct DeviceRpmRequest {
326    /// Device instance number (must be in the device table).
327    pub device_instance: u32,
328    /// ReadAccessSpecifications to send in a single RPM.
329    pub specs: Vec<bacnet_services::rpm::ReadAccessSpecification>,
330}
331
332/// Result of an RPM to a single device within a batch.
333#[derive(Debug)]
334pub struct DeviceRpmResult {
335    /// The device instance this result corresponds to.
336    pub device_instance: u32,
337    /// The RPM result.
338    pub result: Result<bacnet_services::rpm::ReadPropertyMultipleACK, Error>,
339}
340
341/// A request to write a single property on a discovered device.
342#[derive(Debug, Clone)]
343pub struct DeviceWriteRequest {
344    /// Device instance number (must be in the device table).
345    pub device_instance: u32,
346    /// Object to write to.
347    pub object_identifier: bacnet_types::primitives::ObjectIdentifier,
348    /// Property to write.
349    pub property_identifier: bacnet_types::enums::PropertyIdentifier,
350    /// Optional array index.
351    pub property_array_index: Option<u32>,
352    /// Encoded property value bytes.
353    pub property_value: Vec<u8>,
354    /// Optional write priority (1-16).
355    pub priority: Option<u8>,
356}
357
358/// Result of a single-property write to a device within a batch.
359#[derive(Debug)]
360pub struct DeviceWriteResult {
361    /// The device instance this result corresponds to.
362    pub device_instance: u32,
363    /// The write result (Ok = success, Err = protocol/timeout error).
364    pub result: Result<(), Error>,
365}
366
367/// The receive-side promises this client puts in every confirmed request.
368///
369/// Clause 20.1.2.3 and Clause 20.1.2.4 exist "so that the responding device may
370/// determine how to convey its response", so the same values must bound what
371/// the dispatch loop is willing to take back. Keeping the pair together stops
372/// the two halves from drifting as they are threaded through dispatch.
373#[derive(Debug, Clone, Copy)]
374struct ResponseLimits {
375    /// Clause 20.1.2.3 'segmented-response-accepted'.
376    segmented_response_accepted: bool,
377    /// Most segments this client will hold for one reassembly.
378    max_reassembly_segments: usize,
379}
380
381/// In-progress segmented receive state.
382struct SegmentedReceiveState {
383    receiver: SegmentReceiver,
384    owner: TransactionOwner,
385    /// Immediate MAC used to send SegmentAck/Abort PDUs.
386    reply_mac: MacAddr,
387    /// The peer's SNET/SADR when the segments arrive through a router; the
388    /// reply's DNET/DADR, or the router takes the reply for itself instead
389    /// of forwarding it (Clause 6.5.2.1).
390    reply_network: Option<NpduAddress>,
391    /// Next expected sequence number (for gap detection).
392    expected_next_seq: u8,
393    /// Last sequence number in the previously completed receive window.
394    initial_sequence_number: u8,
395    /// Last segment accepted in order.
396    last_sequence_number: u8,
397    /// Duplicates silently discarded in the current receive window.
398    duplicate_count: u8,
399    /// Window position counter for per-window SegmentAck (Clause 5.2.2).
400    window_position: u8,
401    /// Window size accepted for this receive session.
402    actual_window_size: u8,
403    /// How many distinct segments have been stored.
404    ///
405    /// Monotonic, and deliberately not derived from `expected_next_seq`, which
406    /// is a `u8` and wraps: Clause 20.1.5.4 makes sequence numbers modulo 256,
407    /// so after 256 segments the counter returns to a slot already occupied.
408    /// This is the only value that can tell 257 segments from 1.
409    accepted_segments: usize,
410}
411
412/// Key for tracking in-progress segmented receives: (correlation_mac, invoke_id).
413type SegKey = (MacAddr, u8);
414
415struct SegmentAckRoute {
416    owner: TransactionOwner,
417    sender: mpsc::Sender<SegmentAckPdu>,
418}
419
420/// BACnet client with low-level and high-level request APIs.
421pub struct BACnetClient<T: TransportPort> {
422    config: ClientConfig,
423    network: Arc<NetworkLayer<T>>,
424    tsm: Arc<Mutex<Tsm>>,
425    device_table: Arc<Mutex<DeviceTable>>,
426    cov_tx: broadcast::Sender<ReceivedCOVNotification>,
427    device_tx: broadcast::Sender<DeviceEvent>,
428    device_collision_tx: broadcast::Sender<DeviceCollisionEvent>,
429    dispatch_task: Option<JoinHandle<()>>,
430    /// Owner-qualified channels feeding SegmentACKs to in-flight segmented sends.
431    ///
432    /// Dispatch may hold [`Self::tsm`] while acquiring this lock so phase
433    /// validation and delivery cannot race terminal completion. No path may
434    /// acquire the locks in the opposite order.
435    seg_ack_senders: Arc<Mutex<HashMap<SegKey, SegmentAckRoute>>>,
436    cleanup_tx: mpsc::UnboundedSender<TransactionCleanup>,
437    #[cfg(test)]
438    segmented_post_wait_cleanup: Arc<SegmentedPostWaitCleanupHook>,
439    #[cfg(test)]
440    segmented_cleanup: Arc<SegmentedCleanupHook>,
441    local_mac: MacAddr,
442    routed_path_limits: Arc<RoutedPathLimits>,
443}
444
445impl BACnetClient<BipTransport> {
446    /// Create a BIP-specific builder with interface/port/broadcast fields.
447    pub fn bip_builder() -> BipClientBuilder {
448        BipClientBuilder {
449            config: ClientConfig::default(),
450            options: ClientOptions::default(),
451        }
452    }
453
454    pub fn builder() -> BipClientBuilder {
455        Self::bip_builder()
456    }
457
458    /// Read the Broadcast Distribution Table from a BBMD.
459    pub async fn read_bdt(
460        &self,
461        target: &[u8],
462    ) -> Result<Vec<bacnet_transport::bbmd::BdtEntry>, Error> {
463        self.network.transport().read_bdt(target).await
464    }
465
466    /// Write the Broadcast Distribution Table to a BBMD.
467    pub async fn write_bdt(
468        &self,
469        target: &[u8],
470        entries: &[bacnet_transport::bbmd::BdtEntry],
471    ) -> Result<bacnet_types::enums::BvlcResultCode, Error> {
472        self.network.transport().write_bdt(target, entries).await
473    }
474
475    /// Read the Foreign Device Table from a BBMD.
476    pub async fn read_fdt(
477        &self,
478        target: &[u8],
479    ) -> Result<Vec<bacnet_transport::bbmd::FdtEntryWire>, Error> {
480        self.network.transport().read_fdt(target).await
481    }
482
483    /// Delete a Foreign Device Table entry on a BBMD.
484    pub async fn delete_fdt_entry(
485        &self,
486        target: &[u8],
487        ip: [u8; 4],
488        port: u16,
489    ) -> Result<bacnet_types::enums::BvlcResultCode, Error> {
490        self.network
491            .transport()
492            .delete_fdt_entry(target, ip, port)
493            .await
494    }
495
496    /// Register as a foreign device with a BBMD and return the result code.
497    pub async fn register_foreign_device_bvlc(
498        &self,
499        target: &[u8],
500        ttl: u16,
501    ) -> Result<bacnet_types::enums::BvlcResultCode, Error> {
502        self.network
503            .transport()
504            .register_foreign_device_bvlc(target, ttl)
505            .await
506    }
507}
508
509#[cfg(feature = "ipv6")]
510impl BACnetClient<Bip6Transport> {
511    /// Create a BIP6-specific builder for BACnet/IPv6 transport.
512    pub fn bip6_builder() -> Bip6ClientBuilder {
513        Bip6ClientBuilder {
514            config: ClientConfig::default(),
515            options: ClientOptions::default(),
516            interface: Ipv6Addr::UNSPECIFIED,
517            device_instance: None,
518        }
519    }
520}
521
522/// BIP6-specific builder that constructs `Bip6Transport` from IPv6 interface/port/device-instance.
523#[cfg(feature = "ipv6")]
524pub struct Bip6ClientBuilder {
525    config: ClientConfig,
526    options: ClientOptions,
527    interface: Ipv6Addr,
528    device_instance: Option<u32>,
529}
530
531#[cfg(feature = "ipv6")]
532impl Bip6ClientBuilder {
533    /// Set the local IPv6 interface address.
534    pub fn interface(mut self, ip: Ipv6Addr) -> Self {
535        self.interface = ip;
536        self
537    }
538
539    /// Set the UDP port (0 for ephemeral).
540    pub fn port(mut self, port: u16) -> Self {
541        self.config.port = port;
542        self
543    }
544
545    /// Set the device instance for VMAC derivation (Annex U.5).
546    pub fn device_instance(mut self, instance: u32) -> Self {
547        self.device_instance = Some(instance);
548        self
549    }
550
551    /// Set APDU timeout in milliseconds.
552    pub fn apdu_timeout_ms(mut self, ms: u64) -> Self {
553        self.config.apdu_timeout_ms = ms;
554        self
555    }
556
557    /// Set the maximum APDU length this client accepts.
558    pub fn max_apdu_length(mut self, len: u16) -> Self {
559        self.config.max_apdu_length = len;
560        self
561    }
562
563    /// Set the COV notification broadcast channel capacity.
564    pub fn cov_channel_capacity(mut self, capacity: usize) -> Self {
565        self.options.cov_channel_capacity = capacity;
566        self
567    }
568
569    /// Build and start the client, constructing a Bip6Transport from the config.
570    pub async fn build(self) -> Result<BACnetClient<Bip6Transport>, Error> {
571        let transport = Bip6Transport::new(self.interface, self.config.port, self.device_instance);
572        BACnetClient::start_with_options(self.config, transport, self.options).await
573    }
574}
575
576#[cfg(feature = "sc-tls")]
577impl BACnetClient<bacnet_transport::sc::ScTransport<bacnet_transport::sc_tls::TlsWebSocket>> {
578    /// Create an SC-specific builder that connects to a BACnet/SC hub.
579    pub fn sc_builder() -> ScClientBuilder {
580        ScClientBuilder {
581            config: ClientConfig::default(),
582            options: ClientOptions::default(),
583            hub_url: String::new(),
584            tls_config: None,
585            vmac: [0; 6],
586            device_uuid: [0; 16],
587            heartbeat_interval_ms: 30_000,
588            heartbeat_timeout_ms: 60_000,
589            reconnect: None,
590        }
591    }
592}
593
594/// SC-specific client builder.
595///
596/// Created by [`BACnetClient::sc_builder()`].  Requires the `sc-tls` feature.
597#[cfg(feature = "sc-tls")]
598pub struct ScClientBuilder {
599    config: ClientConfig,
600    options: ClientOptions,
601    hub_url: String,
602    tls_config: Option<std::sync::Arc<tokio_rustls::rustls::ClientConfig>>,
603    vmac: bacnet_transport::sc_frame::Vmac,
604    device_uuid: [u8; 16],
605    heartbeat_interval_ms: u64,
606    heartbeat_timeout_ms: u64,
607    reconnect: Option<bacnet_transport::sc::ScReconnectConfig>,
608}
609
610#[cfg(feature = "sc-tls")]
611impl ScClientBuilder {
612    /// Set the hub WebSocket URL (e.g. `wss://hub.example.com/bacnet`).
613    pub fn hub_url(mut self, url: &str) -> Self {
614        self.hub_url = url.to_string();
615        self
616    }
617
618    /// Set the TLS client configuration.
619    pub fn tls_config(
620        mut self,
621        config: std::sync::Arc<tokio_rustls::rustls::ClientConfig>,
622    ) -> Self {
623        self.tls_config = Some(config);
624        self
625    }
626
627    /// Set the local VMAC address.
628    pub fn vmac(mut self, vmac: [u8; 6]) -> Self {
629        self.vmac = vmac;
630        self
631    }
632
633    /// Set the persistent BACnet/SC device UUID.
634    pub fn device_uuid(mut self, uuid: [u8; 16]) -> Self {
635        self.device_uuid = uuid;
636        self
637    }
638
639    /// Set the APDU timeout in milliseconds.
640    pub fn apdu_timeout_ms(mut self, ms: u64) -> Self {
641        self.config.apdu_timeout_ms = ms;
642        self
643    }
644
645    /// Set the COV notification broadcast channel capacity.
646    pub fn cov_channel_capacity(mut self, capacity: usize) -> Self {
647        self.options.cov_channel_capacity = capacity;
648        self
649    }
650
651    /// Set the heartbeat interval in milliseconds (default 30 000).
652    pub fn heartbeat_interval_ms(mut self, ms: u64) -> Self {
653        self.heartbeat_interval_ms = ms;
654        self
655    }
656
657    /// Set the heartbeat timeout in milliseconds (default 60 000).
658    pub fn heartbeat_timeout_ms(mut self, ms: u64) -> Self {
659        self.heartbeat_timeout_ms = ms;
660        self
661    }
662
663    /// Enable automatic reconnection with the given configuration.
664    pub fn reconnect(mut self, config: bacnet_transport::sc::ScReconnectConfig) -> Self {
665        self.reconnect = Some(config);
666        self
667    }
668
669    fn validate_identity(&self) -> Result<(), Error> {
670        match self.vmac {
671            bacnet_transport::sc_frame::UNKNOWN_VMAC => Err(Error::Encoding(
672                "SC client builder: vmac must not be the reserved unknown VMAC".into(),
673            )),
674            bacnet_transport::sc_frame::BROADCAST_VMAC => Err(Error::Encoding(
675                "SC client builder: vmac must not be the reserved broadcast VMAC".into(),
676            )),
677            _ if self.device_uuid == [0; 16] => Err(Error::Encoding(
678                "SC client builder: device_uuid is required and must not be all zero".into(),
679            )),
680            _ => Ok(()),
681        }
682    }
683
684    fn sc_transport<W: bacnet_transport::sc::WebSocketPort>(
685        &self,
686        ws: W,
687    ) -> bacnet_transport::sc::ScTransport<W> {
688        bacnet_transport::sc::ScTransport::new(ws, self.vmac)
689            .with_device_uuid(self.device_uuid)
690            .with_heartbeat_interval_ms(self.heartbeat_interval_ms)
691            .with_heartbeat_timeout_ms(self.heartbeat_timeout_ms)
692    }
693
694    #[cfg(test)]
695    pub(crate) async fn build_with_websocket_for_test<
696        W: bacnet_transport::sc::WebSocketPort + 'static,
697    >(
698        self,
699        ws: W,
700    ) -> Result<BACnetClient<bacnet_transport::sc::ScTransport<W>>, Error> {
701        if let Some(config) = &self.reconnect {
702            config.validate()?;
703        }
704        self.validate_identity()?;
705        self.options.validate()?;
706        validate_max_segments(self.config.max_segments)?;
707        let transport = self.sc_transport(ws);
708        BACnetClient::start_with_options(self.config, transport, self.options).await
709    }
710
711    /// Connect to the hub and start the client.
712    ///
713    /// Reconnect configuration is validated before TLS lookup or dialing, and
714    /// again when the transport starts. An error still consumes this builder
715    /// and drops its inputs; this does not promise generic endpoint rollback.
716    pub async fn build(
717        self,
718    ) -> Result<
719        BACnetClient<bacnet_transport::sc::ScTransport<bacnet_transport::sc_tls::TlsWebSocket>>,
720        Error,
721    > {
722        if let Some(config) = &self.reconnect {
723            config.validate()?;
724        }
725        self.validate_identity()?;
726        self.options.validate()?;
727        validate_max_segments(self.config.max_segments)?;
728        let tls_config =
729            self.tls_config.as_ref().cloned().ok_or_else(|| {
730                Error::Encoding("SC client builder: tls_config is required".into())
731            })?;
732
733        let ws = bacnet_transport::sc_tls::TlsWebSocket::connect(&self.hub_url, tls_config.clone())
734            .await?;
735
736        let mut transport = self.sc_transport(ws);
737        if let Some(rc) = self.reconnect {
738            let hub_url = self.hub_url.clone();
739            let tls_config = tls_config.clone();
740            #[allow(deprecated)]
741            {
742                transport = transport
743                    .with_connector(move || {
744                        let hub_url = hub_url.clone();
745                        let tls_config = tls_config.clone();
746                        async move {
747                            bacnet_transport::sc_tls::TlsWebSocket::connect(&hub_url, tls_config)
748                                .await
749                        }
750                    })
751                    .with_reconnect(rc);
752            }
753        }
754
755        BACnetClient::start_with_options(self.config, transport, self.options).await
756    }
757}
758
759/// Routing target for confirmed requests.
760#[derive(Clone, Copy)]
761enum ConfirmedTarget<'a> {
762    Local {
763        mac: &'a [u8],
764    },
765    Routed {
766        router_mac: &'a [u8],
767        dest_network: u16,
768        dest_mac: &'a [u8],
769    },
770}
771
772impl<'a> ConfirmedTarget<'a> {
773    fn additional_npdu_header_len(&self) -> u16 {
774        match self {
775            Self::Local { .. } => 0,
776            Self::Routed { dest_mac, .. } => u16::try_from(dest_mac.len())
777                .unwrap_or(u16::MAX)
778                .saturating_add(4),
779        }
780    }
781}
782
783fn max_apdu_bucket_at_or_below(limit: u16) -> Option<u16> {
784    VALID_MAX_APDU_LENGTHS
785        .iter()
786        .rev()
787        .copied()
788        .find(|bucket| *bucket <= limit)
789}
790
791fn cap_max_apdu_to_transport(configured: u16, transport_limit: u16) -> Result<u16, Error> {
792    let limit = configured.min(transport_limit);
793    max_apdu_bucket_at_or_below(limit).ok_or_else(|| {
794        Error::Encoding(format!(
795            "transport max-APDU-length {transport_limit} leaves no valid BACnet APDU bucket"
796        ))
797    })
798}
799
800mod audit;
801mod builder_options;
802mod cov;
803mod cov_notifications;
804mod cov_renewal;
805mod device_events;
806mod device_mgmt;
807mod discovery;
808mod dispatch;
809mod file_list;
810mod lifecycle;
811mod object_mgmt;
812mod property;
813mod requests;
814mod response_admission;
815mod routed_path_limits;
816mod segmentation;
817mod segmented_request;
818mod transaction_cleanup;
819mod transaction_peer;
820use routed_path_limits::{routed_path_quarantine_horizon, RoutedPathLease, RoutedPathLimits};
821use transaction_peer::response_transaction_peer;
822
823pub use cov_notifications::{
824    COVNotificationDelivery, ConfirmedCOVNotificationAckPolicy, ConfirmedCOVNotificationResponse,
825    ReceivedCOVNotification,
826};
827pub use cov_renewal::{
828    ManagedCOVSubscription, ManagedCOVSubscriptionEvent, ManagedCOVSubscriptionOptions,
829};
830
831#[cfg(test)]
832mod acknowledge_alarm_tests;
833#[cfg(test)]
834mod audit_tests;
835#[cfg(test)]
836mod builder_options_tests;
837#[cfg(test)]
838mod confirmed_request_dispatch_tests;
839#[cfg(test)]
840mod coordinator_tests;
841#[cfg(test)]
842mod cov_notification_tests;
843#[cfg(test)]
844mod cov_renewal_tests;
845#[cfg(test)]
846mod cov_tests;
847#[cfg(test)]
848mod device_events_tests;
849#[cfg(test)]
850mod peer_max_apdu_tests;
851#[cfg(test)]
852mod peer_segmentation_tests;
853#[cfg(test)]
854mod request_timer_tests;
855#[cfg(test)]
856mod response_correlation_tests;
857#[cfg(test)]
858mod routed_max_apdu_tests;
859#[cfg(test)]
860mod routed_path_limit_tests;
861#[cfg(test)]
862mod routed_reply_tests;
863#[cfg(all(test, feature = "sc-tls"))]
864mod sc_builder_tests;
865#[cfg(test)]
866mod sc_max_apdu_tests;
867#[cfg(test)]
868mod segmentation_retransmit_tests;
869#[cfg(test)]
870mod segmented_receive_duplicate_tests;
871#[cfg(test)]
872mod segmented_receive_lifecycle_tests;
873#[cfg(test)]
874mod segmented_request_ordering_tests;
875#[cfg(test)]
876mod segmented_request_state_tests;
877#[cfg(test)]
878mod segmented_response_admission_tests;
879#[cfg(test)]
880mod segmented_response_capacity_tests;
881#[cfg(test)]
882mod segmented_timeout_tests;
883#[cfg(test)]
884mod tests;
885
886impl<T: TransportPort + 'static> BACnetClient<T> {
887    /// Create a generic builder that accepts a pre-built transport.
888    pub fn generic_builder() -> ClientBuilder<T> {
889        ClientBuilder {
890            config: ClientConfig::default(),
891            options: ClientOptions::default(),
892            transport: None,
893        }
894    }
895}
bacnet_client/client/mod.rs

bacnet_client/client/
mod.rs
