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 source_network = received.source_network.clone();
62 let mut received = Some(received);
63 let handled = if let Apdu::ConfirmedRequest(ref req) = decoded {
64 if req.segmented {
65 let seq = req.sequence_number.unwrap_or(0);
66 let key: SegKey =
67 (source_mac.clone(), source_network.clone(), req.invoke_id);
68
69 let mut ack_to_send: Option<SegmentAckPdu> = None;
70 let mut final_total: Option<usize> = None;
71
72 if seq == 0 {
73 let proposed_window_size =
74 req.proposed_window_size.unwrap_or(0);
75 if !(1..=127).contains(&proposed_window_size) {
76 warn!(
77 invoke_id = req.invoke_id,
78 proposed_window_size,
79 "Rejecting segmented request with invalid proposed window size"
80 );
81 let abort_pdu = Apdu::Abort(AbortPdu {
82 sent_by_server: true,
83 invoke_id: req.invoke_id,
84 abort_reason: AbortReason::WINDOW_SIZE_OUT_OF_RANGE,
85 });
86 let mut abort_buf = BytesMut::new();
87 encode_apdu(&mut abort_buf, &abort_pdu)
88 .expect("valid APDU encoding");
89 let _ = Self::send_confirmed_response_apdu(
90 &network_dispatch,
91 &abort_buf,
92 &source_mac,
93 source_network.as_ref(),
94 )
95 .await;
96 continue;
97 }
98
99 if !seg_receivers.contains_key(&key)
100 && seg_receivers.len() >= MAX_SEG_RECEIVERS
101 {
102 let abort_pdu = Apdu::Abort(AbortPdu {
103 sent_by_server: true,
104 invoke_id: req.invoke_id,
105 abort_reason: AbortReason::BUFFER_OVERFLOW,
106 });
107 let mut abort_buf = BytesMut::new();
108 encode_apdu(&mut abort_buf, &abort_pdu)
109 .expect("valid APDU encoding");
110 let _ = Self::send_confirmed_response_apdu(
111 &network_dispatch,
112 &abort_buf,
113 &source_mac,
114 source_network.as_ref(),
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 _ = Self::send_confirmed_response_apdu(
210 &network_dispatch,
211 &abort_buf,
212 &source_mac,
213 source_network.as_ref(),
214 )
215 .await;
216 continue;
217 }
218
219 if let Some(seg_ack) = ack_to_send {
220 let seg_ack = Apdu::SegmentAck(seg_ack);
221 let mut ack_buf = BytesMut::new();
222 encode_apdu(&mut ack_buf, &seg_ack)
223 .expect("valid APDU encoding");
224 if let Err(e) = Self::send_confirmed_response_apdu(
225 &network_dispatch,
226 &ack_buf,
227 &source_mac,
228 source_network.as_ref(),
229 )
230 .await
231 {
232 warn!(
233 error = %e,
234 "Failed to send SegmentAck for segmented request"
235 );
236 }
237 }
238
239 if let Some(total) = final_total {
240 if let Some(state) = seg_receivers.remove(&key) {
241 match state.receiver.reassemble(total) {
242 Ok(full_data) => {
243 let reassembled =
244 bacnet_encoding::apdu::ConfirmedRequest {
245 segmented: false,
246 more_follows: false,
247 sequence_number: None,
248 proposed_window_size: None,
249 service_request: Bytes::from(full_data),
250 invoke_id: state.first_req.invoke_id,
251 service_choice: state
252 .first_req
253 .service_choice,
254 max_apdu_length: state
255 .first_req
256 .max_apdu_length,
257 segmented_response_accepted: state
258 .first_req
259 .segmented_response_accepted,
260 max_segments: state.first_req.max_segments,
261 };
262 debug!(
263 invoke_id = reassembled.invoke_id,
264 segments = total,
265 payload_len = reassembled.service_request.len(),
266 "Reassembled segmented ConfirmedRequest"
267 );
268 Self::dispatch(
269 &db_dispatch,
270 &network_dispatch,
271 &cov_dispatch,
272 &seg_ack_dispatch,
273 &cov_in_flight_dispatch,
274 &server_tsm_dispatch,
275 &comm_state_dispatch,
276 &dcc_timer_dispatch,
277 &config_dispatch,
278 &source_mac,
279 Apdu::ConfirmedRequest(reassembled),
280 received.take().unwrap_or_else(|| {
281 warn!("received consumed twice - using empty fallback");
282 bacnet_network::layer::ReceivedApdu {
283 apdu: bytes::Bytes::new(),
284 source_mac: bacnet_types::MacAddr::new(),
285 source_network: None,
286 data_attributes: Vec::new(),
287 reply_tx: None,
288 }
289 }),
290 )
291 .await;
292 }
293 Err(e) => {
294 warn!(
295 error = %e,
296 "Failed to reassemble segmented request"
297 );
298 }
299 }
300 }
301 }
302
303 true
304 } else {
305 false
306 }
307 } else {
308 false
309 };
310
311 if !handled {
312 Self::dispatch(
313 &db_dispatch,
314 &network_dispatch,
315 &cov_dispatch,
316 &seg_ack_dispatch,
317 &cov_in_flight_dispatch,
318 &server_tsm_dispatch,
319 &comm_state_dispatch,
320 &dcc_timer_dispatch,
321 &config_dispatch,
322 &source_mac,
323 decoded,
324 received.take().unwrap_or_else(|| {
325 warn!("received consumed twice — using empty fallback");
326 bacnet_network::layer::ReceivedApdu {
327 apdu: bytes::Bytes::new(),
328 source_mac: bacnet_types::MacAddr::new(),
329 source_network: None,
330 data_attributes: Vec::new(),
331 reply_tx: None,
332 }
333 }),
334 )
335 .await;
336 }
337 }
338 Err(e) => {
339 warn!(error = %e, "Server failed to decode received APDU");
340 }
341 }
342 }
343 });
344
345 let cov_table_for_purge = Arc::clone(&cov_table);
346 let cov_purge_task = tokio::spawn(async move {
347 let mut interval = tokio::time::interval(Duration::from_secs(30));
348 loop {
349 interval.tick().await;
350 let mut table = cov_table_for_purge.write().await;
351 let purged = table.purge_expired();
352 if purged > 0 {
353 debug!(purged, "Purged expired COV subscriptions");
354 }
355 }
356 });
357
358 let fault_detection_task = if config.enable_fault_detection {
359 let db_fault = Arc::clone(&db);
360 Some(tokio::spawn(async move {
361 let detector = crate::fault_detection::FaultDetector::default();
362 let mut interval = tokio::time::interval(Duration::from_secs(10));
363 loop {
364 interval.tick().await;
365 let mut db_guard = db_fault.write().await;
366 let changes = detector.evaluate(&mut db_guard);
367 for change in &changes {
368 debug!(
369 object = %change.object_id,
370 old = change.old_reliability,
371 new = change.new_reliability,
372 "Fault detection: reliability changed"
373 );
374 }
375 }
376 }))
377 } else {
378 None
379 };
380
381 let event_enrollment_task = if config.enable_fault_detection {
382 let db_ee = Arc::clone(&db);
383 Some(tokio::spawn(async move {
384 let mut interval = tokio::time::interval(Duration::from_secs(10));
385 loop {
386 interval.tick().await;
387 let mut db_guard = db_ee.write().await;
388 let transitions =
389 crate::event_enrollment::evaluate_event_enrollments(&mut db_guard);
390 for t in &transitions {
391 debug!(
392 enrollment = %t.enrollment_oid,
393 monitored = %t.monitored_oid,
394 from = ?t.change.from,
395 to = ?t.change.to,
396 "Event enrollment: state changed"
397 );
398 }
399 }
400 }))
401 } else {
402 None
403 };
404
405 let db_trend = Arc::clone(&db);
406 let trend_log_state: crate::trend_log::TrendLogState =
407 Arc::new(tokio::sync::Mutex::new(std::collections::HashMap::new()));
408 let trend_log_task = Some(tokio::spawn(async move {
409 let mut interval = tokio::time::interval(Duration::from_secs(1));
410 loop {
411 interval.tick().await;
412 crate::trend_log::poll_trend_logs(&db_trend, &trend_log_state).await;
413 }
414 }));
415
416 let db_schedule = Arc::clone(&db);
417 let schedule_tick_task = Some(tokio::spawn(async move {
418 let mut interval = tokio::time::interval(Duration::from_secs(60));
419 loop {
420 interval.tick().await;
421 crate::schedule::tick_schedules(&db_schedule, 0).await;
423 }
424 }));
425
426 Ok(Self {
427 config,
428 network,
429 db,
430 cov_table,
431 seg_ack_senders,
432 cov_in_flight,
433 server_tsm,
434 comm_state,
435 dcc_timer,
436 dispatch_task: Some(dispatch_task),
437 cov_purge_task: Some(cov_purge_task),
438 fault_detection_task,
439 event_enrollment_task,
440 trend_log_task,
441 schedule_tick_task,
442 local_mac,
443 })
444 }
445
446 pub fn local_mac(&self) -> &[u8] {
448 &self.local_mac
449 }
450
451 pub fn database(&self) -> &Arc<RwLock<ObjectDatabase>> {
453 &self.db
454 }
455
456 pub fn i_am_broadcaster(&self) -> IAmBroadcaster<T> {
458 IAmBroadcaster {
459 config: self.config.clone(),
460 network: Arc::clone(&self.network),
461 db: Arc::clone(&self.db),
462 }
463 }
464
465 pub fn comm_state(&self) -> u8 {
469 self.comm_state.load(Ordering::Acquire)
470 }
471
472 pub async fn generate_pics(&self, pics_config: &crate::pics::PicsConfig) -> crate::pics::Pics {
477 let db = self.db.read().await;
478 crate::pics::PicsGenerator::new(&db, &self.config, pics_config).generate()
479 }
480
481 pub async fn broadcast_i_am(&self) -> Result<(), Error> {
483 broadcast_i_am_from(&self.config, &self.db, &self.network).await
484 }
485
486 pub async fn stop(&mut self) -> Result<(), Error> {
488 if let Some(task) = self.fault_detection_task.take() {
489 task.abort();
490 let _ = task.await;
491 }
492 if let Some(task) = self.event_enrollment_task.take() {
493 task.abort();
494 let _ = task.await;
495 }
496 if let Some(task) = self.trend_log_task.take() {
497 task.abort();
498 let _ = task.await;
499 }
500 if let Some(task) = self.schedule_tick_task.take() {
501 task.abort();
502 let _ = task.await;
503 }
504 if let Some(task) = self.cov_purge_task.take() {
505 task.abort();
506 let _ = task.await;
507 }
508 if let Some(task) = self.dispatch_task.take() {
509 task.abort();
510 let _ = task.await;
511 }
512 Ok(())
513 }
514}
515
516impl<T: TransportPort + 'static> IAmBroadcaster<T> {
517 pub async fn broadcast_i_am(&self) -> Result<(), Error> {
519 broadcast_i_am_from(&self.config, &self.db, &self.network).await
520 }
521}
522
523async fn broadcast_i_am_from<T: TransportPort + 'static>(
524 config: &ServerConfig,
525 db: &Arc<RwLock<ObjectDatabase>>,
526 network: &Arc<NetworkLayer<T>>,
527) -> Result<(), Error> {
528 let device_oid = {
529 let db = db.read().await;
530 db.list_objects()
531 .into_iter()
532 .find(|oid| oid.object_type() == ObjectType::DEVICE)
533 .ok_or_else(|| Error::Encoding("no Device object in database".into()))?
534 };
535
536 let i_am = IAmRequest {
537 object_identifier: device_oid,
538 max_apdu_length: config.max_apdu_length,
539 segmentation_supported: config.segmentation_supported,
540 vendor_id: config.vendor_id,
541 };
542
543 let mut service_buf = BytesMut::new();
544 i_am.encode(&mut service_buf);
545
546 let pdu = Apdu::UnconfirmedRequest(UnconfirmedRequestPdu {
547 service_choice: UnconfirmedServiceChoice::I_AM,
548 service_request: service_buf.freeze(),
549 });
550
551 let mut buf = BytesMut::new();
552 encode_apdu(&mut buf, &pdu)?;
553
554 network
555 .broadcast_apdu(&buf, false, NetworkPriority::NORMAL)
556 .await
557}