bacnet_server/server/
mod.rs

1//! BACnetServer: builder, APDU dispatch, and lifecycle management.
2//!
3//! The server wraps a NetworkLayer behind Arc (shared with the dispatch task),
4//! owns an ObjectDatabase via Arc<Mutex>, and spawns a dispatch task that
5//! routes incoming APDUs to service handlers.
6
7use std::collections::HashMap;
8use std::net::Ipv4Addr;
9use std::sync::atomic::{AtomicU8, Ordering};
10use std::sync::Arc;
11use std::time::Instant;
12
13use bytes::{Bytes, BytesMut};
14use tokio::sync::{mpsc, oneshot, Mutex, RwLock, Semaphore};
15use tokio::task::JoinHandle;
16use tokio::time::Duration;
17use tracing::{debug, warn};
18
19use bacnet_encoding::apdu::{
20    self, encode_apdu, validate_max_apdu_length, AbortPdu, Apdu, ComplexAck,
21    ConfirmedRequest as ConfirmedRequestPdu, ErrorPdu, RejectPdu, SegmentAck as SegmentAckPdu,
22    SimpleAck, UnconfirmedRequest as UnconfirmedRequestPdu,
23};
24use bacnet_encoding::primitives::encode_property_value;
25use bacnet_encoding::segmentation::{
26    max_segment_payload, split_payload, SegmentReceiver, SegmentedPduType,
27};
28use bacnet_network::layer::NetworkLayer;
29use bacnet_objects::database::ObjectDatabase;
30use bacnet_objects::notification_class::get_notification_recipients;
31use bacnet_services::alarm_event::EventNotificationRequest;
32use bacnet_services::common::BACnetPropertyValue;
33use bacnet_services::cov::COVNotificationRequest;
34use bacnet_services::who_is::{IAmRequest, WhoIsRequest};
35use bacnet_transport::bip::BipTransport;
36use bacnet_transport::port::TransportPort;
37use bacnet_types::enums::{
38    AbortReason, ConfirmedServiceChoice, ErrorClass, ErrorCode, NetworkPriority, NotifyType,
39    ObjectType, PropertyIdentifier, RejectReason, Segmentation, UnconfirmedServiceChoice,
40};
41use bacnet_types::error::Error;
42use bacnet_types::primitives::{BACnetTimeStamp, ObjectIdentifier, PropertyValue, Time};
43use bacnet_types::MacAddr;
44
45use crate::cov::CovSubscriptionTable;
46use crate::handlers;
47
48/// Maximum number of concurrent segmented reassembly sessions.
49const MAX_SEG_RECEIVERS: usize = 128;
50
51/// Timeout for idle segmented reassembly sessions.
52const SEG_RECEIVER_TIMEOUT: Duration = Duration::from_secs(4);
53
54/// Maximum negative SegmentAck retries during segmented response send.
55const MAX_NEG_SEGMENT_ACK_RETRIES: u8 = 3;
56
57/// Default number of APDU retries for confirmed COV notifications.
58const DEFAULT_APDU_RETRIES: u8 = 3;
59
60// ---------------------------------------------------------------------------
61// Server-side Transaction State Machine (TSM) for outgoing confirmed requests
62// ---------------------------------------------------------------------------
63
64/// Result of a confirmed COV notification from the subscriber's perspective.
65#[derive(Debug, Clone, Copy, PartialEq, Eq)]
66pub enum CovAckResult {
67    /// SimpleAck received — subscriber accepted the notification.
68    Ack,
69    /// Error or Reject/Abort received — subscriber rejected the notification.
70    Error,
71}
72
73/// Lightweight TSM for tracking outgoing confirmed COV notifications.
74///
75/// The server allocates an invoke ID for each confirmed notification and the
76/// dispatch loop writes the result into a shared map when a SimpleAck, Error,
77/// Reject, or Abort is received.  The per-subscriber retry task polls the map
78/// after each timeout to decide whether to resend.
79pub struct ServerTsm {
80    next_invoke_id: u8,
81    /// Oneshot senders keyed by peer MAC and invoke ID. When a result arrives
82    /// from the dispatch loop, we send it directly — no polling needed.
83    pending: HashMap<(MacAddr, u8), oneshot::Sender<CovAckResult>>,
84}
85
86impl ServerTsm {
87    fn new() -> Self {
88        Self {
89            next_invoke_id: 0,
90            pending: HashMap::new(),
91        }
92    }
93
94    /// Allocate the next invoke ID and register a oneshot channel for the result.
95    /// Returns (invoke_id, receiver).
96    fn allocate(&mut self, peer: MacAddr) -> (u8, oneshot::Receiver<CovAckResult>) {
97        let id = self.next_invoke_id;
98        self.next_invoke_id = self.next_invoke_id.wrapping_add(1);
99        let rx = self.register(peer, id);
100        (id, rx)
101    }
102
103    /// Register or replace the pending receiver for a peer/invoke-id pair.
104    fn register(&mut self, peer: MacAddr, invoke_id: u8) -> oneshot::Receiver<CovAckResult> {
105        let (tx, rx) = oneshot::channel();
106        self.pending.insert((peer, invoke_id), tx);
107        rx
108    }
109
110    /// Record a result from the dispatch loop (SimpleAck, Error, etc.).
111    /// Sends immediately through the oneshot channel.
112    fn record_result(&mut self, peer: &MacAddr, invoke_id: u8, result: CovAckResult) -> bool {
113        if let Some(tx) = self.pending.remove(&(peer.clone(), invoke_id)) {
114            let _ = tx.send(result);
115            true
116        } else {
117            false
118        }
119    }
120
121    /// Remove a pending entry (cleanup on completion or exhaustion).
122    fn remove(&mut self, peer: &MacAddr, invoke_id: u8) {
123        self.pending.remove(&(peer.clone(), invoke_id));
124    }
125}
126
127/// Data from a TimeSynchronization request.
128#[derive(Debug, Clone)]
129pub struct TimeSyncData {
130    /// Raw service request bytes (caller can decode if needed).
131    pub raw_service_data: Bytes,
132    /// Whether this was a UTC time sync (vs. local).
133    pub is_utc: bool,
134}
135
136/// Server configuration.
137#[derive(Clone)]
138pub struct ServerConfig {
139    /// Local interface to bind.
140    pub interface: Ipv4Addr,
141    /// UDP port (default 0xBAC0 = 47808).
142    pub port: u16,
143    /// Directed broadcast address.
144    pub broadcast_address: Ipv4Addr,
145    /// Maximum APDU length accepted.
146    pub max_apdu_length: u32,
147    /// Segmentation support level.
148    pub segmentation_supported: Segmentation,
149    /// Vendor identifier.
150    pub vendor_id: u16,
151    /// Timeout in ms before retrying a failed confirmed COV notification send (default 3000ms).
152    pub cov_retry_timeout_ms: u64,
153    /// Optional callback invoked when a TimeSynchronization request is received.
154    pub on_time_sync: Option<Arc<dyn Fn(TimeSyncData) + Send + Sync>>,
155    /// Optional password required for DeviceCommunicationControl.
156    pub dcc_password: Option<String>,
157    /// Optional password required for ReinitializeDevice.
158    pub reinit_password: Option<String>,
159    /// Enable periodic fault detection / reliability evaluation.
160    /// When true, the server evaluates analog objects every 10 s for
161    /// OVER_RANGE / UNDER_RANGE faults.
162    pub enable_fault_detection: bool,
163}
164
165impl std::fmt::Debug for ServerConfig {
166    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
167        f.debug_struct("ServerConfig")
168            .field("interface", &self.interface)
169            .field("port", &self.port)
170            .field("broadcast_address", &self.broadcast_address)
171            .field("max_apdu_length", &self.max_apdu_length)
172            .field("segmentation_supported", &self.segmentation_supported)
173            .field("vendor_id", &self.vendor_id)
174            .field("cov_retry_timeout_ms", &self.cov_retry_timeout_ms)
175            .field(
176                "on_time_sync",
177                &self.on_time_sync.as_ref().map(|_| "<callback>"),
178            )
179            .field("dcc_password", &self.dcc_password.as_ref().map(|_| "***"))
180            .field(
181                "reinit_password",
182                &self.reinit_password.as_ref().map(|_| "***"),
183            )
184            .field("enable_fault_detection", &self.enable_fault_detection)
185            .finish()
186    }
187}
188
189impl Default for ServerConfig {
190    fn default() -> Self {
191        Self {
192            interface: Ipv4Addr::UNSPECIFIED,
193            port: 0xBAC0,
194            broadcast_address: Ipv4Addr::BROADCAST,
195            max_apdu_length: 1476,
196            segmentation_supported: Segmentation::NONE,
197            vendor_id: 0,
198            cov_retry_timeout_ms: 3000,
199            on_time_sync: None,
200            dcc_password: None,
201            reinit_password: None,
202            enable_fault_detection: false,
203        }
204    }
205}
206
207/// Generic builder for BACnetServer with a pre-built transport.
208pub struct ServerBuilder<T: TransportPort> {
209    config: ServerConfig,
210    db: ObjectDatabase,
211    transport: Option<T>,
212}
213
214impl<T: TransportPort + 'static> ServerBuilder<T> {
215    /// Set the object database (transfers ownership).
216    pub fn database(mut self, db: ObjectDatabase) -> Self {
217        self.db = db;
218        self
219    }
220
221    /// Set the pre-built transport.
222    pub fn transport(mut self, transport: T) -> Self {
223        self.transport = Some(transport);
224        self
225    }
226
227    /// Set the password required for DeviceCommunicationControl requests.
228    pub fn dcc_password(mut self, password: impl Into<String>) -> Self {
229        self.config.dcc_password = Some(password.into());
230        self
231    }
232
233    /// Set the password required for ReinitializeDevice requests.
234    pub fn reinit_password(mut self, password: impl Into<String>) -> Self {
235        self.config.reinit_password = Some(password.into());
236        self
237    }
238
239    /// Enable periodic fault detection / reliability evaluation.
240    pub fn enable_fault_detection(mut self, enabled: bool) -> Self {
241        self.config.enable_fault_detection = enabled;
242        self
243    }
244
245    /// Set the vendor identifier (used in IAm responses and protocol operations).
246    pub fn vendor_id(mut self, id: u16) -> Self {
247        self.config.vendor_id = id;
248        self
249    }
250
251    /// Build and start the server.
252    pub async fn build(self) -> Result<BACnetServer<T>, Error> {
253        let transport = self
254            .transport
255            .ok_or_else(|| Error::Encoding("transport not set on ServerBuilder".into()))?;
256        BACnetServer::start(self.config, self.db, transport).await
257    }
258}
259
260/// BIP-specific builder that constructs `BipTransport` from interface/port/broadcast fields.
261pub struct BipServerBuilder {
262    config: ServerConfig,
263    db: ObjectDatabase,
264}
265
266impl BipServerBuilder {
267    /// Set the local interface IP.
268    pub fn interface(mut self, ip: Ipv4Addr) -> Self {
269        self.config.interface = ip;
270        self
271    }
272
273    /// Set the UDP port.
274    pub fn port(mut self, port: u16) -> Self {
275        self.config.port = port;
276        self
277    }
278
279    /// Set the directed broadcast address.
280    pub fn broadcast_address(mut self, addr: Ipv4Addr) -> Self {
281        self.config.broadcast_address = addr;
282        self
283    }
284
285    /// Set the object database (transfers ownership).
286    pub fn database(mut self, db: ObjectDatabase) -> Self {
287        self.db = db;
288        self
289    }
290
291    /// Set the password required for DeviceCommunicationControl requests.
292    pub fn dcc_password(mut self, password: impl Into<String>) -> Self {
293        self.config.dcc_password = Some(password.into());
294        self
295    }
296
297    /// Set the password required for ReinitializeDevice requests.
298    pub fn reinit_password(mut self, password: impl Into<String>) -> Self {
299        self.config.reinit_password = Some(password.into());
300        self
301    }
302
303    /// Enable periodic fault detection / reliability evaluation.
304    pub fn enable_fault_detection(mut self, enabled: bool) -> Self {
305        self.config.enable_fault_detection = enabled;
306        self
307    }
308
309    /// Set the vendor identifier advertised in I-Am responses.
310    pub fn vendor_id(mut self, id: u16) -> Self {
311        self.config.vendor_id = id;
312        self
313    }
314
315    /// Build and start the server, constructing a BipTransport from the config.
316    pub async fn build(self) -> Result<BACnetServer<BipTransport>, Error> {
317        let transport = BipTransport::new(
318            self.config.interface,
319            self.config.port,
320            self.config.broadcast_address,
321        );
322        BACnetServer::start(self.config, self.db, transport).await
323    }
324}
325
326/// Key for tracking in-progress segmented sends: (source_mac, invoke_id).
327type SegKey = (MacAddr, u8);
328
329struct SegmentedRequestState {
330    receiver: SegmentReceiver,
331    first_req: bacnet_encoding::apdu::ConfirmedRequest,
332    last_activity: Instant,
333    expected_seq: u8,
334    last_acked_seq: u8,
335    window_pos: u8,
336    actual_window_size: u8,
337}
338
339/// BACnet server with APDU dispatch and service handling.
340pub struct BACnetServer<T: TransportPort> {
341    #[allow(dead_code)]
342    config: ServerConfig,
343    /// Shared network layer (also held by dispatch task).
344    #[allow(dead_code)]
345    network: Arc<NetworkLayer<T>>,
346    /// Shared object database.
347    db: Arc<RwLock<ObjectDatabase>>,
348    /// COV subscription table (held to keep Arc alive for dispatch task).
349    #[allow(dead_code)]
350    cov_table: Arc<RwLock<CovSubscriptionTable>>,
351    /// Channels for routing SegmentAck PDUs to in-progress segmented sends.
352    #[allow(dead_code)]
353    seg_ack_senders: Arc<Mutex<HashMap<SegKey, mpsc::Sender<SegmentAckPdu>>>>,
354    /// Semaphore that caps confirmed COV notifications at 255 in-flight
355    /// to prevent invoke ID reuse (invoke IDs are u8 = 0..255).
356    #[allow(dead_code)]
357    cov_in_flight: Arc<Semaphore>,
358    /// Server-side TSM for outgoing confirmed COV notifications.
359    #[allow(dead_code)]
360    server_tsm: Arc<Mutex<ServerTsm>>,
361    /// Communication state: 0 = Enable, 1 = Disable, 2 = DisableInitiation.
362    comm_state: Arc<AtomicU8>,
363    /// Handle for the DCC auto-re-enable timer. A new DCC request aborts
364    /// any previous timer.
365    #[allow(dead_code)]
366    dcc_timer: Arc<Mutex<Option<JoinHandle<()>>>>,
367    dispatch_task: Option<JoinHandle<()>>,
368    cov_purge_task: Option<JoinHandle<()>>,
369    fault_detection_task: Option<JoinHandle<()>>,
370    event_enrollment_task: Option<JoinHandle<()>>,
371    trend_log_task: Option<JoinHandle<()>>,
372    schedule_tick_task: Option<JoinHandle<()>>,
373    local_mac: MacAddr,
374}
375
376/// Cloneable handle for sending unsolicited I-Am announcements.
377pub struct IAmBroadcaster<T: TransportPort> {
378    config: ServerConfig,
379    network: Arc<NetworkLayer<T>>,
380    db: Arc<RwLock<ObjectDatabase>>,
381}
382
383impl<T: TransportPort> Clone for IAmBroadcaster<T> {
384    fn clone(&self) -> Self {
385        Self {
386            config: self.config.clone(),
387            network: Arc::clone(&self.network),
388            db: Arc::clone(&self.db),
389        }
390    }
391}
392
393impl BACnetServer<BipTransport> {
394    /// Create a BIP-specific builder with interface/port/broadcast fields.
395    pub fn bip_builder() -> BipServerBuilder {
396        BipServerBuilder {
397            config: ServerConfig::default(),
398            db: ObjectDatabase::new(),
399        }
400    }
401
402    /// Create a BIP-specific builder (alias for backward compatibility).
403    pub fn builder() -> BipServerBuilder {
404        Self::bip_builder()
405    }
406}
407
408#[cfg(feature = "sc-tls")]
409impl BACnetServer<bacnet_transport::sc::ScTransport<bacnet_transport::sc_tls::TlsWebSocket>> {
410    /// Create an SC-specific builder that connects to a BACnet/SC hub.
411    pub fn sc_builder() -> ScServerBuilder {
412        ScServerBuilder {
413            config: ServerConfig::default(),
414            db: ObjectDatabase::new(),
415            hub_url: String::new(),
416            tls_config: None,
417            vmac: [0; 6],
418            heartbeat_interval_ms: 30_000,
419            heartbeat_timeout_ms: 60_000,
420            reconnect: None,
421        }
422    }
423}
424
425/// SC-specific server builder.
426///
427/// Created by [`BACnetServer::sc_builder()`].  Requires the `sc-tls` feature.
428#[cfg(feature = "sc-tls")]
429pub struct ScServerBuilder {
430    config: ServerConfig,
431    db: ObjectDatabase,
432    hub_url: String,
433    tls_config: Option<std::sync::Arc<tokio_rustls::rustls::ClientConfig>>,
434    vmac: bacnet_transport::sc_frame::Vmac,
435    heartbeat_interval_ms: u64,
436    heartbeat_timeout_ms: u64,
437    reconnect: Option<bacnet_transport::sc::ScReconnectConfig>,
438}
439
440#[cfg(feature = "sc-tls")]
441impl ScServerBuilder {
442    /// Set the hub WebSocket URL (e.g. `wss://hub.example.com/bacnet`).
443    pub fn hub_url(mut self, url: &str) -> Self {
444        self.hub_url = url.to_string();
445        self
446    }
447
448    /// Set the TLS client configuration.
449    pub fn tls_config(
450        mut self,
451        config: std::sync::Arc<tokio_rustls::rustls::ClientConfig>,
452    ) -> Self {
453        self.tls_config = Some(config);
454        self
455    }
456
457    /// Set the local VMAC address.
458    pub fn vmac(mut self, vmac: [u8; 6]) -> Self {
459        self.vmac = vmac;
460        self
461    }
462
463    /// Set the object database (transfers ownership).
464    pub fn database(mut self, db: ObjectDatabase) -> Self {
465        self.db = db;
466        self
467    }
468
469    /// Set the heartbeat interval in milliseconds (default 30 000).
470    pub fn heartbeat_interval_ms(mut self, ms: u64) -> Self {
471        self.heartbeat_interval_ms = ms;
472        self
473    }
474
475    /// Set the heartbeat timeout in milliseconds (default 60 000).
476    pub fn heartbeat_timeout_ms(mut self, ms: u64) -> Self {
477        self.heartbeat_timeout_ms = ms;
478        self
479    }
480
481    /// Enable automatic reconnection with the given configuration.
482    pub fn reconnect(mut self, config: bacnet_transport::sc::ScReconnectConfig) -> Self {
483        self.reconnect = Some(config);
484        self
485    }
486
487    /// Set the password required for DeviceCommunicationControl requests.
488    pub fn dcc_password(mut self, password: impl Into<String>) -> Self {
489        self.config.dcc_password = Some(password.into());
490        self
491    }
492
493    /// Set the password required for ReinitializeDevice requests.
494    pub fn reinit_password(mut self, password: impl Into<String>) -> Self {
495        self.config.reinit_password = Some(password.into());
496        self
497    }
498
499    /// Enable periodic fault detection / reliability evaluation.
500    pub fn enable_fault_detection(mut self, enabled: bool) -> Self {
501        self.config.enable_fault_detection = enabled;
502        self
503    }
504
505    /// Connect to the hub and start the server.
506    pub async fn build(
507        self,
508    ) -> Result<
509        BACnetServer<bacnet_transport::sc::ScTransport<bacnet_transport::sc_tls::TlsWebSocket>>,
510        Error,
511    > {
512        let tls_config = self
513            .tls_config
514            .ok_or_else(|| Error::Encoding("SC server builder: tls_config is required".into()))?;
515
516        let ws = bacnet_transport::sc_tls::TlsWebSocket::connect(&self.hub_url, tls_config).await?;
517
518        let mut transport = bacnet_transport::sc::ScTransport::new(ws, self.vmac)
519            .with_heartbeat_interval_ms(self.heartbeat_interval_ms)
520            .with_heartbeat_timeout_ms(self.heartbeat_timeout_ms);
521        if let Some(rc) = self.reconnect {
522            #[allow(deprecated)]
523            {
524                transport = transport.with_reconnect(rc);
525            }
526        }
527
528        BACnetServer::start(self.config, self.db, transport).await
529    }
530}
531
532mod cov_notifications;
533mod dispatch;
534mod event_notifications;
535mod lifecycle;
536mod requests;
537mod segmentation;
538
539#[cfg(test)]
540mod tests;
541
542impl<T: TransportPort + 'static> BACnetServer<T> {
543    pub fn generic_builder() -> ServerBuilder<T> {
544        ServerBuilder {
545            config: ServerConfig::default(),
546            db: ObjectDatabase::new(),
547            transport: None,
548        }
549    }
550}
bacnet_server/server/mod.rs

bacnet_server/server/
mod.rs
