bacnet_server/server/lifecycle.rs
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)); // 0 = Enable (default)
36 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 reply_tx: None,
289 }
290 }),
291 )
292 .await;
293 }
294 Err(e) => {
295 warn!(
296 error = %e,
297 "Failed to reassemble segmented request"
298 );
299 }
300 }
301 }
302 }
303
304 true
305 } else {
306 false
307 }
308 } else {
309 false
310 };
311
312 if !handled {
313 Self::dispatch(
314 &db_dispatch,
315 &network_dispatch,
316 &cov_dispatch,
317 &seg_ack_dispatch,
318 &cov_in_flight_dispatch,
319 &server_tsm_dispatch,
320 &comm_state_dispatch,
321 &dcc_timer_dispatch,
322 &config_dispatch,
323 &source_mac,
324 decoded,
325 received.take().unwrap_or_else(|| {
326 warn!("received consumed twice — using empty fallback");
327 bacnet_network::layer::ReceivedApdu {
328 apdu: bytes::Bytes::new(),
329 source_mac: bacnet_types::MacAddr::new(),
330 source_network: None,
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 // TODO: Read UTC_Offset from Device object for local time
422 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 /// Get the server's local MAC address.
447 pub fn local_mac(&self) -> &[u8] {
448 &self.local_mac
449 }
450
451 /// Get a reference to the shared object database.
452 pub fn database(&self) -> &Arc<RwLock<ObjectDatabase>> {
453 &self.db
454 }
455
456 /// Get the communication state per DeviceCommunicationControl.
457 ///
458 /// Returns 0 (Enable), 1 (Disable), or 2 (DisableInitiation).
459 pub fn comm_state(&self) -> u8 {
460 self.comm_state.load(Ordering::Acquire)
461 }
462
463 /// Generate a PICS document from the current object database and server configuration.
464 ///
465 /// The caller must supply a [`PicsConfig`] for fields not available from the server
466 /// (vendor name, model, firmware revision, etc.).
467 pub async fn generate_pics(&self, pics_config: &crate::pics::PicsConfig) -> crate::pics::Pics {
468 let db = self.db.read().await;
469 crate::pics::PicsGenerator::new(&db, &self.config, pics_config).generate()
470 }
471
472 /// Stop the server.
473 pub async fn stop(&mut self) -> Result<(), Error> {
474 if let Some(task) = self.fault_detection_task.take() {
475 task.abort();
476 let _ = task.await;
477 }
478 if let Some(task) = self.event_enrollment_task.take() {
479 task.abort();
480 let _ = task.await;
481 }
482 if let Some(task) = self.trend_log_task.take() {
483 task.abort();
484 let _ = task.await;
485 }
486 if let Some(task) = self.schedule_tick_task.take() {
487 task.abort();
488 let _ = task.await;
489 }
490 if let Some(task) = self.cov_purge_task.take() {
491 task.abort();
492 let _ = task.await;
493 }
494 if let Some(task) = self.dispatch_task.take() {
495 task.abort();
496 let _ = task.await;
497 }
498 Ok(())
499 }
500}