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<RwLock>`, and spawns a dispatch task that
5//! routes incoming APDUs to service handlers.
6
7use std::collections::HashMap;
8use std::net::Ipv4Addr;
9#[cfg(test)]
10pub(crate) use std::sync::atomic::AtomicBool;
11use std::sync::atomic::Ordering;
12use std::sync::Arc;
13use std::time::Instant;
14
15use bytes::{Bytes, BytesMut};
16use tokio::sync::{mpsc, oneshot, watch, Mutex, RwLock, Semaphore};
17use tokio::task::JoinHandle;
18use tokio::time::Duration;
19use tracing::{debug, warn};
20
21use event_delivery::EventDelivery;
22
23use bacnet_encoding::apdu::{
24    self, encode_apdu, AbortPdu, Apdu, ComplexAck, ConfirmedRequest as ConfirmedRequestPdu,
25    ErrorPdu, RejectPdu, SegmentAck as SegmentAckPdu, SimpleAck,
26    UnconfirmedRequest as UnconfirmedRequestPdu,
27};
28use bacnet_encoding::npdu::NpduAddress;
29use bacnet_encoding::primitives::encode_property_value;
30use bacnet_encoding::segmentation::{
31    duplicate_in_window, max_segment_payload, split_payload, SegmentReceiver, SegmentedPduType,
32};
33use bacnet_network::layer::NetworkLayer;
34use bacnet_objects::database::ObjectDatabase;
35use bacnet_objects::notification_class::{
36    lookup_notification_recipients, resolve_transition_priority_ack,
37};
38use bacnet_services::alarm_event::EventNotificationRequest;
39use bacnet_services::common::BACnetPropertyValue;
40use bacnet_services::cov::COVNotificationRequest;
41use bacnet_services::cov_multiple::COVNotificationMultipleRequest;
42use bacnet_services::who_is::{DeviceInstanceRange, IAmRequest, WhoIsRequest};
43use bacnet_transport::bip::BipTransport;
44use bacnet_transport::port::TransportPort;
45use bacnet_types::enums::{
46    AbortReason, ConfirmedServiceChoice, ErrorClass, ErrorCode, LifeSafetyOperation,
47    NetworkPriority, NotifyType, ObjectType, PropertyIdentifier, RejectReason, Segmentation,
48    UnconfirmedServiceChoice,
49};
50use bacnet_types::error::Error;
51use bacnet_types::primitives::{ObjectIdentifier, PropertyValue};
52use bacnet_types::MacAddr;
53
54use crate::audit_notification::{
55    AuditNotificationAuthorizationContext, AuditNotificationAuthorizer,
56    UnconfirmedAuditNotificationAuthorizationContext, UnconfirmedAuditNotificationAuthorizer,
57    MAX_AUDIT_NOTIFICATIONS, MAX_AUDIT_NOTIFICATION_BYTES,
58};
59pub use crate::cov::{CovCounters, CovPolicy};
60use crate::cov::{
61    CovNotificationKind, CovSubscription, CovSubscriptionSnapshot, CovSubscriptionTable,
62};
63use crate::handlers;
64use crate::life_safety::{LifeSafetyOperationAuthorizationContext, LifeSafetyOperationAuthorizer};
65use confirmed_request_tracker::{
66    ConfirmedRequestAdmission, ConfirmedRequestTracker, PendingConfirmedRequest,
67};
68pub use device_bindings::DeviceBinding;
69use device_bindings::{register_configured_binding, DeviceBindingTable};
70use learned_router_cache::LearnedRouterCache;
71use lso_replay::{LsoAdmission, PendingLsoReplay};
72use notification_transactions::{
73    canonical_direct_peer, canonical_routed_peer, run_notification_under_dcc,
74    run_notification_worker, InitiationRestricted, NotificationTransactions,
75    NotificationWorkerResult,
76};
77use request_services::{DispatchContext, RequestOrigin, RequestServices, UnconfirmedServices};
78use requests::confirmed_response::ResponseTarget;
79use requests::ConfirmedRequestOwnership;
80use segmentation::{ComplexAckParams, SegmentedSendResources};
81
82/// Maximum number of concurrent segmented reassembly sessions.
83const MAX_SEG_RECEIVERS: usize = 128;
84
85/// Hard per-request reassembly ceiling: the sequence-number space (#364).
86///
87/// A local storage bound, not a protocol one. Clause 20.1.2.7 makes the
88/// request sequence number modulo 256, so a longer request is entirely
89/// representable on the wire — this server simply keys its segment store by
90/// that `u8` and cannot tell segment 256 from segment 0. A Device object
91/// configured to receive segments does publish a tighter advertisement — its
92/// `Max_Segments_Accepted` (Clause 12.11) defaults to `Unsigned(65)` — but
93/// enforcing it here is deliberately not done: accepting more segments than
94/// advertised is permissive, not a violation, while the sequence space is the
95/// line past which acceptance silently corrupts. Exactly 256 segments
96/// reassemble correctly and must keep working; 257 is the first that would
97/// corrupt the payload.
98const MAX_REQUEST_SEGMENTS: usize = 256;
99
100/// Maximum number of concurrent segmented response send sessions.
101const MAX_SEG_SENDERS: usize = 128;
102
103/// Timeout for idle segmented reassembly sessions.
104const SEG_RECEIVER_TIMEOUT: Duration = Duration::from_secs(4);
105
106/// Maximum negative SegmentAck retries during segmented response send.
107const MAX_NEG_SEGMENT_ACK_RETRIES: u8 = 3;
108
109/// Default timeout while waiting for SegmentACK during segmented response send.
110const DEFAULT_APDU_SEGMENT_TIMEOUT: Duration = Duration::from_secs(5);
111
112/// Default retransmission budget for segmented response segments.
113const DEFAULT_APDU_SEGMENT_RETRIES: u8 = MAX_NEG_SEGMENT_ACK_RETRIES;
114
115/// Default number of APDU retries for confirmed notifications and the
116/// WriteProperty requests Command actions send to other devices.
117const DEFAULT_APDU_RETRIES: u8 = 3;
118
119/// Data from a TimeSynchronization request.
120#[derive(Debug, Clone)]
121pub struct TimeSyncData {
122    /// Raw service request bytes (caller can decode if needed).
123    pub raw_service_data: Bytes,
124    /// Whether this was a UTC time sync (vs. local).
125    pub is_utc: bool,
126    /// Transport-native source MAC; on SC this is the source VMAC, not a
127    /// certificate principal. This metadata is a claimed identity only.
128    pub source_mac: MacAddr,
129    /// Claimed routed NPDU source, if present; takes precedence for policy matching.
130    pub source_network: Option<NpduAddress>,
131    /// Honest transport + origin provenance, threaded for diagnostics only
132    /// (RB-07 compat mode: decisions unchanged; RB-09 consumes it later).
133    pub provenance: bacnet_transport::port::TransportProvenance,
134}
135
136mod config;
137pub use config::ServerConfig;
138mod audit_batch_queue;
139mod audit_batch_runtime;
140mod audit_context_preparation;
141mod audit_forwarder;
142#[cfg(test)]
143mod audit_forwarder_tests;
144mod audit_recipient;
145mod audit_recipient_routes;
146mod audit_reporter;
147mod audit_reporter_changes;
148mod audit_send_now;
149mod heap_futures;
150pub use audit_reporter::{valid_bip_audit_address, AuditReportersConfig};
151#[cfg(test)]
152mod audit_reporter_tests;
153
154/// Generic builder for BACnetServer with a pre-built transport.
155pub struct ServerBuilder<T: TransportPort> {
156    config: ServerConfig,
157    db: ObjectDatabase,
158    transport: Option<T>,
159    configured_device_bindings: Vec<DeviceBinding>,
160}
161
162impl<T: TransportPort + 'static> ServerBuilder<T> {
163    /// Set the object database (transfers ownership).
164    pub fn database(mut self, db: ObjectDatabase) -> Self {
165        self.db = db;
166        self
167    }
168
169    /// Set the pre-built transport.
170    pub fn transport(mut self, transport: T) -> Self {
171        self.transport = Some(transport);
172        self
173    }
174
175    /// Register one explicit unicast route for a Device recipient.
176    pub fn device_binding(mut self, binding: DeviceBinding) -> Result<Self, Error> {
177        register_configured_binding(&mut self.configured_device_bindings, binding)?;
178        Ok(self)
179    }
180
181    /// Set the password required for ReinitializeDevice requests.
182    pub fn reinit_password(mut self, password: impl Into<String>) -> Self {
183        self.config.reinit_password = Some(password.into());
184        self
185    }
186
187    /// Set the policy that authorizes inbound LifeSafetyOperation requests.
188    pub fn life_safety_operation_authorizer<F>(mut self, authorizer: F) -> Self
189    where
190        F: Fn(&LifeSafetyOperationAuthorizationContext) -> bool + Send + Sync + 'static,
191    {
192        self.config.life_safety_operation_authorizer = Some(Arc::new(authorizer));
193        self
194    }
195
196    /// Select the only local Audit Log that receives authorized notifications.
197    pub fn audit_notification_sink(mut self, sink: ObjectIdentifier) -> Self {
198        self.config.audit_notification_sink = Some(sink);
199        self
200    }
201
202    /// Set the fail-closed ConfirmedAuditNotification authorization policy.
203    pub fn audit_notification_authorizer<F>(mut self, authorizer: F) -> Self
204    where
205        F: Fn(&AuditNotificationAuthorizationContext) -> bool + Send + Sync + 'static,
206    {
207        self.config.audit_notification_authorizer = Some(Arc::new(authorizer));
208        self
209    }
210
211    /// Set the fail-closed UnconfirmedAuditNotification authorization policy.
212    pub fn unconfirmed_audit_notification_authorizer<F>(mut self, authorizer: F) -> Self
213    where
214        F: Fn(&UnconfirmedAuditNotificationAuthorizationContext) -> bool + Send + Sync + 'static,
215    {
216        self.config.unconfirmed_audit_notification_authorizer = Some(Arc::new(authorizer));
217        self
218    }
219
220    /// Enable periodic fault detection / reliability evaluation.
221    ///
222    /// When enabled, every object's opt-in reliability hook runs every 10
223    /// seconds; the default hook is a no-op.
224    ///
225    /// Reliability evaluation only; Event Enrollment evaluation is configured
226    /// by [`enable_event_enrollment`](Self::enable_event_enrollment).
227    pub fn enable_fault_detection(mut self, enabled: bool) -> Self {
228        self.config.enable_fault_detection = enabled;
229        self
230    }
231
232    /// Enable periodic Event Enrollment evaluation (default `true`).
233    pub fn enable_event_enrollment(mut self, enabled: bool) -> Self {
234        self.config.enable_event_enrollment = enabled;
235        self
236    }
237
238    /// Set the interval in seconds between Event Enrollment evaluation passes
239    /// (default 10).
240    pub fn event_enrollment_interval_secs(mut self, secs: u64) -> Self {
241        self.config.event_enrollment_interval_secs = secs;
242        self
243    }
244
245    /// Set the segmentation support this device advertises and enforces.
246    ///
247    /// The dispatch loop honors the advertisement (Clause 5.4.5.1): inbound
248    /// segmented requests are reassembled only under `BOTH` or `RECEIVE`, and
249    /// draw a SEGMENTATION_NOT_SUPPORTED Abort otherwise. The default is
250    /// `NONE`.
251    pub fn segmentation_supported(mut self, segmentation: Segmentation) -> Self {
252        self.config.segmentation_supported = segmentation;
253        self
254    }
255
256    /// Set the vendor identifier (used in IAm responses and protocol operations).
257    pub fn vendor_id(mut self, id: u16) -> Self {
258        self.config.vendor_id = id;
259        self
260    }
261
262    /// Set the discovery rate-limiting and duplicate suppression policy.
263    pub fn discovery_policy(mut self, policy: DiscoveryPolicy) -> Self {
264        self.config.discovery_policy = policy;
265        self
266    }
267
268    /// Set the COV quota and notification work budget policy, checked by
269    /// [`CovPolicy::validate`] before transport startup.
270    pub fn cov_policy(mut self, policy: CovPolicy) -> Self {
271        self.config.cov_policy = policy;
272        self
273    }
274
275    /// Build and start the server.
276    pub async fn build(self) -> Result<BACnetServer<T>, Error> {
277        let transport = self
278            .transport
279            .ok_or_else(|| Error::Encoding("transport not set on ServerBuilder".into()))?;
280        BACnetServer::start_with_clock_mode_and_bindings(
281            self.config,
282            self.db,
283            transport,
284            Some(ClockConfig::default()),
285            self.configured_device_bindings,
286        )
287        .await
288    }
289}
290
291/// BIP-specific builder that constructs `BipTransport` from interface/port/broadcast fields.
292pub struct BipServerBuilder {
293    config: ServerConfig,
294    db: ObjectDatabase,
295    configured_device_bindings: Vec<DeviceBinding>,
296}
297
298impl BipServerBuilder {
299    /// Set the local interface IP.
300    pub fn interface(mut self, ip: Ipv4Addr) -> Self {
301        self.config.interface = ip;
302        self
303    }
304
305    /// Set the UDP port.
306    pub fn port(mut self, port: u16) -> Self {
307        self.config.port = port;
308        self
309    }
310
311    /// Set the directed broadcast address.
312    pub fn broadcast_address(mut self, addr: Ipv4Addr) -> Self {
313        self.config.broadcast_address = addr;
314        self
315    }
316
317    /// Set the object database (transfers ownership).
318    pub fn database(mut self, db: ObjectDatabase) -> Self {
319        self.db = db;
320        self
321    }
322
323    /// Register one explicit unicast route for a Device recipient.
324    pub fn device_binding(mut self, binding: DeviceBinding) -> Result<Self, Error> {
325        register_configured_binding(&mut self.configured_device_bindings, binding)?;
326        Ok(self)
327    }
328
329    /// Set the password required for ReinitializeDevice requests.
330    pub fn reinit_password(mut self, password: impl Into<String>) -> Self {
331        self.config.reinit_password = Some(password.into());
332        self
333    }
334
335    /// Set the policy that authorizes inbound LifeSafetyOperation requests.
336    pub fn life_safety_operation_authorizer<F>(mut self, authorizer: F) -> Self
337    where
338        F: Fn(&LifeSafetyOperationAuthorizationContext) -> bool + Send + Sync + 'static,
339    {
340        self.config.life_safety_operation_authorizer = Some(Arc::new(authorizer));
341        self
342    }
343
344    /// Select the only local Audit Log that receives authorized notifications.
345    pub fn audit_notification_sink(mut self, sink: ObjectIdentifier) -> Self {
346        self.config.audit_notification_sink = Some(sink);
347        self
348    }
349
350    /// Set the fail-closed ConfirmedAuditNotification authorization policy.
351    pub fn audit_notification_authorizer<F>(mut self, authorizer: F) -> Self
352    where
353        F: Fn(&AuditNotificationAuthorizationContext) -> bool + Send + Sync + 'static,
354    {
355        self.config.audit_notification_authorizer = Some(Arc::new(authorizer));
356        self
357    }
358
359    /// Set the fail-closed UnconfirmedAuditNotification authorization policy.
360    pub fn unconfirmed_audit_notification_authorizer<F>(mut self, authorizer: F) -> Self
361    where
362        F: Fn(&UnconfirmedAuditNotificationAuthorizationContext) -> bool + Send + Sync + 'static,
363    {
364        self.config.unconfirmed_audit_notification_authorizer = Some(Arc::new(authorizer));
365        self
366    }
367
368    /// Enable periodic fault detection / reliability evaluation.
369    ///
370    /// When enabled, every object's opt-in reliability hook runs every 10
371    /// seconds; the default hook is a no-op.
372    ///
373    /// Reliability evaluation only; Event Enrollment evaluation is configured
374    /// by [`enable_event_enrollment`](Self::enable_event_enrollment).
375    pub fn enable_fault_detection(mut self, enabled: bool) -> Self {
376        self.config.enable_fault_detection = enabled;
377        self
378    }
379
380    /// Enable periodic Event Enrollment evaluation (default `true`).
381    pub fn enable_event_enrollment(mut self, enabled: bool) -> Self {
382        self.config.enable_event_enrollment = enabled;
383        self
384    }
385
386    /// Set the interval in seconds between Event Enrollment evaluation passes
387    /// (default 10).
388    pub fn event_enrollment_interval_secs(mut self, secs: u64) -> Self {
389        self.config.event_enrollment_interval_secs = secs;
390        self
391    }
392
393    /// Set the segmentation support this device advertises and enforces.
394    ///
395    /// The dispatch loop honors the advertisement (Clause 5.4.5.1): inbound
396    /// segmented requests are reassembled only under `BOTH` or `RECEIVE`, and
397    /// draw a SEGMENTATION_NOT_SUPPORTED Abort otherwise. The default is
398    /// `NONE`.
399    pub fn segmentation_supported(mut self, segmentation: Segmentation) -> Self {
400        self.config.segmentation_supported = segmentation;
401        self
402    }
403
404    /// Set the vendor identifier advertised in I-Am responses.
405    pub fn vendor_id(mut self, id: u16) -> Self {
406        self.config.vendor_id = id;
407        self
408    }
409
410    /// Set the discovery rate-limiting and duplicate suppression policy.
411    pub fn discovery_policy(mut self, policy: DiscoveryPolicy) -> Self {
412        self.config.discovery_policy = policy;
413        self
414    }
415
416    /// Set the COV quota and notification work budget policy, checked by
417    /// [`CovPolicy::validate`] before transport startup.
418    pub fn cov_policy(mut self, policy: CovPolicy) -> Self {
419        self.config.cov_policy = policy;
420        self
421    }
422
423    /// Build and start the server, constructing a BipTransport from the config.
424    pub async fn build(self) -> Result<BACnetServer<BipTransport>, Error> {
425        let transport = BipTransport::new(
426            self.config.interface,
427            self.config.port,
428            self.config.broadcast_address,
429        );
430        BACnetServer::start_with_clock_mode_and_bindings(
431            self.config,
432            self.db,
433            transport,
434            Some(ClockConfig::default()),
435            self.configured_device_bindings,
436        )
437        .await
438    }
439}
440
441type TransportCleanup<T> = JoinHandle<(NetworkLayer<T>, Result<(), Error>)>;
442
443/// BACnet server with APDU dispatch and service handling.
444pub struct BACnetServer<T: TransportPort> {
445    target_audit: Option<Arc<audit_recipient::TargetAudit<T>>>,
446    config: ServerConfig,
447    discovery_limiter: Arc<DiscoveryLimiter>,
448    #[allow(dead_code)] // Retained with the server, including direct dispatch tests.
449    time_sync_limiter: Arc<TimeSyncLimiter>,
450    /// Server-owned clock controller; absent in explicit clockless mode.
451    _clock: Option<Arc<ServerClock>>,
452    /// Shared network layer (also held by dispatch task; read by
453    /// [`write_local`](Self::write_local) for post-write COV/event sends).
454    network: Option<Arc<NetworkLayer<T>>>,
455    broadcaster: Arc<broadcaster::BroadcasterState<T>>,
456    transport_cleanup: Option<TransportCleanup<T>>,
457    transport_cleanup_error: Option<String>,
458    network_number_task: Option<JoinHandle<()>>,
459    /// Shared object database.
460    db: Arc<RwLock<ObjectDatabase>>,
461    /// COV subscription table (also held by dispatch task; read by
462    /// [`write_local`](Self::write_local) to fire post-write notifications).
463    cov_table: Arc<RwLock<CovSubscriptionTable>>,
464    cov_counters: Arc<crate::cov::AtomicCovCounters>,
465    /// Channels for routing segmented-send events to in-progress segmented sends.
466    #[allow(dead_code)]
467    seg_ack_senders: Arc<segmented_send::SegmentedSendRegistry>,
468    /// Permits that cap live segmented response sender tasks, including
469    /// cancelled senders that have not yet exited a transport send.
470    #[allow(dead_code)]
471    seg_send_permits: Arc<Semaphore>,
472    /// Operational cap of 255 concurrent confirmed COV notification workers.
473    /// Invoke-ID ownership is handled by `notification_transactions`.
474    cov_in_flight: Arc<Semaphore>,
475    /// Bounded DNET-to-router cache learned from admitted notification terminals.
476    learned_routers: Arc<Mutex<LearnedRouterCache>>,
477    /// Invoke-ID ownership and terminal admission for confirmed notifications.
478    notification_transactions: Arc<NotificationTransactions>,
479    /// Server-lifetime exact inbound ConfirmedRequest duplicate state.
480    #[allow(dead_code)]
481    confirmed_request_tracker: Arc<ConfirmedRequestTracker>,
482    /// Shared configured and passively observed Device recipient authority.
483    device_bindings: Arc<RwLock<DeviceBindingTable>>,
484    /// DeviceCommunicationControl state; only `dcc_timer` changes it.
485    comm_state: Arc<CommState>,
486    /// DCC timer owner and replacement/expiry serialization boundary.
487    /// Valid replacement and explicit stop abort and join before clearing it.
488    dcc_timer: Arc<Mutex<crate::server::dcc_timer::TimerSlot>>,
489    dcc_outcomes: Arc<dcc_outcomes::DccOutcomes>,
490    /// Lifetime totals of undelivered event notifications (#1142).
491    event_suppressions: Arc<event_suppression::EventSuppressions>,
492    mutation_decisions: Arc<crate::mutation::MutationDecisions>,
493    dispatch_task: Option<JoinHandle<()>>,
494    request_tasks: Arc<request_tasks::RequestTasks>,
495    cov_purge_task: Option<JoinHandle<()>>,
496    fault_detection_task: Option<JoinHandle<()>>,
497    event_enrollment_task: Option<JoinHandle<()>>,
498    trend_log_task: Option<JoinHandle<()>>,
499    schedule_tick_task: Option<JoinHandle<()>>,
500    /// One-second `Time_Delay` confirmation task for intrinsic reporting.
501    intrinsic_reporting_task: Option<JoinHandle<()>>,
502    /// Monotonic object-operation task: Binary Lighting Output
503    /// WARN_OFF/WARN_RELINQUISH egress and Access Door pulse relock (#1073).
504    binary_lighting_operation_task: Option<JoinHandle<()>>,
505    /// Follow-up fanout after acknowledged confirmed COV reports (#896).
506    cov_revisit_task: Option<JoinHandle<()>>,
507    local_mac: MacAddr,
508}
509
510/// Cloneable handle for unsolicited I-Am announcements while its server runs.
511///
512/// At most 32 local broadcasts may be in flight. Admission fails immediately
513/// at capacity or after shutdown begins. Admitted sends belong to the server:
514/// cancelling their caller only drops its waiter; stop cancels and joins them.
515/// Retaining this handle does not retain the transport after shutdown.
516pub struct IAmBroadcaster<T: TransportPort> {
517    state: std::sync::Weak<broadcaster::BroadcasterState<T>>,
518}
519
520impl<T: TransportPort> Clone for IAmBroadcaster<T> {
521    fn clone(&self) -> Self {
522        Self {
523            state: self.state.clone(),
524        }
525    }
526}
527
528impl BACnetServer<BipTransport> {
529    /// Create a BIP-specific builder with interface/port/broadcast fields.
530    pub fn bip_builder() -> BipServerBuilder {
531        BipServerBuilder {
532            config: ServerConfig::default(),
533            db: ObjectDatabase::new(),
534            configured_device_bindings: Vec::new(),
535        }
536    }
537}
538
539mod clock;
540mod command_runs;
541mod time_sync_policy;
542mod write_group;
543#[cfg(test)]
544pub(crate) use clock::clocked_test_database;
545pub use clock::ClockConfig;
546use clock::ServerClock;
547use time_sync_policy::{request_limiters, TimeSyncLimiter};
548pub use time_sync_policy::{
549    TimeSyncPolicy, TimeSyncRateLimit, TimeSyncSource, TimeSyncSourceRestriction,
550};
551mod binary_lighting_lifecycle;
552#[cfg(test)]
553mod confirmed_issuance_tests;
554mod confirmed_request_tracker;
555mod cov_clock;
556mod cov_encoding;
557mod cov_fanout;
558mod cov_notifications;
559mod cov_notify_context;
560mod cov_snapshot;
561mod dcc_disable_rate;
562pub(crate) mod dcc_outcomes;
563mod dcc_policy;
564mod lso_replay;
565pub use dcc_disable_rate::DccDisableRateLimit;
566mod dcc_timer;
567pub use dcc_outcomes::DccOutcomeCounters;
568pub use dcc_policy::{DccPolicy, DccSource, DccSourceRestriction};
569use dcc_timer::CommState;
570pub use dcc_timer::DccState;
571mod binding_probes;
572mod device_bindings;
573mod discovery;
574#[doc(hidden)]
575pub use discovery::iam_request_for as discovery_iam_for_test;
576pub use discovery::{DiscoveryCounters, DiscoveryPolicy};
577pub(crate) use discovery::{DiscoveryLimiter, PreCheckDecision, WhoHasTarget};
578mod audit_log_purge;
579mod dispatch;
580mod durable_writes;
581mod event_delivery;
582mod event_enrollment_lifecycle;
583mod event_forwarding;
584pub use event_forwarding::MAX_FORWARDED_DESTINATIONS;
585mod event_forwarding_repeats;
586mod event_message_policy;
587pub(crate) mod event_notification_payload;
588mod event_notifications;
589mod event_recipient_route;
590mod event_send;
591mod event_suppression;
592pub use event_suppression::EventNotificationCounters;
593mod received_event_log;
594pub use received_event_log::{RECEIVED_EVENT_LOG_RATE, RECEIVED_EVENT_LOG_SOURCES};
595mod confirmed_answer;
596pub(crate) mod event_timestamp;
597mod handles;
598mod learned_router_cache;
599mod lifecycle;
600mod local_writes;
601mod network_port;
602#[cfg(test)]
603mod network_port_tests;
604pub use confirmed_answer::{CovAckResult, Refusal};
605mod remote_writes;
606pub(crate) use remote_writes::RemoteRequestError;
607mod notification_transactions;
608#[doc(hidden)]
609pub use notification_transactions::{
610    canonical_direct_peer as __endpoint_canonical_direct_peer,
611    canonical_routed_peer as __endpoint_canonical_routed_peer,
612    run_notification_worker as __endpoint_run_notification_worker,
613    AuditFailureContext as __endpoint_AuditFailureContext,
614    AuditFailureQueue as __endpoint_AuditFailureQueue,
615    AuditFailureTicket as __endpoint_AuditFailureTicket,
616    NotificationOperation as __endpoint_NotificationOperation,
617    NotificationReserveError as __endpoint_NotificationReserveError,
618    NotificationTransactions as __endpoint_NotificationTransactions,
619    NotificationWorkerResult as __endpoint_NotificationWorkerResult,
620};
621mod requests;
622#[doc(hidden)]
623pub use requests::endpoint_responder::EndpointResponder as __endpoint_EndpointResponder;
624#[cfg(feature = "sc-tls")]
625mod sc_builder;
626#[cfg(test)]
627pub(crate) use requests::{EXECUTED_CONFIRMED, EXECUTED_UNCONFIRMED};
628#[cfg(test)]
629mod audit_notification_tests;
630#[cfg(test)]
631mod unconfirmed_audit_notification_tests;
632#[cfg(feature = "sc-tls")]
633pub use sc_builder::ScServerBuilder;
634mod responses;
635mod segmentation;
636mod segmented_receive;
637mod segmented_send;
638pub(crate) use segmented_send::*;
639mod request_admission;
640mod rpm_budget;
641pub use rpm_budget::ReadPropertyMultipleBudget;
642mod alarm_summary_budget;
643pub use alarm_summary_budget::GetAlarmSummaryBudget;
644mod atomic_read_file_budget;
645mod atomic_write_file_budget;
646mod read_range_budget;
647pub use read_range_budget::ReadRangeBudget;
648mod event_information_budget;
649pub use event_information_budget::GetEventInformationBudget;
650mod enrollment_summary_budget;
651pub use atomic_read_file_budget::AtomicReadFileBudget;
652pub use atomic_write_file_budget::AtomicWriteFileBudget;
653pub use enrollment_summary_budget::GetEnrollmentSummaryBudget;
654#[cfg(test)]
655mod atomic_read_file_tests;
656#[cfg(test)]
657mod atomic_write_file_tests;
658#[cfg(test)]
659mod enrollment_summary_tests;
660mod request_peer;
661mod request_services;
662mod request_tasks;
663pub use request_admission::{RequestAdmissionCounters, RequestAdmissionPolicy};
664mod shutdown;
665
666#[cfg(test)]
667mod access_door_event_tests;
668#[cfg(test)]
669mod access_door_pulse_task_tests;
670#[cfg(test)]
671mod access_zone_event_tests;
672#[cfg(test)]
673mod acknowledge_alarm_tests;
674#[cfg(test)]
675mod active_cov_subscriptions_tests;
676#[cfg(test)]
677mod audit_log_query_tests;
678#[cfg(test)]
679mod averaging_reference_write_tests;
680#[cfg(test)]
681mod averaging_sample_tests;
682#[cfg(test)]
683mod binary_lighting_task_tests;
684#[cfg(test)]
685mod channel_member_concurrency_tests;
686#[cfg(test)]
687mod channel_reliability_tests;
688#[cfg(test)]
689mod channel_remote_datatype_tests;
690#[cfg(test)]
691mod channel_remote_write_tests;
692#[cfg(test)]
693mod channel_run_tests;
694#[cfg(test)]
695mod channel_wire_tests;
696#[cfg(test)]
697mod command_action_run_tests;
698#[cfg(test)]
699mod command_action_wire_tests;
700#[cfg(test)]
701mod command_remote_write_tests;
702#[cfg(test)]
703mod command_run_cancel_tests;
704#[cfg(test)]
705mod command_run_stop_tests;
706#[cfg(test)]
707mod confirmed_broadcast_tests;
708#[cfg(test)]
709mod cov_background_tests;
710#[cfg(test)]
711mod cov_budget_tests;
712#[cfg(test)]
713mod cov_confirmed_baseline_tests;
714#[cfg(test)]
715mod cov_confirmed_context_tests;
716#[cfg(test)]
717mod cov_multiple_admission_tests;
718#[cfg(test)]
719mod cov_notifications_tests;
720#[cfg(test)]
721mod cov_quota_tests;
722#[cfg(test)]
723mod cov_timed_capture_order_tests;
724#[cfg(test)]
725mod cov_timed_chunk_tests;
726#[cfg(test)]
727mod cov_timed_deadline_tests;
728#[cfg(test)]
729mod cov_timed_multiple_tests;
730#[cfg(test)]
731mod cov_timed_producer_tests;
732#[cfg(test)]
733mod cov_timed_value_split_tests;
734#[cfg(test)]
735mod cov_untimed_split_tests;
736#[cfg(test)]
737mod cov_wire_test_support;
738#[cfg(test)]
739mod dcc_event_detection_tests;
740#[cfg(test)]
741mod device_bindings_tests;
742#[cfg(test)]
743mod device_recipient_routing_tests;
744#[cfg(all(test, feature = "sc-tls"))]
745mod direct_principal_tests;
746#[cfg(test)]
747mod discovery_dcc_tests;
748#[cfg(test)]
749mod discovery_tests;
750#[cfg(test)]
751mod event_confirmed_routing_tests;
752#[cfg(test)]
753mod event_enable_distribution_tests;
754#[cfg(test)]
755mod event_enrollment_task_tests;
756#[cfg(test)]
757mod event_forwarding_bounds_tests;
758#[cfg(test)]
759mod event_forwarding_cap_tests;
760#[cfg(test)]
761mod event_forwarding_origin_tests;
762#[cfg(test)]
763mod event_forwarding_rule_tests;
764#[cfg(test)]
765mod event_forwarding_tests;
766#[cfg(test)]
767mod event_log_notification_tests;
768#[cfg(test)]
769mod event_network_priority_tests;
770#[cfg(test)]
771mod event_notifications_tests;
772#[cfg(test)]
773mod event_recipient_routing_tests;
774#[cfg(test)]
775mod life_safety_application_tests;
776#[cfg(test)]
777mod life_safety_cov_tests;
778#[cfg(test)]
779mod life_safety_operation_replay_tests;
780#[cfg(test)]
781mod life_safety_operation_tests;
782#[cfg(test)]
783mod lighting_command_member_tests;
784#[cfg(test)]
785mod lighting_engine_task_tests;
786#[cfg(test)]
787mod lighting_present_value_tests;
788#[cfg(test)]
789mod list_element_event_tests;
790#[cfg(test)]
791mod local_network_number_tests;
792#[cfg(test)]
793mod local_network_outbound_tests;
794#[cfg(test)]
795mod log_reference_resize_tests;
796#[cfg(test)]
797mod loop_controlled_variable_tests;
798#[cfg(test)]
799mod loop_cov_tests;
800#[cfg(test)]
801mod mistagged_request_wire_tests;
802#[cfg(test)]
803mod noncommandable_null_cov_tests;
804#[cfg(test)]
805mod notification_dcc_tests;
806#[cfg(test)]
807mod notification_transactions_tests;
808#[cfg(test)]
809mod pulse_converter_cov_tests;
810#[cfg(test)]
811mod pulse_converter_reference_tests;
812#[cfg(test)]
813mod rb07_provenance_tests;
814#[cfg(test)]
815mod reference_write_wire_tests;
816#[cfg(test)]
817mod remote_write_answer_tests;
818#[cfg(test)]
819mod remote_write_discovery_tests;
820#[cfg(test)]
821mod run_cycle_tests;
822#[cfg(test)]
823mod schedule_create_retry_tests;
824#[cfg(test)]
825mod schedule_reference_list_tests;
826#[cfg(test)]
827mod schedule_reference_write_tests;
828#[cfg(test)]
829mod schedule_write_tests;
830#[cfg(test)]
831mod segmentation_tests;
832#[cfg(test)]
833mod staging_cov_tests;
834#[cfg(test)]
835mod staging_reference_write_tests;
836#[cfg(test)]
837mod subscribed_recipients_expiry_tests;
838#[cfg(test)]
839mod table_13_1_cov_tests;
840#[cfg(test)]
841pub(crate) mod test_forwarder;
842#[cfg(test)]
843pub(crate) mod test_transport;
844#[cfg(test)]
845mod tests;
846#[cfg(test)]
847mod trailing_octet_wire_tests;
848#[cfg(test)]
849mod truncated_request_wire_tests;
850#[cfg(test)]
851mod value_cov_increment_tests;
852
853impl<T: TransportPort + 'static> BACnetServer<T> {
854    /// Start a server builder for a caller-supplied transport type, with default configuration and
855    /// an empty object database.
856    pub fn generic_builder() -> ServerBuilder<T> {
857        ServerBuilder {
858            config: ServerConfig::default(),
859            db: ObjectDatabase::new(),
860            transport: None,
861            configured_device_bindings: Vec::new(),
862        }
863    }
864
865    /// Get a snapshot of discovery rate-limiting counters.
866    pub fn discovery_counters(&self) -> DiscoveryCounters {
867        self.discovery_limiter.counters()
868    }
869
870    /// Get a snapshot of COV operational and telemetry counters.
871    pub fn cov_counters(&self) -> CovCounters {
872        self.cov_counters.snapshot()
873    }
874
875    /// Sample lifetime mutation authorization decisions, including after `stop()`.
876    /// Counts decisions, not completed handlers or response delivery; see
877    /// [`MutationDecisionCounters`](crate::mutation::MutationDecisionCounters).
878    pub fn mutation_decision_counters(&self) -> crate::mutation::MutationDecisionCounters {
879        self.mutation_decisions.snapshot()
880    }
881
882    /// Remove COV subscriptions using this current immediate MAC and routed source.
883    /// Cleanup of an obsolete router leaves subscriptions renewed onto another route.
884    /// Only removed entries release their contribution to the shared quota group.
885    pub async fn remove_peer_subscriptions(
886        &self,
887        mac: &[u8],
888        network: Option<&NpduAddress>,
889    ) -> usize {
890        let mut table = self.cov_table.write().await;
891        table.remove_peer_subscriptions(mac, network)
892    }
893}
894
895#[cfg(test)]
896mod cov_identity_completion_tests;
897
898mod broadcaster;
bacnet_server/server/mod.rs

bacnet_server/server/
mod.rs
