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

bacnet_server/server/
mod.rs
