Skip to main content

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}