1use super::*;
2
3impl<T: TransportPort + 'static> BACnetServer<T> {
4 pub async fn start(
5 mut config: ServerConfig,
6 db: ObjectDatabase,
7 transport: T,
8 ) -> Result<Self, Error> {
9 let transport_max = transport.max_apdu_length() as u32;
10 config.max_apdu_length = config.max_apdu_length.min(transport_max);
11 let max_apdu = u16::try_from(config.max_apdu_length).map_err(|_| {
12 Error::Encoding(format!(
13 "invalid max_apdu_length {}; expected one of 50, 128, 206, 480, 1024, 1476",
14 config.max_apdu_length
15 ))
16 })?;
17 validate_max_apdu_length(max_apdu)?;
18
19 if config.vendor_id == 0 {
20 warn!("vendor_id is 0 (ASHRAE reserved); set a valid vendor ID for production use");
21 }
22
23 let mut network = NetworkLayer::new(transport);
24 let apdu_rx = network.start().await?;
25 let local_mac = MacAddr::from_slice(network.local_mac());
26
27 let network = Arc::new(network);
28 let db = Arc::new(RwLock::new(db));
29 let cov_table = Arc::new(RwLock::new(CovSubscriptionTable::new()));
30 let seg_ack_senders: Arc<Mutex<HashMap<SegKey, mpsc::Sender<SegmentAckPdu>>>> =
31 Arc::new(Mutex::new(HashMap::new()));
32
33 let cov_in_flight = Arc::new(Semaphore::new(255));
34 let server_tsm = Arc::new(Mutex::new(ServerTsm::new()));
35 let comm_state = Arc::new(AtomicU8::new(0)); let dcc_timer: Arc<Mutex<Option<JoinHandle<()>>>> = Arc::new(Mutex::new(None));
37
38 let network_dispatch = Arc::clone(&network);
39 let db_dispatch = Arc::clone(&db);
40 let cov_dispatch = Arc::clone(&cov_table);
41 let seg_ack_dispatch = Arc::clone(&seg_ack_senders);
42 let cov_in_flight_dispatch = Arc::clone(&cov_in_flight);
43 let server_tsm_dispatch = Arc::clone(&server_tsm);
44 let comm_state_dispatch = Arc::clone(&comm_state);
45 let dcc_timer_dispatch = Arc::clone(&dcc_timer);
46 let config_dispatch = Arc::new(config.clone());
47
48 let dispatch_task = tokio::spawn(async move {
49 let mut apdu_rx = apdu_rx;
50 let mut seg_receivers: HashMap<SegKey, SegmentedRequestState> = HashMap::new();
51
52 while let Some(received) = apdu_rx.recv().await {
53 let now = Instant::now();
54 seg_receivers.retain(|_key, state| {
55 now.duration_since(state.last_activity) < SEG_RECEIVER_TIMEOUT
56 });
57
58 match apdu::decode_apdu(received.apdu.clone()) {
59 Ok(decoded) => {
60 let source_mac = received.source_mac.clone();
61 let mut received = Some(received);
62 let handled = if let Apdu::ConfirmedRequest(ref req) = decoded {
63 if req.segmented {
64 let seq = req.sequence_number.unwrap_or(0);
65 let key: SegKey = (source_mac.clone(), req.invoke_id);
66
67 let mut ack_to_send: Option<SegmentAckPdu> = None;
68 let mut final_total: Option<usize> = None;
69
70 if seq == 0 {
71 let proposed_window_size =
72 req.proposed_window_size.unwrap_or(0);
73 if !(1..=127).contains(&proposed_window_size) {
74 warn!(
75 invoke_id = req.invoke_id,
76 proposed_window_size,
77 "Rejecting segmented request with invalid proposed window size"
78 );
79 let abort_pdu = Apdu::Abort(AbortPdu {
80 sent_by_server: true,
81 invoke_id: req.invoke_id,
82 abort_reason: AbortReason::WINDOW_SIZE_OUT_OF_RANGE,
83 });
84 let mut abort_buf = BytesMut::new();
85 encode_apdu(&mut abort_buf, &abort_pdu)
86 .expect("valid APDU encoding");
87 let _ = network_dispatch
88 .send_apdu(
89 &abort_buf,
90 &source_mac,
91 false,
92 NetworkPriority::NORMAL,
93 )
94 .await;
95 continue;
96 }
97
98 if !seg_receivers.contains_key(&key)
99 && seg_receivers.len() >= MAX_SEG_RECEIVERS
100 {
101 let abort_pdu = Apdu::Abort(AbortPdu {
102 sent_by_server: true,
103 invoke_id: req.invoke_id,
104 abort_reason: AbortReason::BUFFER_OVERFLOW,
105 });
106 let mut abort_buf = BytesMut::new();
107 encode_apdu(&mut abort_buf, &abort_pdu)
108 .expect("valid APDU encoding");
109 let _ = network_dispatch
110 .send_apdu(
111 &abort_buf,
112 &source_mac,
113 false,
114 NetworkPriority::NORMAL,
115 )
116 .await;
117 continue;
118 }
119
120 let mut receiver = SegmentReceiver::new();
121 if let Err(e) =
122 receiver.receive(seq, req.service_request.clone())
123 {
124 warn!(error = %e, "Rejecting oversized segment");
125 continue;
126 }
127 let actual_window_size = proposed_window_size;
128 let mut state = SegmentedRequestState {
129 receiver,
130 first_req: req.clone(),
131 last_activity: Instant::now(),
132 expected_seq: 1,
133 last_acked_seq: 0,
134 window_pos: 1,
135 actual_window_size,
136 };
137 let should_ack =
138 !req.more_follows || state.window_pos >= actual_window_size;
139 if should_ack {
140 state.window_pos = 0;
141 ack_to_send = Some(SegmentAckPdu {
142 negative_ack: false,
143 sent_by_server: true,
144 invoke_id: req.invoke_id,
145 sequence_number: seq,
146 actual_window_size,
147 });
148 }
149 if !req.more_follows {
150 final_total = Some(1);
151 }
152 seg_receivers.insert(key.clone(), state);
153 } else if let Some(state) = seg_receivers.get_mut(&key) {
154 state.last_activity = Instant::now();
155 if seq != state.expected_seq {
156 warn!(
157 invoke_id = req.invoke_id,
158 expected = state.expected_seq,
159 received = seq,
160 "Segment gap detected, sending negative SegmentAck"
161 );
162 ack_to_send = Some(SegmentAckPdu {
163 negative_ack: true,
164 sent_by_server: true,
165 invoke_id: req.invoke_id,
166 sequence_number: state.last_acked_seq,
167 actual_window_size: state.actual_window_size,
168 });
169 } else {
170 if let Err(e) =
171 state.receiver.receive(seq, req.service_request.clone())
172 {
173 warn!(error = %e, "Rejecting oversized segment");
174 continue;
175 }
176 state.expected_seq = seq.wrapping_add(1);
177 state.last_acked_seq = seq;
178 state.window_pos += 1;
179 let should_ack = !req.more_follows
180 || state.window_pos >= state.actual_window_size;
181 if should_ack {
182 state.window_pos = 0;
183 ack_to_send = Some(SegmentAckPdu {
184 negative_ack: false,
185 sent_by_server: true,
186 invoke_id: req.invoke_id,
187 sequence_number: seq,
188 actual_window_size: state.actual_window_size,
189 });
190 }
191 if !req.more_follows {
192 final_total = Some(seq as usize + 1);
193 }
194 }
195 } else {
196 warn!(
197 invoke_id = req.invoke_id,
198 seq = seq,
199 "Received non-initial segment without prior segment 0, aborting"
200 );
201 let abort_pdu = Apdu::Abort(AbortPdu {
202 sent_by_server: true,
203 invoke_id: req.invoke_id,
204 abort_reason: AbortReason::INVALID_APDU_IN_THIS_STATE,
205 });
206 let mut abort_buf = BytesMut::new();
207 encode_apdu(&mut abort_buf, &abort_pdu)
208 .expect("valid APDU encoding");
209 let _ = network_dispatch
210 .send_apdu(
211 &abort_buf,
212 &source_mac,
213 false,
214 NetworkPriority::NORMAL,
215 )
216 .await;
217 continue;
218 }
219
220 if let Some(seg_ack) = ack_to_send {
221 let seg_ack = Apdu::SegmentAck(seg_ack);
222 let mut ack_buf = BytesMut::new();
223 encode_apdu(&mut ack_buf, &seg_ack)
224 .expect("valid APDU encoding");
225 if let Err(e) = network_dispatch
226 .send_apdu(
227 &ack_buf,
228 &source_mac,
229 false,
230 NetworkPriority::NORMAL,
231 )
232 .await
233 {
234 warn!(
235 error = %e,
236 "Failed to send SegmentAck for segmented request"
237 );
238 }
239 }
240
241 if let Some(total) = final_total {
242 if let Some(state) = seg_receivers.remove(&key) {
243 match state.receiver.reassemble(total) {
244 Ok(full_data) => {
245 let reassembled =
246 bacnet_encoding::apdu::ConfirmedRequest {
247 segmented: false,
248 more_follows: false,
249 sequence_number: None,
250 proposed_window_size: None,
251 service_request: Bytes::from(full_data),
252 invoke_id: state.first_req.invoke_id,
253 service_choice: state
254 .first_req
255 .service_choice,
256 max_apdu_length: state
257 .first_req
258 .max_apdu_length,
259 segmented_response_accepted: state
260 .first_req
261 .segmented_response_accepted,
262 max_segments: state.first_req.max_segments,
263 };
264 debug!(
265 invoke_id = reassembled.invoke_id,
266 segments = total,
267 payload_len = reassembled.service_request.len(),
268 "Reassembled segmented ConfirmedRequest"
269 );
270 Self::dispatch(
271 &db_dispatch,
272 &network_dispatch,
273 &cov_dispatch,
274 &seg_ack_dispatch,
275 &cov_in_flight_dispatch,
276 &server_tsm_dispatch,
277 &comm_state_dispatch,
278 &dcc_timer_dispatch,
279 &config_dispatch,
280 &source_mac,
281 Apdu::ConfirmedRequest(reassembled),
282 received.take().unwrap_or_else(|| {
283 warn!("received consumed twice - using empty fallback");
284 bacnet_network::layer::ReceivedApdu {
285 apdu: bytes::Bytes::new(),
286 source_mac: bacnet_types::MacAddr::new(),
287 source_network: None,
288 data_attributes: Vec::new(),
289 reply_tx: None,
290 }
291 }),
292 )
293 .await;
294 }
295 Err(e) => {
296 warn!(
297 error = %e,
298 "Failed to reassemble segmented request"
299 );
300 }
301 }
302 }
303 }
304
305 true
306 } else {
307 false
308 }
309 } else {
310 false
311 };
312
313 if !handled {
314 Self::dispatch(
315 &db_dispatch,
316 &network_dispatch,
317 &cov_dispatch,
318 &seg_ack_dispatch,
319 &cov_in_flight_dispatch,
320 &server_tsm_dispatch,
321 &comm_state_dispatch,
322 &dcc_timer_dispatch,
323 &config_dispatch,
324 &source_mac,
325 decoded,
326 received.take().unwrap_or_else(|| {
327 warn!("received consumed twice — using empty fallback");
328 bacnet_network::layer::ReceivedApdu {
329 apdu: bytes::Bytes::new(),
330 source_mac: bacnet_types::MacAddr::new(),
331 source_network: None,
332 data_attributes: Vec::new(),
333 reply_tx: None,
334 }
335 }),
336 )
337 .await;
338 }
339 }
340 Err(e) => {
341 warn!(error = %e, "Server failed to decode received APDU");
342 }
343 }
344 }
345 });
346
347 let cov_table_for_purge = Arc::clone(&cov_table);
348 let cov_purge_task = tokio::spawn(async move {
349 let mut interval = tokio::time::interval(Duration::from_secs(30));
350 loop {
351 interval.tick().await;
352 let mut table = cov_table_for_purge.write().await;
353 let purged = table.purge_expired();
354 if purged > 0 {
355 debug!(purged, "Purged expired COV subscriptions");
356 }
357 }
358 });
359
360 let fault_detection_task = if config.enable_fault_detection {
361 let db_fault = Arc::clone(&db);
362 Some(tokio::spawn(async move {
363 let detector = crate::fault_detection::FaultDetector::default();
364 let mut interval = tokio::time::interval(Duration::from_secs(10));
365 loop {
366 interval.tick().await;
367 let mut db_guard = db_fault.write().await;
368 let changes = detector.evaluate(&mut db_guard);
369 for change in &changes {
370 debug!(
371 object = %change.object_id,
372 old = change.old_reliability,
373 new = change.new_reliability,
374 "Fault detection: reliability changed"
375 );
376 }
377 }
378 }))
379 } else {
380 None
381 };
382
383 let event_enrollment_task = if config.enable_fault_detection {
384 let db_ee = Arc::clone(&db);
385 Some(tokio::spawn(async move {
386 let mut interval = tokio::time::interval(Duration::from_secs(10));
387 loop {
388 interval.tick().await;
389 let mut db_guard = db_ee.write().await;
390 let transitions =
391 crate::event_enrollment::evaluate_event_enrollments(&mut db_guard);
392 for t in &transitions {
393 debug!(
394 enrollment = %t.enrollment_oid,
395 monitored = %t.monitored_oid,
396 from = ?t.change.from,
397 to = ?t.change.to,
398 "Event enrollment: state changed"
399 );
400 }
401 }
402 }))
403 } else {
404 None
405 };
406
407 let db_trend = Arc::clone(&db);
408 let trend_log_state: crate::trend_log::TrendLogState =
409 Arc::new(tokio::sync::Mutex::new(std::collections::HashMap::new()));
410 let trend_log_task = Some(tokio::spawn(async move {
411 let mut interval = tokio::time::interval(Duration::from_secs(1));
412 loop {
413 interval.tick().await;
414 crate::trend_log::poll_trend_logs(&db_trend, &trend_log_state).await;
415 }
416 }));
417
418 let db_schedule = Arc::clone(&db);
419 let schedule_tick_task = Some(tokio::spawn(async move {
420 let mut interval = tokio::time::interval(Duration::from_secs(60));
421 loop {
422 interval.tick().await;
423 crate::schedule::tick_schedules(&db_schedule, 0).await;
425 }
426 }));
427
428 Ok(Self {
429 config,
430 network,
431 db,
432 cov_table,
433 seg_ack_senders,
434 cov_in_flight,
435 server_tsm,
436 comm_state,
437 dcc_timer,
438 dispatch_task: Some(dispatch_task),
439 cov_purge_task: Some(cov_purge_task),
440 fault_detection_task,
441 event_enrollment_task,
442 trend_log_task,
443 schedule_tick_task,
444 local_mac,
445 })
446 }
447
448 pub fn local_mac(&self) -> &[u8] {
450 &self.local_mac
451 }
452
453 pub fn database(&self) -> &Arc<RwLock<ObjectDatabase>> {
455 &self.db
456 }
457
458 pub fn i_am_broadcaster(&self) -> IAmBroadcaster<T> {
460 IAmBroadcaster {
461 config: self.config.clone(),
462 network: Arc::clone(&self.network),
463 db: Arc::clone(&self.db),
464 }
465 }
466
467 pub fn comm_state(&self) -> u8 {
471 self.comm_state.load(Ordering::Acquire)
472 }
473
474 pub async fn generate_pics(&self, pics_config: &crate::pics::PicsConfig) -> crate::pics::Pics {
479 let db = self.db.read().await;
480 crate::pics::PicsGenerator::new(&db, &self.config, pics_config).generate()
481 }
482
483 pub async fn broadcast_i_am(&self) -> Result<(), Error> {
485 broadcast_i_am_from(&self.config, &self.db, &self.network).await
486 }
487
488 pub async fn stop(&mut self) -> Result<(), Error> {
490 if let Some(task) = self.fault_detection_task.take() {
491 task.abort();
492 let _ = task.await;
493 }
494 if let Some(task) = self.event_enrollment_task.take() {
495 task.abort();
496 let _ = task.await;
497 }
498 if let Some(task) = self.trend_log_task.take() {
499 task.abort();
500 let _ = task.await;
501 }
502 if let Some(task) = self.schedule_tick_task.take() {
503 task.abort();
504 let _ = task.await;
505 }
506 if let Some(task) = self.cov_purge_task.take() {
507 task.abort();
508 let _ = task.await;
509 }
510 if let Some(task) = self.dispatch_task.take() {
511 task.abort();
512 let _ = task.await;
513 }
514 Ok(())
515 }
516}
517
518impl<T: TransportPort + 'static> IAmBroadcaster<T> {
519 pub async fn broadcast_i_am(&self) -> Result<(), Error> {
521 broadcast_i_am_from(&self.config, &self.db, &self.network).await
522 }
523}
524
525async fn broadcast_i_am_from<T: TransportPort + 'static>(
526 config: &ServerConfig,
527 db: &Arc<RwLock<ObjectDatabase>>,
528 network: &Arc<NetworkLayer<T>>,
529) -> Result<(), Error> {
530 let device_oid = {
531 let db = db.read().await;
532 db.list_objects()
533 .into_iter()
534 .find(|oid| oid.object_type() == ObjectType::DEVICE)
535 .ok_or_else(|| Error::Encoding("no Device object in database".into()))?
536 };
537
538 let i_am = IAmRequest {
539 object_identifier: device_oid,
540 max_apdu_length: config.max_apdu_length,
541 segmentation_supported: config.segmentation_supported,
542 vendor_id: config.vendor_id,
543 };
544
545 let mut service_buf = BytesMut::new();
546 i_am.encode(&mut service_buf);
547
548 let pdu = Apdu::UnconfirmedRequest(UnconfirmedRequestPdu {
549 service_choice: UnconfirmedServiceChoice::I_AM,
550 service_request: service_buf.freeze(),
551 });
552
553 let mut buf = BytesMut::new();
554 encode_apdu(&mut buf, &pdu)?;
555
556 network
557 .broadcast_apdu(&buf, false, NetworkPriority::NORMAL)
558 .await
559}