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    /// Build and start the server, constructing a BipTransport from the config.
310    pub async fn build(self) -> Result<BACnetServer<BipTransport>, Error> {
311        let transport = BipTransport::new(
312            self.config.interface,
313            self.config.port,
314            self.config.broadcast_address,
315        );
316        BACnetServer::start(self.config, self.db, transport).await
317    }
318}
319
320/// Key for tracking in-progress segmented sends: (source_mac, invoke_id).
321type SegKey = (MacAddr, u8);
322
323struct SegmentedRequestState {
324    receiver: SegmentReceiver,
325    first_req: bacnet_encoding::apdu::ConfirmedRequest,
326    last_activity: Instant,
327    expected_seq: u8,
328    last_acked_seq: u8,
329    window_pos: u8,
330    actual_window_size: u8,
331}
332
333/// BACnet server with APDU dispatch and service handling.
334pub struct BACnetServer<T: TransportPort> {
335    #[allow(dead_code)]
336    config: ServerConfig,
337    /// Shared network layer (also held by dispatch task).
338    #[allow(dead_code)]
339    network: Arc<NetworkLayer<T>>,
340    /// Shared object database.
341    db: Arc<RwLock<ObjectDatabase>>,
342    /// COV subscription table (held to keep Arc alive for dispatch task).
343    #[allow(dead_code)]
344    cov_table: Arc<RwLock<CovSubscriptionTable>>,
345    /// Channels for routing SegmentAck PDUs to in-progress segmented sends.
346    #[allow(dead_code)]
347    seg_ack_senders: Arc<Mutex<HashMap<SegKey, mpsc::Sender<SegmentAckPdu>>>>,
348    /// Semaphore that caps confirmed COV notifications at 255 in-flight
349    /// to prevent invoke ID reuse (invoke IDs are u8 = 0..255).
350    #[allow(dead_code)]
351    cov_in_flight: Arc<Semaphore>,
352    /// Server-side TSM for outgoing confirmed COV notifications.
353    #[allow(dead_code)]
354    server_tsm: Arc<Mutex<ServerTsm>>,
355    /// Communication state: 0 = Enable, 1 = Disable, 2 = DisableInitiation.
356    comm_state: Arc<AtomicU8>,
357    /// Handle for the DCC auto-re-enable timer. A new DCC request aborts
358    /// any previous timer.
359    #[allow(dead_code)]
360    dcc_timer: Arc<Mutex<Option<JoinHandle<()>>>>,
361    dispatch_task: Option<JoinHandle<()>>,
362    cov_purge_task: Option<JoinHandle<()>>,
363    fault_detection_task: Option<JoinHandle<()>>,
364    event_enrollment_task: Option<JoinHandle<()>>,
365    trend_log_task: Option<JoinHandle<()>>,
366    schedule_tick_task: Option<JoinHandle<()>>,
367    local_mac: MacAddr,
368}
369
370impl BACnetServer<BipTransport> {
371    /// Create a BIP-specific builder with interface/port/broadcast fields.
372    pub fn bip_builder() -> BipServerBuilder {
373        BipServerBuilder {
374            config: ServerConfig::default(),
375            db: ObjectDatabase::new(),
376        }
377    }
378
379    /// Create a BIP-specific builder (alias for backward compatibility).
380    pub fn builder() -> BipServerBuilder {
381        Self::bip_builder()
382    }
383}
384
385#[cfg(feature = "sc-tls")]
386impl BACnetServer<bacnet_transport::sc::ScTransport<bacnet_transport::sc_tls::TlsWebSocket>> {
387    /// Create an SC-specific builder that connects to a BACnet/SC hub.
388    pub fn sc_builder() -> ScServerBuilder {
389        ScServerBuilder {
390            config: ServerConfig::default(),
391            db: ObjectDatabase::new(),
392            hub_url: String::new(),
393            tls_config: None,
394            vmac: [0; 6],
395            heartbeat_interval_ms: 30_000,
396            heartbeat_timeout_ms: 60_000,
397            reconnect: None,
398        }
399    }
400}
401
402/// SC-specific server builder.
403///
404/// Created by [`BACnetServer::sc_builder()`].  Requires the `sc-tls` feature.
405#[cfg(feature = "sc-tls")]
406pub struct ScServerBuilder {
407    config: ServerConfig,
408    db: ObjectDatabase,
409    hub_url: String,
410    tls_config: Option<std::sync::Arc<tokio_rustls::rustls::ClientConfig>>,
411    vmac: bacnet_transport::sc_frame::Vmac,
412    heartbeat_interval_ms: u64,
413    heartbeat_timeout_ms: u64,
414    reconnect: Option<bacnet_transport::sc::ScReconnectConfig>,
415}
416
417#[cfg(feature = "sc-tls")]
418impl ScServerBuilder {
419    /// Set the hub WebSocket URL (e.g. `wss://hub.example.com/bacnet`).
420    pub fn hub_url(mut self, url: &str) -> Self {
421        self.hub_url = url.to_string();
422        self
423    }
424
425    /// Set the TLS client configuration.
426    pub fn tls_config(
427        mut self,
428        config: std::sync::Arc<tokio_rustls::rustls::ClientConfig>,
429    ) -> Self {
430        self.tls_config = Some(config);
431        self
432    }
433
434    /// Set the local VMAC address.
435    pub fn vmac(mut self, vmac: [u8; 6]) -> Self {
436        self.vmac = vmac;
437        self
438    }
439
440    /// Set the object database (transfers ownership).
441    pub fn database(mut self, db: ObjectDatabase) -> Self {
442        self.db = db;
443        self
444    }
445
446    /// Set the heartbeat interval in milliseconds (default 30 000).
447    pub fn heartbeat_interval_ms(mut self, ms: u64) -> Self {
448        self.heartbeat_interval_ms = ms;
449        self
450    }
451
452    /// Set the heartbeat timeout in milliseconds (default 60 000).
453    pub fn heartbeat_timeout_ms(mut self, ms: u64) -> Self {
454        self.heartbeat_timeout_ms = ms;
455        self
456    }
457
458    /// Enable automatic reconnection with the given configuration.
459    pub fn reconnect(mut self, config: bacnet_transport::sc::ScReconnectConfig) -> Self {
460        self.reconnect = Some(config);
461        self
462    }
463
464    /// Set the password required for DeviceCommunicationControl requests.
465    pub fn dcc_password(mut self, password: impl Into<String>) -> Self {
466        self.config.dcc_password = Some(password.into());
467        self
468    }
469
470    /// Set the password required for ReinitializeDevice requests.
471    pub fn reinit_password(mut self, password: impl Into<String>) -> Self {
472        self.config.reinit_password = Some(password.into());
473        self
474    }
475
476    /// Enable periodic fault detection / reliability evaluation.
477    pub fn enable_fault_detection(mut self, enabled: bool) -> Self {
478        self.config.enable_fault_detection = enabled;
479        self
480    }
481
482    /// Connect to the hub and start the server.
483    pub async fn build(
484        self,
485    ) -> Result<
486        BACnetServer<bacnet_transport::sc::ScTransport<bacnet_transport::sc_tls::TlsWebSocket>>,
487        Error,
488    > {
489        let tls_config = self
490            .tls_config
491            .ok_or_else(|| Error::Encoding("SC server builder: tls_config is required".into()))?;
492
493        let ws = bacnet_transport::sc_tls::TlsWebSocket::connect(&self.hub_url, tls_config).await?;
494
495        let mut transport = bacnet_transport::sc::ScTransport::new(ws, self.vmac)
496            .with_heartbeat_interval_ms(self.heartbeat_interval_ms)
497            .with_heartbeat_timeout_ms(self.heartbeat_timeout_ms);
498        if let Some(rc) = self.reconnect {
499            #[allow(deprecated)]
500            {
501                transport = transport.with_reconnect(rc);
502            }
503        }
504
505        BACnetServer::start(self.config, self.db, transport).await
506    }
507}
508
509mod cov_notifications;
510mod dispatch;
511mod event_notifications;
512mod lifecycle;
513mod requests;
514mod segmentation;
515
516#[cfg(test)]
517mod tests;
518
519impl<T: TransportPort + 'static> BACnetServer<T> {
520    pub fn generic_builder() -> ServerBuilder<T> {
521        ServerBuilder {
522            config: ServerConfig::default(),
523            db: ObjectDatabase::new(),
524            transport: None,
525        }
526    }
527}
bacnet_server/server/mod.rs

bacnet_server/server/
mod.rs
