bacnet_endpoint/roles.rs
1//! Session-bound role handles above the sibling client/server roles.
2//!
3//! [`ClientRoleHandle`] exposes unsegmented reads and direct B/IP WriteProperty initiation;
4//! [`ServerRoleHandle`] exposes inbound handling, session liveness, the
5//! one-shot deferred-reply arm, and notification admit-complete. Neither
6//! handle exposes lifecycle: `start`/`stop` exist only on
7//! [`EndpointSession`](crate::session::EndpointSession), and the owner-only
8//! shutdown helpers are `pub(crate)`.
9//!
10//! Both handles are `Send + Sync` (statically asserted below), so a handle
11//! cloned out of the session may move across tasks and survive the session
12//! drop as a value — every call then fails closed via the [`Weak`] token.
13//!
14//! ```compile_fail,E0599
15//! // No lifecycle on roles: `start` exists only on `EndpointSession`.
16//! # use bacnet_endpoint::roles::ServerRoleHandle;
17//! # async fn forbidden(handle: ServerRoleHandle) {
18//! handle.start().await;
19//! # }
20//! ```
21
22use std::sync::atomic::{AtomicBool, Ordering};
23use std::sync::{Arc, Weak};
24
25use bacnet_encoding::apdu::{Apdu, decode_apdu};
26use bacnet_endpoint_core::coordinator::{
27 AdmissionOutcome, CanonicalPeer, OutboundTransactionCoordinator,
28};
29pub use bacnet_endpoint_core::endpoint_ingress::EndpointApduDestination;
30use bacnet_network::layer::ReceivedApdu;
31use bacnet_transport::port::DataAttribute;
32use bacnet_types::enums::PropertyIdentifier;
33use bacnet_types::error::Error;
34use bacnet_types::primitives::ObjectIdentifier;
35
36use bacnet_client::EndpointRequester;
37use bacnet_server::server::{
38 __endpoint_EndpointResponder as EndpointResponder,
39 __endpoint_NotificationTransactions as NotificationTransactions,
40};
41
42fn shutdown_error() -> Error {
43 Error::Encoding("endpoint shutdown".into())
44}
45
46/// Session-owned open state shared with role handles.
47///
48/// Roles hold only a [`Weak`] to this token: dropping the session ends role
49/// work even when a role [`Arc`] was cloned out beforehand.
50#[doc(hidden)]
51pub struct SessionToken {
52 open: AtomicBool,
53 /// Shared device-wide outbound lease pool (client + server notifications).
54 pub coordinator: Arc<OutboundTransactionCoordinator>,
55 /// RB-17 one-shot deferred-reply arm (MS/TP `ReplyPostponed` wiring).
56 ///
57 /// When set, the session dispatch strips the one-use prompt reply sender
58 /// from the next `reply_tx`-bearing inbound request before the responder
59 /// sees it, forcing the token-owned egress path with identical
60 /// addressing/invoke ID. Consumed exactly once via [`take_suspend`](Self::take_suspend);
61 /// broadcasts (no `reply_tx`) never consume the arm.
62 suspend_next: AtomicBool,
63}
64
65impl SessionToken {
66 #[doc(hidden)]
67 pub fn new(coordinator: Arc<OutboundTransactionCoordinator>) -> Arc<Self> {
68 Arc::new(Self {
69 open: AtomicBool::new(true),
70 coordinator,
71 suspend_next: AtomicBool::new(false),
72 })
73 }
74
75 #[doc(hidden)]
76 pub fn shutdown(&self) {
77 self.open.store(false, Ordering::Release);
78 }
79
80 #[doc(hidden)]
81 pub fn is_open(&self) -> bool {
82 self.open.load(Ordering::Acquire)
83 }
84
85 /// Arms one-shot deferred-reply suspension (RB-17 MS/TP wiring).
86 ///
87 /// Test + slow-application seam: the next `reply_tx`-bearing inbound
88 /// request is answered token-owned via egress (after the MAC releases
89 /// `ReplyPostponed`) instead of promptly via `reply_tx`.
90 #[doc(hidden)]
91 pub fn suspend_next_reply(&self) {
92 self.suspend_next.store(true, Ordering::Release);
93 }
94
95 /// Consumes the suspension arm exactly once (single dispatch consumer).
96 #[doc(hidden)]
97 pub fn take_suspend(&self) -> bool {
98 self.suspend_next.swap(false, Ordering::AcqRel)
99 }
100}
101
102/// Caller assertion about the remote target property's command semantics.
103/// No local object lookup, priority argument, or remote discovery supplies this fact.
104#[derive(Debug, Clone, Copy, PartialEq, Eq)]
105pub enum Commandability {
106 /// The property accepts prioritized commands; omitted priority is effective 16.
107 Commandable,
108 /// The property is not commandable; Audit ignores any supplied wire priority.
109 Noncommandable,
110}
111
112/// Client role: reads and direct B/IP WriteProperty over shared egress/coordinator.
113///
114/// Borrowed from a running [`EndpointSession`](crate::session::EndpointSession)
115/// via `client()` / `cloned_client_handle()`. No lifecycle methods: after the
116/// owning session stops or drops, every call returns
117/// [`Error::Encoding`] (`"endpoint
118/// shutdown"`). `Send + Sync`, so clones may outlive the session borrow and
119/// move across tasks.
120///
121/// Service scope is unsegmented RP/RR/RPM and direct B/IP WriteProperty.
122/// Explicit destinations preserve data attributes and ingress provenance. When source Audit
123/// reporting is selected, only direct B/IP IPv4 unicast targets are admitted.
124/// Audited calls are session-owned before egress: dropping their caller does not
125/// cancel an admitted request or its terminal observation. Other calls retain
126/// caller-owned cancellation. Stop/drop may discard undelivered source records.
127#[derive(Clone)]
128pub struct ClientRoleHandle {
129 token: Weak<SessionToken>,
130 requester: EndpointRequester,
131 bip_broadcast: Option<std::net::Ipv4Addr>,
132 source_audit: Option<Weak<crate::source_audit::SourceAudit>>,
133}
134
135impl ClientRoleHandle {
136 #[doc(hidden)]
137 pub fn new(token: &Arc<SessionToken>, requester: EndpointRequester) -> Self {
138 Self {
139 token: Arc::downgrade(token),
140 requester,
141 bip_broadcast: None,
142 source_audit: None,
143 }
144 }
145
146 pub(crate) fn with_bip_broadcast(mut self, broadcast: Option<std::net::Ipv4Addr>) -> Self {
147 self.bip_broadcast = broadcast;
148 self
149 }
150
151 /// Binding preflight for encoded WriteProperty input, without a lease or traffic.
152 #[doc(hidden)]
153 pub fn validate_write_property(
154 &self,
155 request: &bacnet_services::write_property::WritePropertyRequest,
156 ) -> Result<(), Error> {
157 self.requester
158 .validate_operation(&bacnet_client::EndpointOperationRequest::Write(
159 request.clone(),
160 ))
161 }
162
163 /// Write one property to a direct B/IP IPv4 unicast peer through the shared requester.
164 ///
165 /// `commandability` is required even without a source Reporter. It controls Audit
166 /// filtering only: the supplied wire priority and complete encoded value are preserved.
167 /// Empty lists are allowed; malformed TLVs and oversized unsegmented requests fail
168 /// before traffic. An eligible admitted write remains session-owned after caller
169 /// cancellation. Retries share one Invoke ID and produce at most one source record.
170 pub async fn write_property(
171 &self,
172 destination_mac: &[u8],
173 request: bacnet_services::write_property::WritePropertyRequest,
174 commandability: Commandability,
175 ) -> Result<(), Error> {
176 self.check_open()?;
177 let broadcast = self
178 .bip_broadcast
179 .ok_or_else(|| Error::Encoding("endpoint WriteProperty requires IPv4 B/IP".into()))?;
180 let (ip, port) = bacnet_transport::bvll::decode_bip_mac(destination_mac)?;
181 let ip = std::net::Ipv4Addr::from(ip);
182 if ip.is_unspecified()
183 || ip.is_multicast()
184 || ip.is_broadcast()
185 || ip == broadcast
186 || port == 0
187 {
188 return Err(Error::Encoding(
189 "endpoint WriteProperty requires direct B/IP IPv4 unicast".into(),
190 ));
191 }
192 let destination = EndpointApduDestination::Direct {
193 destination_mac: bacnet_types::MacAddr::from_slice(destination_mac),
194 };
195 if let Some(source) = &self.source_audit {
196 source
197 .upgrade()
198 .ok_or_else(shutdown_error)?
199 .write(&self.requester, destination, request, commandability)
200 .await
201 } else {
202 self.requester
203 .prepare_write(destination, Vec::new(), request)
204 .await?
205 .execute()
206 .await
207 .result?
208 .into_write()
209 }
210 }
211
212 pub(crate) fn with_source_audit(
213 mut self,
214 source: &Arc<crate::source_audit::SourceAudit>,
215 ) -> Self {
216 self.source_audit = Some(Arc::downgrade(source));
217 self
218 }
219
220 fn session(&self) -> Result<Arc<SessionToken>, Error> {
221 self.token.upgrade().ok_or_else(shutdown_error)
222 }
223
224 fn check_open(&self) -> Result<Arc<SessionToken>, Error> {
225 let session = self.session()?;
226 if !session.is_open() {
227 return Err(shutdown_error());
228 }
229 Ok(session)
230 }
231
232 /// Direct ReadProperty (no routing attributes).
233 ///
234 /// ACK object/property/index correlation matches the standalone client.
235 /// Device/Network Port wildcard requests accept a same-type concrete ACK;
236 /// other mismatches return a decoding error.
237 ///
238 /// Fails closed with `"endpoint shutdown"` once the owning session stops
239 /// or drops. Timeouts/retries come from the session config (or the
240 /// composed identity's max-APDU override for the length clamp).
241 pub async fn read_property(
242 &self,
243 destination_mac: &[u8],
244 object_identifier: ObjectIdentifier,
245 property_identifier: PropertyIdentifier,
246 property_array_index: Option<u32>,
247 ) -> Result<bacnet_services::read_property::ReadPropertyACK, Error> {
248 self.read_property_with_destination(
249 EndpointApduDestination::Direct {
250 destination_mac: bacnet_types::MacAddr::from_slice(destination_mac),
251 },
252 Vec::new(),
253 object_identifier,
254 property_identifier,
255 property_array_index,
256 )
257 .await
258 }
259
260 /// Explicit-destination ReadProperty with pass-through attributes.
261 ///
262 /// Covers routed, local-broadcast and addressed destinations the direct
263 /// shorthand cannot spell. Same fail-closed shutdown semantics as
264 /// [`read_property`](Self::read_property). A routed destination names one
265 /// device: a network outside 1..=65534, or a MAC that is empty or longer
266 /// than 18 octets, fails with [`Error::Encoding`] before a transaction
267 /// is reserved, as does a broadcast destination. Once the session knows
268 /// its own network's number, a routed destination naming it goes to its
269 /// MAC with no DNET (#1403). The answer from that MAC completes the read,
270 /// and so does one a router relays back with that number as its SNET and
271 /// that MAC as its SADR (#1465).
272 pub async fn read_property_with_destination(
273 &self,
274 destination: EndpointApduDestination,
275 data_attributes: Vec<DataAttribute>,
276 object_identifier: ObjectIdentifier,
277 property_identifier: PropertyIdentifier,
278 property_array_index: Option<u32>,
279 ) -> Result<bacnet_services::read_property::ReadPropertyACK, Error> {
280 self.check_open()?;
281 if let Some(source) = &self.source_audit {
282 let source = source.upgrade().ok_or_else(shutdown_error)?;
283 return source
284 .read(
285 &self.requester,
286 destination,
287 data_attributes,
288 bacnet_client::EndpointReadRequest::Property(
289 bacnet_services::read_property::ReadPropertyRequest {
290 object_identifier,
291 property_identifier,
292 property_array_index,
293 },
294 ),
295 )
296 .await?
297 .into_property();
298 }
299 self.requester
300 .read_property_with_destination(
301 destination,
302 data_attributes,
303 object_identifier,
304 property_identifier,
305 property_array_index,
306 )
307 .await
308 }
309
310 /// Read 1–64 explicit property references across nonempty concrete objects.
311 ///
312 /// ALL/REQUIRED/OPTIONAL and wildcard objects are excluded from this endpoint
313 /// profile. Index zero is valid. Responses must match complete ordered
314 /// occurrences; inline errors may omit a requested index. An active source
315 /// Reporter emits one value-free record per eligible occurrence, sharing one
316 /// operation, timestamp, invoke ID and recipient snapshot.
317 pub async fn read_property_multiple(
318 &self,
319 destination_mac: &[u8],
320 specs: Vec<bacnet_types::constructed::ReadAccessSpecification>,
321 ) -> Result<bacnet_services::rpm::ReadPropertyMultipleACK, Error> {
322 self.read_property_multiple_with_destination(
323 EndpointApduDestination::Direct {
324 destination_mac: bacnet_types::MacAddr::from_slice(destination_mac),
325 },
326 Vec::new(),
327 specs,
328 )
329 .await
330 }
331
332 /// Explicit RPM destination and pass-through attributes. Audited operations
333 /// require direct B/IP IPv4 unicast, which includes a routed destination
334 /// naming the session's own network number once it is known (#1403); all
335 /// responses remain unsegmented.
336 pub async fn read_property_multiple_with_destination(
337 &self,
338 destination: EndpointApduDestination,
339 data_attributes: Vec<DataAttribute>,
340 specs: Vec<bacnet_types::constructed::ReadAccessSpecification>,
341 ) -> Result<bacnet_services::rpm::ReadPropertyMultipleACK, Error> {
342 self.check_open()?;
343 let request = bacnet_client::EndpointReadRequest::Multiple(
344 bacnet_services::rpm::ReadPropertyMultipleRequest {
345 list_of_read_access_specs: specs,
346 },
347 );
348 if let Some(source) = &self.source_audit {
349 source
350 .upgrade()
351 .ok_or_else(shutdown_error)?
352 .read(&self.requester, destination, data_attributes, request)
353 .await?
354 .into_multiple()
355 } else {
356 self.requester
357 .prepare_read(destination, data_attributes, request)
358 .await?
359 .execute()
360 .await
361 .result?
362 .into_multiple()
363 }
364 }
365
366 /// Read a list/log range through the shared unsegmented requester.
367 ///
368 /// An active source Reporter emits one value-free READ record per attempted
369 /// operation. Audited destinations must be direct B/IP unicast. Invalid
370 /// selectors, array index zero, counts and ByTime components fail prewire.
371 pub async fn read_range(
372 &self,
373 destination_mac: &[u8],
374 object_identifier: ObjectIdentifier,
375 property_identifier: PropertyIdentifier,
376 property_array_index: Option<u32>,
377 range: Option<bacnet_services::read_range::RangeSpec>,
378 ) -> Result<bacnet_services::read_range::ReadRangeAck, Error> {
379 self.read_range_with_destination(
380 EndpointApduDestination::Direct {
381 destination_mac: bacnet_types::MacAddr::from_slice(destination_mac),
382 },
383 Vec::new(),
384 object_identifier,
385 property_identifier,
386 property_array_index,
387 range,
388 )
389 .await
390 }
391
392 /// Explicit destination and pass-through attributes for ReadRange.
393 /// Routed destinations are available when source reporting is not selected
394 /// for the operation, under the same one-device rule as
395 /// [`read_property_with_destination`](Self::read_property_with_destination).
396 /// A routed destination naming the session's own network number, once it
397 /// is known, goes direct and can be audited (#1403). Responses remain
398 /// unsegmented.
399 pub async fn read_range_with_destination(
400 &self,
401 destination: EndpointApduDestination,
402 data_attributes: Vec<DataAttribute>,
403 object_identifier: ObjectIdentifier,
404 property_identifier: PropertyIdentifier,
405 property_array_index: Option<u32>,
406 range: Option<bacnet_services::read_range::RangeSpec>,
407 ) -> Result<bacnet_services::read_range::ReadRangeAck, Error> {
408 let request = bacnet_services::read_range::ReadRangeRequest {
409 object_identifier,
410 property_identifier,
411 property_array_index,
412 range,
413 };
414 self.read_range_checked(
415 destination,
416 data_attributes,
417 request,
418 bacnet_services::read_range::ReadRangeValidation::Strict,
419 )
420 .await
421 .map(|reply| reply.ack)
422 }
423
424 /// Read a range of items, choosing how to treat an acknowledgement that
425 /// breaks a rule, as
426 /// [`BACnetClient::read_range_with`](bacnet_client::client::BACnetClient::read_range_with)
427 /// does. Audited like [`read_range`](Self::read_range).
428 pub async fn read_range_with(
429 &self,
430 destination_mac: &[u8],
431 request: &bacnet_services::read_range::ReadRangeRequest,
432 validation: bacnet_services::read_range::ReadRangeValidation,
433 ) -> Result<bacnet_services::read_range::ReadRangeReply, Error> {
434 self.read_range_checked(
435 EndpointApduDestination::Direct {
436 destination_mac: bacnet_types::MacAddr::from_slice(destination_mac),
437 },
438 Vec::new(),
439 request.clone(),
440 validation,
441 )
442 .await
443 }
444
445 /// Read one page of `log`'s Log_Buffer from `cursor`, as
446 /// [`bacnet_client::log_reader::read_log_page`] does; each request is
447 /// audited like any other read, and paced by the session's
448 /// [`min_request_interval_ms`](crate::session::SessionConfig::min_request_interval_ms)
449 /// as a paced `BACnetClient`'s are.
450 pub async fn read_log_page(
451 &self,
452 destination_mac: &[u8],
453 log: ObjectIdentifier,
454 cursor: bacnet_client::log_reader::LogCursor,
455 page_size: u16,
456 ) -> Result<bacnet_client::log_reader::LogPage, Error> {
457 bacnet_client::log_reader::read_log_page(self, destination_mac, log, cursor, page_size)
458 .await
459 }
460
461 async fn read_range_checked(
462 &self,
463 destination: EndpointApduDestination,
464 data_attributes: Vec<DataAttribute>,
465 request: bacnet_services::read_range::ReadRangeRequest,
466 validation: bacnet_services::read_range::ReadRangeValidation,
467 ) -> Result<bacnet_services::read_range::ReadRangeReply, Error> {
468 self.check_open()?;
469 if let Some(source) = &self.source_audit {
470 let source = source.upgrade().ok_or_else(shutdown_error)?;
471 return source
472 .read(
473 &self.requester,
474 destination,
475 data_attributes,
476 bacnet_client::EndpointReadRequest::Range(request, validation),
477 )
478 .await?
479 .into_range();
480 }
481 self.requester
482 .read_range_with_destination(destination, data_attributes, request, validation)
483 .await
484 }
485
486 /// Completes one pre-admitted response (session dispatch only).
487 ///
488 /// Preserves routing/attributes/provenance via the inner requester;
489 /// never allocates an Invoke ID (the lease was reserved before the send).
490 #[doc(hidden)]
491 pub async fn complete_pre_admitted(
492 &self,
493 admission: bacnet_endpoint_core::coordinator::Admission,
494 apdu: Apdu,
495 received: ReceivedApdu,
496 ) -> bool {
497 if self.check_open().is_err() {
498 return false;
499 }
500 self.requester
501 .complete_pre_admitted(admission, apdu, received)
502 .await
503 }
504
505 /// Internal close for the session owner only.
506 ///
507 /// Owner-only seam (`pub(crate)`): the session [`Drop`]
508 /// path currently closes the inner requester directly; this stays for
509 /// owner-driven shutdown without exposing lifecycle on the handle.
510 #[allow(dead_code)]
511 pub(crate) fn close_for_owner(&self) {
512 self.requester.close();
513 }
514}
515
516impl bacnet_client::log_reader::LogRequester for ClientRoleHandle {
517 async fn read_unsigned(
518 &self,
519 mac: &[u8],
520 log: ObjectIdentifier,
521 property: PropertyIdentifier,
522 ) -> Result<u64, Error> {
523 let ack = self.read_property(mac, log, property, None).await?;
524 bacnet_client::log_reader::unsigned_value(&ack.property_value)
525 }
526
527 async fn read_range_lenient(
528 &self,
529 mac: &[u8],
530 request: &bacnet_services::read_range::ReadRangeRequest,
531 ) -> Result<bacnet_services::read_range::ReadRangeReply, Error> {
532 self.read_range_with(
533 mac,
534 request,
535 bacnet_services::read_range::ReadRangeValidation::Lenient,
536 )
537 .await
538 }
539}
540
541/// Server role: narrow responder + shared notification pool.
542///
543/// Service scope stays narrow: `ReadProperty`, optionally authorized local
544/// Device.Description `WriteProperty`, optionally `ReinitializeDevice` through
545/// its handler, and `Reject`/`Abort`. Full
546/// `bacnet-server` parity is a later packet. No
547/// lifecycle methods on this handle; the session owner drives dispatch +
548/// `close()`. `Send + Sync`; every method fails closed after shutdown.
549#[derive(Clone)]
550pub struct ServerRoleHandle {
551 token: Weak<SessionToken>,
552 responder: Arc<EndpointResponder>,
553 notifications: Arc<NotificationTransactions>,
554}
555
556impl ServerRoleHandle {
557 #[doc(hidden)]
558 pub fn new(
559 token: &Arc<SessionToken>,
560 responder: Arc<EndpointResponder>,
561 notifications: Arc<NotificationTransactions>,
562 ) -> Self {
563 Self {
564 token: Arc::downgrade(token),
565 responder,
566 notifications,
567 }
568 }
569
570 fn check_open(&self) -> Result<(), Error> {
571 let session = self.token.upgrade().ok_or_else(shutdown_error)?;
572 if !session.is_open() {
573 return Err(shutdown_error());
574 }
575 Ok(())
576 }
577
578 /// Returns `true` while the owning session is alive and open.
579 ///
580 /// Test seam proving no detached role outlives the endpoint: after the
581 /// session drops, [`Weak::upgrade`] fails and this returns `false`.
582 /// Cloned handles survive as values but report `false` and fail closed.
583 pub fn is_session_alive(&self) -> bool {
584 self.token.upgrade().is_some_and(|s| s.is_open())
585 }
586
587 /// Arms one-shot deferred-reply suspension (RB-17 MS/TP wiring).
588 ///
589 /// The next ordinary non-direct `reply_tx`-bearing request handled while this arm is
590 /// set answers token-owned via egress (after the MAC releases
591 /// `ReplyPostponed`) instead of promptly via `reply_tx`. Fails closed
592 /// once the owning session shuts down. Broadcasts (no `reply_tx`) never
593 /// consume the arm; neither do direct provenance or capability-bearing envelopes.
594 pub fn suspend_next_reply(&self) -> Result<(), Error> {
595 let session = self.token.upgrade().ok_or_else(shutdown_error)?;
596 if !session.is_open() {
597 return Err(shutdown_error());
598 }
599 session.suspend_next_reply();
600 Ok(())
601 }
602
603 /// Handles one inbound request via the wire invoke ID directly.
604 ///
605 /// Session dispatch only. Never touches the outbound coordinator pool.
606 /// Preserves the envelope's identity and response capability for the responder.
607 /// Honors a pending RB-17 suspension arm identically to session dispatch:
608 /// ordinary prompt work drops `reply_tx` and answers token-owned via egress.
609 /// Direct/capability-bearing envelopes retain checked original-route policy.
610 pub async fn handle_inbound(&self, mut received: ReceivedApdu) -> Result<bool, Error> {
611 self.check_open()?;
612 if !received.provenance.is_direct_peer()
613 && received.direct_response.is_none()
614 && received.reply_tx.is_some()
615 && self.token.upgrade().is_some_and(|s| s.take_suspend())
616 {
617 let _ = received.reply_tx.take();
618 }
619 self.responder.handle(received).await
620 }
621
622 /// Completes one already-admitted notification lease (dispatch only).
623 ///
624 /// The session's single [`admit_once`] owns exact-once claim; this
625 /// releases the exact lease without re-admitting. Returns `false` after
626 /// shutdown.
627 pub fn complete_notification_pre_admitted(
628 &self,
629 admission: bacnet_endpoint_core::coordinator::Admission,
630 apdu: &Apdu,
631 ) -> bool {
632 if self.check_open().is_err() {
633 return false;
634 }
635 self.notifications.complete_pre_admitted(admission, apdu)
636 }
637
638 /// Internal close for the session owner only.
639 ///
640 /// Owner-only seam (`pub(crate)`): retained for owner-driven shutdown
641 /// without exposing lifecycle on the handle.
642 #[allow(dead_code)]
643 pub(crate) fn close_for_owner(&self) {
644 self.responder.close();
645 self.notifications.close();
646 }
647}
648
649fn assert_send_sync<T: Send + Sync>() {}
650
651const _: fn() = || {
652 assert_send_sync::<ClientRoleHandle>();
653 assert_send_sync::<ServerRoleHandle>();
654};
655
656/// Derives the canonical peer for one received envelope (direct vs routed),
657/// given `local_network`, the session's known network number, if any: an
658/// envelope relayed with that number as its SNET is the direct station at
659/// its SADR ([`CanonicalPeer::from_source`], #1465).
660///
661/// RB-07 compat: provenance is preserved by the caller and never gates here.
662#[doc(hidden)]
663pub fn inbound_canonical_peer(
664 received: &ReceivedApdu,
665 local_network: Option<u16>,
666) -> CanonicalPeer {
667 CanonicalPeer::from_source(
668 received.source_mac.as_slice(),
669 received.source_network.as_ref(),
670 local_network,
671 )
672}
673
674/// Decodes one terminal/segment APDU, preserving the full envelope for the
675/// caller (source/dest, raw + effective group, attributes, provenance).
676#[doc(hidden)]
677pub fn decode_terminal(received: &ReceivedApdu) -> Option<Apdu> {
678 decode_apdu(received.apdu.clone()).ok()
679}
680
681/// Classifies whether an admitted lease belongs to the client requester.
682///
683/// Equal inbound/outbound numeric IDs are legal: ownership (`Requester` vs
684/// `Notification`), not the numeric value, selects the consumer.
685#[doc(hidden)]
686pub fn is_requester_lease(admission: &bacnet_endpoint_core::coordinator::Admission) -> bool {
687 admission.metadata().owner() == bacnet_endpoint_core::coordinator::LeaseOwner::Requester
688}
689
690/// Single-admit helper for session dispatch: exactly one
691/// [`OutboundTransactionCoordinator::admit_from_source`] per received
692/// terminal APDU, with `local_network`, the session's known network number.
693///
694/// Returns the coordinator outcome without releasing the lease; the selected
695/// role completes it exactly once via its pre-admitted path, keyed to the
696/// peer the admission matched.
697#[doc(hidden)]
698pub fn admit_once(
699 coordinator: &OutboundTransactionCoordinator,
700 received: &ReceivedApdu,
701 local_network: Option<u16>,
702 apdu: &Apdu,
703) -> Result<AdmissionOutcome, bacnet_endpoint_core::coordinator::CoordinatorError> {
704 coordinator.admit_from_source(
705 received.source_mac.as_slice(),
706 received.source_network.as_ref(),
707 local_network,
708 apdu,
709 )
710}