bacnet_network/router/
replies.rs

1//! Weak application reply authority for one running router.
2use std::sync::{
3    Arc, Weak,
4    atomic::{AtomicBool, Ordering},
5};
6
7use bacnet_router_core::router::ReplyPlan;
8use bacnet_transport::port::{DirectResponseScope, ResponseScopeGuard};
9use bacnet_types::{MacAddr, error::Error};
10use bytes::Bytes;
11use tokio::sync::oneshot;
12
13use super::{SendRequest, dispatch::Outputs};
14
15#[derive(Debug, Default)]
16pub(crate) struct ReplyLifetime {
17    stopped: AtomicBool,
18    pub(crate) direct: DirectResponseScope,
19}
20
21impl ReplyLifetime {
22    pub(super) fn seal(&self) {
23        self.stopped.store(true, Ordering::Release);
24        self.direct.seal();
25    }
26    fn check(&self) -> Result<(), Error> {
27        if self.stopped.load(Ordering::Acquire) {
28            Err(unavailable())
29        } else {
30            Ok(())
31        }
32    }
33}
34
35/// Saved router endpoint and return path. It owns no router, queue or socket.
36/// Only local delivery by a running router creates this capability.
37#[derive(Clone)]
38pub struct RouterResponse {
39    /// Shared, so the capability is one pointer in every locally delivered
40    /// APDU and in a server's saved response route, and a clone is a
41    /// reference count (#1713). Every field is fixed at creation.
42    inner: Arc<RouterResponseInner>,
43}
44
45struct RouterResponseInner {
46    plan: ReplyPlan,
47    max_npdu: u16,
48    outputs: Weak<Outputs>,
49    lifetime: Weak<ReplyLifetime>,
50}
51
52impl std::fmt::Debug for RouterResponse {
53    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
54        f.debug_struct("RouterResponse")
55            .field("plan", &self.inner.plan)
56            .finish_non_exhaustive()
57    }
58}
59
60impl RouterResponse {
61    pub(super) fn new(plan: ReplyPlan, max_npdu: u16, outputs: &Arc<Outputs>) -> Self {
62        Self {
63            inner: Arc::new(RouterResponseInner {
64                plan,
65                max_npdu,
66                outputs: Arc::downgrade(outputs),
67                lifetime: Arc::downgrade(&outputs.replies),
68            }),
69        }
70    }
71    /// The saved reply addressing.
72    pub(crate) fn plan(&self) -> &ReplyPlan {
73        &self.inner.plan
74    }
75    /// The return port's NPDU limit when the request arrived.
76    pub(crate) fn max_npdu(&self) -> u16 {
77        self.inner.max_npdu
78    }
79    pub(crate) fn same_origin(&self, other: &Self) -> bool {
80        let (this, other) = (&*self.inner, &*other.inner);
81        let same_port = this.plan.port() == other.plan.port();
82        this.lifetime.ptr_eq(&other.lifetime)
83            && this.plan.source() == other.plan.source()
84            && this.plan.destination() == other.plan.destination()
85            // SNET/SADR identifies a routed requester independently of its
86            // immediate return path. Port-local addresses remain scoped to
87            // their ingress link; equal MACs on two links are different peers.
88            && (this.plan.destination().is_some()
89                || (same_port && this.plan.peer() == other.plan.peer()))
90            // With no explicitly addressed endpoint, the ingress port selects
91            // the local endpoint. An explicit endpoint may be reached via any
92            // port of this same router. The saved plan itself never changes.
93            && (this.plan.source().is_some() || same_port)
94    }
95    pub(crate) fn check(&self) -> Result<(), Error> {
96        self.lifetime()?.check()
97    }
98    pub(crate) fn lifetime(&self) -> Result<Arc<ReplyLifetime>, Error> {
99        let lifetime = self.inner.lifetime.upgrade().ok_or_else(unavailable)?;
100        lifetime.check()?;
101        Ok(lifetime)
102    }
103    pub(crate) async fn send(
104        &self,
105        npdu: Bytes,
106        priority: bacnet_npdu::NetworkPriority,
107        owner: ResponseScopeGuard,
108        on_issuance: impl FnOnce() + Send + 'static,
109    ) -> Result<(), Error> {
110        self.check()?;
111        if !owner.is_open() {
112            return Err(unavailable());
113        }
114        let (done, completed) = oneshot::channel();
115        {
116            let outputs = self.inner.outputs.upgrade().ok_or_else(unavailable)?;
117            let queue = outputs
118                .send_txs
119                .get(usize::from(self.inner.plan.port().0))
120                .ok_or_else(unavailable)?;
121            queue
122                .try_send(
123                    priority,
124                    SendRequest::Unicast {
125                        npdu,
126                        mac: MacAddr::from_slice(self.inner.plan.peer()),
127                        data_attributes: vec![],
128                        reject_origin: None,
129                        response: Some(ReplyCompletion {
130                            done,
131                            owner,
132                            lifetime: self.inner.lifetime.clone(),
133                            on_issuance: Box::new(on_issuance),
134                        }),
135                    },
136                )
137                .map_err(|_| unavailable())?;
138        }
139        completed.await.map_err(|_| unavailable())?
140    }
141}
142
143pub(super) struct ReplyCompletion {
144    done: oneshot::Sender<Result<(), Error>>,
145    owner: ResponseScopeGuard,
146    lifetime: Weak<ReplyLifetime>,
147    on_issuance: Box<dyn FnOnce() + Send>,
148}
149
150impl std::fmt::Debug for ReplyCompletion {
151    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
152        f.debug_struct("ReplyCompletion").finish_non_exhaustive()
153    }
154}
155
156impl ReplyCompletion {
157    /// Cancellation or owner retirement before selection emits no bytes.
158    pub(super) fn start(self) -> Option<oneshot::Sender<Result<(), Error>>> {
159        if !self.owner.is_open()
160            || self.done.is_closed()
161            || self
162                .lifetime
163                .upgrade()
164                .is_none_or(|lifetime| lifetime.check().is_err())
165        {
166            let _ = self.done.send(Err(unavailable()));
167            return None;
168        }
169        (self.on_issuance)();
170        Some(self.done)
171    }
172}
173
174fn unavailable() -> Error {
175    Error::Transport(std::io::Error::new(
176        std::io::ErrorKind::NotConnected,
177        "original router response route unavailable",
178    ))
179}
bacnet_network/router/replies.rs

bacnet_network/router/
replies.rs
