1use super::segmentation_context::{ConfirmedWait, SegmentedRequestLimits};
2use super::*;
3use bacnet_encoding::apdu::MINIMUM_MESSAGE_SIZE;
4
5enum LengthBoundedBy {
12 DiscoveredPeer(u16),
14 LocalConfig(u16),
16 Transport(u16),
18}
19
20impl core::fmt::Display for LengthBoundedBy {
21 fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
22 match self {
23 Self::DiscoveredPeer(v) => {
24 write!(f, "the peer's advertised Max APDU Length Accepted of {v}")
25 }
26 Self::LocalConfig(v) => write!(f, "this client's configured maximum of {v}"),
27 Self::Transport(v) => write!(f, "the transport's limit of {v}"),
28 }
29 }
30}
31
32fn check_transmittable_length(peer_or_local: LengthBoundedBy, transport: u16) -> Result<(), Error> {
52 let advertised = match peer_or_local {
53 LengthBoundedBy::DiscoveredPeer(v) | LengthBoundedBy::LocalConfig(v) => v,
54 LengthBoundedBy::Transport(v) => v,
55 };
56 let combined = advertised.min(transport);
57 if combined >= MINIMUM_MESSAGE_SIZE {
58 return Ok(());
59 }
60 let binding = if transport < advertised {
61 LengthBoundedBy::Transport(transport)
62 } else {
63 peer_or_local
64 };
65 Err(Error::Encoding(format!(
66 "maximum transmittable length {combined} is below the {MINIMUM_MESSAGE_SIZE}-octet \
67 MinimumMessageSize every BACnet device accepts (Clause 5.2.1.2); the binding limit is \
68 {binding}"
69 )))
70}
71
72impl<T: TransportPort + 'static> BACnetClient<T> {
73 pub async fn confirmed_request(
88 &self,
89 destination_mac: &[u8],
90 service_choice: ConfirmedServiceChoice,
91 service_data: &[u8],
92 ) -> Result<Bytes, Error> {
93 self.confirmed_request_inner(
94 ConfirmedTarget::Local {
95 mac: destination_mac,
96 },
97 service_choice,
98 service_data,
99 )
100 .await
101 }
102
103 pub async fn confirmed_request_routed(
124 &self,
125 router_mac: &[u8],
126 dest_network: u16,
127 dest_mac: &[u8],
128 service_choice: ConfirmedServiceChoice,
129 service_data: &[u8],
130 ) -> Result<Bytes, Error> {
131 self.confirmed_request_inner(
132 ConfirmedTarget::Routed {
133 router_mac,
134 dest_network,
135 dest_mac,
136 },
137 service_choice,
138 service_data,
139 )
140 .await
141 }
142
143 pub(super) async fn confirmed_request_inner(
144 &self,
145 target: ConfirmedTarget<'_>,
146 service_choice: ConfirmedServiceChoice,
147 service_data: &[u8],
148 ) -> Result<Bytes, Error> {
149 let named = target;
156 if let ConfirmedTarget::Routed {
157 dest_network,
158 dest_mac,
159 ..
160 } = named
161 {
162 check_routed_unicast(dest_network, dest_mac.len())?;
163 }
164 let target = named.localized(self.network.local_network_number().get());
165 if let ConfirmedTarget::Local { mac } = target {
168 if self.network.transport().is_group_destination(mac) {
169 return Err(Error::Encoding(
170 "a confirmed request goes to one device, not to a broadcast or group \
171 address (a local broadcast carries only an UNCONFIRMED_REQUEST APDU, \
172 Clause 6.3)"
173 .into(),
174 ));
175 }
176 }
177 let routed_forwarded_npci_len = match target {
178 ConfirmedTarget::Local { .. } => None,
179 ConfirmedTarget::Routed { dest_mac, .. } => {
180 Some(forwarded_npci_len(dest_mac.len(), self.local_mac.len())?)
181 }
182 };
183 let _pace = self.pacer.wait(pacing::PaceKey::of(target)).await;
188 let path_lease = match target {
191 ConfirmedTarget::Local { .. } => None,
192 ConfirmedTarget::Routed {
193 router_mac,
194 dest_network,
195 ..
196 } => Some(
197 self.routed_path_limits
198 .acquire(router_mac, dest_network)
199 .await?,
200 ),
201 };
202 let routed_path_max_apdu = path_lease
203 .as_ref()
204 .zip(routed_forwarded_npci_len)
205 .map(|(lease, npci_len)| lease.max_apdu(npci_len));
206 let transaction_peer = target.transaction_peer();
207 let tsm_mac = transaction_peer.tsm_mac;
208 let unsegmented_apdu_size = 4 + service_data.len();
209 let target_transport_max_apdu = self.target_transport_max_apdu_length(target);
210
211 let (peer_max_apdu, remote_max_segments, advertised, peer_segmentation) = {
212 let dt = self.device_table.lock().await;
213 let (device, peer_segmentation) = match named {
218 ConfirmedTarget::Local { mac } => {
219 (dt.get_by_mac(mac), dt.local_peer_segmentation(mac))
220 }
221 ConfirmedTarget::Routed {
222 dest_network,
223 dest_mac,
224 ..
225 } => (
226 dt.get_by_network_address(dest_network, dest_mac),
227 dt.routed_peer_segmentation(dest_network, dest_mac),
228 ),
229 };
230 let max_apdu = device
231 .map(|d| u16::try_from(d.max_apdu_length).unwrap_or(u16::MAX))
232 .unwrap_or(self.config.max_apdu_length);
233 let max_seg = device.and_then(|d| d.max_segments_accepted);
234 let advertised = if device.is_some() {
235 LengthBoundedBy::DiscoveredPeer(max_apdu)
236 } else {
237 LengthBoundedBy::LocalConfig(max_apdu)
238 };
239 (max_apdu, max_seg, advertised, peer_segmentation)
240 };
241 check_transmittable_length(advertised, target_transport_max_apdu)?;
242 if routed_path_max_apdu.is_some_and(|path_max_apdu| {
243 path_max_apdu < MINIMUM_MESSAGE_SIZE
244 && path_max_apdu <= peer_max_apdu.min(target_transport_max_apdu)
245 }) {
246 let ConfirmedTarget::Routed { dest_network, .. } = target else {
247 unreachable!("a routed path limit exists only for a routed target")
248 };
249 return Err(Error::RoutedPathTooLong { dnet: dest_network });
250 }
251 let remote_max_apdu = peer_max_apdu
252 .min(target_transport_max_apdu)
253 .min(routed_path_max_apdu.unwrap_or(u16::MAX));
254 if unsegmented_apdu_size > remote_max_apdu as usize {
255 if let Some(crate::discovery::PeerSegmentation::Authoritative(capability)) =
262 peer_segmentation
263 {
264 if matches!(
265 capability,
266 bacnet_types::enums::Segmentation::NONE
267 | bacnet_types::enums::Segmentation::TRANSMIT
268 ) {
269 return Err(bacnet_types::error::SegmentationError::PeerCannotReceive {
270 capability,
271 }
272 .into());
273 }
274 }
275 return self
276 .segmented_confirmed_request(
277 target,
278 service_choice,
279 service_data,
280 SegmentedRequestLimits {
281 remote_max_apdu,
282 remote_max_segments,
283 routed_forwarded_npci_len,
284 routed_path_lease: path_lease.as_ref(),
285 },
286 )
287 .await;
288 }
289
290 let advertised_max_apdu = self.advertised_max_apdu_length_for_target(target)?;
291 let (invoke_id, registration) = {
292 let mut tsm = self.tsm.lock().await;
293 tsm.register_coordinated_transaction_with_progress(
294 tsm_mac.clone(),
295 transaction_peer.canonical,
296 service_choice,
297 false,
298 )
299 .map_err(|error| Error::Encoding(error.to_string()))?
300 };
301
302 let owner = registration.owner.clone();
303 let mut guard = TransactionGuard::new(
304 Arc::clone(&self.tsm),
305 self.cleanup_tx.clone(),
306 tsm_mac.clone(),
307 invoke_id,
308 owner.clone(),
309 None,
310 );
311
312 let pdu = Apdu::ConfirmedRequest(ConfirmedRequestPdu {
313 segmented: false,
314 more_follows: false,
315 segmented_response_accepted: self.config.segmented_response_accepted,
316 max_segments: self.config.max_segments,
317 max_apdu_length: advertised_max_apdu,
318 invoke_id,
319 sequence_number: None,
320 proposed_window_size: None,
321 service_choice,
322 service_request: Bytes::copy_from_slice(service_data),
323 });
324
325 let mut buf = BytesMut::with_capacity(6 + service_data.len());
326 encode_apdu(&mut buf, &pdu)?;
327
328 if let Some(forwarded_npci_len) = routed_forwarded_npci_len {
329 self.routed_path_limits.install_active(
330 target,
331 tsm_mac.clone(),
332 invoke_id,
333 owner.clone(),
334 forwarded_npci_len,
335 self.network.network_control_ingress_sequence(),
336 );
337 }
338 if !self.routed_path_limits.authorize_attempt(
339 target,
340 &tsm_mac,
341 invoke_id,
342 &owner,
343 buf.len(),
344 ) {
345 guard.mark_completed();
346 self.tsm
347 .lock()
348 .await
349 .cancel_transaction_for_owner(&tsm_mac, invoke_id, &owner);
350 self.enqueue_transaction_cleanup(&tsm_mac, invoke_id, &owner, false, None);
351 return Err(Error::Encoding(
352 "routed send lost its active path authorization".into(),
353 ));
354 }
355
356 if let Err(e) = self.send_confirmed_target_apdu(target, &buf).await {
357 guard.mark_completed();
358 let mut tsm = self.tsm.lock().await;
359 tsm.cancel_transaction_for_owner(&tsm_mac, invoke_id, &owner);
360 drop(tsm);
361 self.enqueue_transaction_cleanup(&tsm_mac, invoke_id, &owner, false, None);
362 return Err(e);
363 }
364
365 let response = self
366 .wait_for_confirmed_response(
367 ConfirmedWait {
368 target,
369 tsm_mac: &tsm_mac,
370 invoke_id,
371 owner: &owner,
372 retry_apdu: Some(&buf),
373 },
374 registration.response,
375 registration.progress,
376 )
377 .await;
378 if response.is_ok() {
379 if let Some(lease) = path_lease.as_ref() {
380 lease.mark_terminal_observed();
381 }
382 }
383 guard.mark_completed();
384 response.and_then(Self::confirmed_response_result)
385 }
386
387 pub(super) async fn wait_for_confirmed_response(
390 &self,
391 wait_for: ConfirmedWait<'_>,
392 mut response_rx: oneshot::Receiver<TsmResponse>,
393 mut progress_rx: ProgressReceiver,
394 ) -> Result<TsmResponse, Error> {
395 let ConfirmedWait {
396 target,
397 tsm_mac,
398 invoke_id,
399 owner,
400 retry_apdu,
401 } = wait_for;
402 let (request_timeout, segment_timeout, max_retries) = {
403 let tsm = self.tsm.lock().await;
404 let config = tsm.config();
405 (
406 Duration::from_millis(config.apdu_timeout_ms),
407 Duration::from_millis(config.apdu_segment_timeout_ms.saturating_mul(4)),
408 config.apdu_retries,
409 )
410 };
411 let mut retries_sent = 0u16;
412 let mut progress = progress_rx.borrow_and_update();
413
414 loop {
415 let wait = match progress {
416 TransactionProgress::AwaitingResponse => request_timeout,
417 TransactionProgress::SegmentedResponse { .. } => segment_timeout,
418 };
419 let timer = tokio::time::sleep(wait);
420 tokio::pin!(timer);
421
422 tokio::select! {
423 response = &mut response_rx => {
424 return response.map_err(|_| Error::Encoding("TSM response channel closed".into()));
425 }
426 () = progress_rx.changed() => {
429 progress = progress_rx.borrow_and_update();
430 }
431 _ = &mut timer => {
432 match progress {
433 TransactionProgress::AwaitingResponse => {
434 let retry = retry_apdu.is_some()
435 && retries_sent < u16::from(max_retries);
436 let mut tsm = self.tsm.lock().await;
437 match tsm.expire_request_timer(tsm_mac, invoke_id, owner, !retry) {
438 RequestTimerExpiration::Retry => {
439 let Some(retry_apdu) = retry_apdu else {
443 tsm.cancel_transaction_for_owner(tsm_mac, invoke_id, owner);
444 drop(tsm);
445 self.enqueue_transaction_cleanup(
446 tsm_mac,
447 invoke_id,
448 owner,
449 false,
450 None,
451 );
452 return Err(Error::Timeout(request_timeout));
453 };
454 retries_sent += 1;
455 drop(tsm);
456 if !self.routed_path_limits.authorize_attempt(
457 target,
458 tsm_mac,
459 invoke_id,
460 owner,
461 retry_apdu.len(),
462 ) {
463 return (&mut response_rx).await.map_err(|_| {
464 Error::Encoding("TSM response channel closed".into())
465 });
466 }
467 let send_result = self
468 .send_confirmed_target_apdu(target, retry_apdu)
469 .await;
470 if let Err(error) = send_result {
471 let mut tsm = self.tsm.lock().await;
472 match tsm.expire_request_timer(tsm_mac, invoke_id, owner, true) {
473 RequestTimerExpiration::SegmentedResponse { generation } => {
474 progress = TransactionProgress::SegmentedResponse { generation };
475 continue;
476 }
477 RequestTimerExpiration::NoTransaction => {
478 drop(tsm);
479 return (&mut response_rx)
480 .await
481 .map_err(|_| Error::Encoding("TSM response channel closed".into()));
482 }
483 RequestTimerExpiration::TimedOut => {
484 drop(tsm);
485 self.enqueue_transaction_cleanup(
486 tsm_mac,
487 invoke_id,
488 owner,
489 false,
490 None,
491 );
492 return Err(error);
493 }
494 RequestTimerExpiration::Retry => unreachable!(
495 "final retry-send failure disposition cannot authorize another retry"
496 ),
497 }
498 }
499 debug!(
500 invoke_id,
501 attempt = retries_sent,
502 max_retries,
503 "APDU timeout, retrying confirmed request"
504 );
505 }
506 RequestTimerExpiration::SegmentedResponse { generation } => {
507 progress = TransactionProgress::SegmentedResponse { generation };
508 }
509 RequestTimerExpiration::TimedOut => {
510 drop(tsm);
511 self.enqueue_transaction_cleanup(
512 tsm_mac,
513 invoke_id,
514 owner,
515 false,
516 None,
517 );
518 return Err(Error::Abort {
519 reason: bacnet_types::enums::AbortReason::TSM_TIMEOUT,
520 });
521 }
522 RequestTimerExpiration::NoTransaction => {
523 drop(tsm);
524 return (&mut response_rx)
525 .await
526 .map_err(|_| Error::Encoding("TSM response channel closed".into()));
527 }
528 }
529 }
530 TransactionProgress::SegmentedResponse { generation } => {
531 let mut tsm = self.tsm.lock().await;
532 match tsm.expire_segment_timer(tsm_mac, invoke_id, owner, generation) {
533 SegmentTimerExpiration::Activity { generation } => {
534 progress = TransactionProgress::SegmentedResponse { generation };
535 }
536 SegmentTimerExpiration::AwaitingResponse => {
537 progress = TransactionProgress::AwaitingResponse;
538 }
539 SegmentTimerExpiration::TimedOut => {
540 drop(tsm);
541 #[cfg(test)]
542 self.segmented_cleanup.pause_if_enabled().await;
543 self.enqueue_transaction_cleanup(
544 tsm_mac,
545 invoke_id,
546 owner,
547 false,
548 None,
549 );
550 return Err(Error::Abort {
551 reason: bacnet_types::enums::AbortReason::TSM_TIMEOUT,
552 });
553 }
554 SegmentTimerExpiration::NoTransaction => {
555 drop(tsm);
556 return (&mut response_rx)
557 .await
558 .map_err(|_| Error::Encoding("TSM response channel closed".into()));
559 }
560 }
561 }
562 }
563 }
564 }
565 }
566 }
567
568 pub(super) fn enqueue_transaction_cleanup(
569 &self,
570 mac: &MacAddr,
571 invoke_id: u8,
572 owner: &TransactionOwner,
573 cancel_tsm: bool,
574 seg_ack_sender: Option<mpsc::Sender<SegmentAckPdu>>,
575 ) {
576 let _ = self.cleanup_tx.send(TransactionCleanup {
577 mac: mac.clone(),
578 invoke_id,
579 owner: owner.clone(),
580 cancel_tsm,
581 seg_ack_sender,
582 });
583 }
584
585 pub(super) fn confirmed_response_result(response: TsmResponse) -> Result<Bytes, Error> {
586 super::confirmed_response_result(response)
587 }
588
589 pub(super) async fn send_confirmed_target_apdu(
590 &self,
591 target: ConfirmedTarget<'_>,
592 apdu: &[u8],
593 ) -> Result<(), Error> {
594 match target {
595 ConfirmedTarget::Local { mac } => {
596 self.network
597 .send_apdu(apdu, mac, true, NetworkPriority::Normal)
598 .await
599 }
600 ConfirmedTarget::Routed {
601 router_mac,
602 dest_network,
603 dest_mac,
604 } => {
605 self.network
606 .send_apdu_routed(
607 apdu,
608 dest_network,
609 dest_mac,
610 router_mac,
611 true,
612 NetworkPriority::Normal,
613 )
614 .await
615 }
616 }
617 }
618
619 pub(super) fn target_transport_max_apdu_length(&self, target: ConfirmedTarget<'_>) -> u16 {
620 self.network
621 .transport()
622 .egress_apdu_limit()
623 .saturating_sub(target.additional_npdu_header_len())
624 }
625
626 pub(super) fn advertised_max_apdu_length_for_target(
627 &self,
628 target: ConfirmedTarget<'_>,
629 ) -> Result<u16, Error> {
630 cap_max_apdu_to_transport(
631 self.config.max_apdu_length,
632 self.target_transport_max_apdu_length(target),
633 )
634 }
635
636 pub async fn unconfirmed_request(
638 &self,
639 destination_mac: &[u8],
640 service_choice: UnconfirmedServiceChoice,
641 service_data: &[u8],
642 ) -> Result<(), Error> {
643 let pdu = Apdu::UnconfirmedRequest(bacnet_encoding::apdu::UnconfirmedRequest {
644 service_choice,
645 service_request: Bytes::copy_from_slice(service_data),
646 });
647
648 let mut buf = BytesMut::with_capacity(2 + service_data.len());
649 encode_apdu(&mut buf, &pdu)?;
650
651 self.network
652 .send_apdu(&buf, destination_mac, false, NetworkPriority::Normal)
653 .await
654 }
655
656 pub async fn broadcast_unconfirmed(
658 &self,
659 service_choice: UnconfirmedServiceChoice,
660 service_data: &[u8],
661 ) -> Result<(), Error> {
662 let pdu = Apdu::UnconfirmedRequest(bacnet_encoding::apdu::UnconfirmedRequest {
663 service_choice,
664 service_request: Bytes::copy_from_slice(service_data),
665 });
666
667 let mut buf = BytesMut::with_capacity(2 + service_data.len());
668 encode_apdu(&mut buf, &pdu)?;
669
670 self.network
671 .broadcast_apdu(&buf, false, NetworkPriority::Normal)
672 .await
673 }
674
675 pub async fn broadcast_global_unconfirmed(
677 &self,
678 service_choice: UnconfirmedServiceChoice,
679 service_data: &[u8],
680 ) -> Result<(), Error> {
681 let pdu = Apdu::UnconfirmedRequest(bacnet_encoding::apdu::UnconfirmedRequest {
682 service_choice,
683 service_request: Bytes::copy_from_slice(service_data),
684 });
685
686 let mut buf = BytesMut::with_capacity(2 + service_data.len());
687 encode_apdu(&mut buf, &pdu)?;
688
689 self.network
690 .broadcast_global_apdu(&buf, false, NetworkPriority::Normal)
691 .await
692 }
693
694 pub async fn broadcast_network_unconfirmed(
700 &self,
701 service_choice: UnconfirmedServiceChoice,
702 service_data: &[u8],
703 dest_network: u16,
704 ) -> Result<(), Error> {
705 let pdu = Apdu::UnconfirmedRequest(bacnet_encoding::apdu::UnconfirmedRequest {
706 service_choice,
707 service_request: Bytes::copy_from_slice(service_data),
708 });
709
710 let mut buf = BytesMut::with_capacity(2 + service_data.len());
711 encode_apdu(&mut buf, &pdu)?;
712
713 if self.network.local_network_number().get() == Some(dest_network) {
714 return self
715 .network
716 .broadcast_apdu(&buf, false, NetworkPriority::Normal)
717 .await;
718 }
719 self.network
720 .broadcast_to_network(&buf, dest_network, false, NetworkPriority::Normal)
721 .await
722 }
723}
724
725#[cfg(test)]
726#[path = "requests_tests.rs"]
727mod transmittable_length_tests;