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

bacnet_client/client/
mod.rs
