bacnet_server/server/
network_port.rs

1//! Explicit single B/IP or B/IPv6 port admission and cancellation-owned
2//! startup cleanup.
3//!
4//! The registered port may run in NORMAL, FOREIGN or BBMD mode. Startup
5//! publishes the mode with the actual bind; in BBMD mode the Network Port
6//! object then reads the transport's own tables on each property read. A
7//! B/IPv6 port registers the same way, through
8//! [`TransportPort::bip6_port`], and publishes its VMAC and multicast group
9//! too.
10use super::*;
11use crate::network_port_registration as registration;
12
13impl<T: TransportPort + 'static> ServerBuilder<T> {
14    /// Select the concrete built-in Network Port for this owned B/IP or
15    /// B/IPv6 link, in whichever mode the transport runs. The object must
16    /// already describe the concrete interface and configured UDP port
17    /// (zero is allowed): an IPV4 object for B/IP, an IPV6 one for B/IPv6.
18    /// A B/IPv6 transport must be built with the served Device's instance,
19    /// which is its VMAC (#2057). Startup validates, then publishes the
20    /// actual bind and mode.
21    pub fn registered_network_port(mut self, oid: ObjectIdentifier) -> Self {
22        self.config.registered_network_port = Some(oid);
23        self
24    }
25}
26impl BipServerBuilder {
27    /// Select the built-in Network Port that represents this owned B/IP link.
28    pub fn registered_network_port(mut self, oid: ObjectIdentifier) -> Self {
29        self.config.registered_network_port = Some(oid);
30        self
31    }
32}
33
34/// Validate and reserve the registered port `oid`, if any, before the bind.
35/// `device_instance` is the served Device's, which a B/IPv6 VMAC must match.
36pub(super) fn prepare<T: TransportPort>(
37    db: &mut ObjectDatabase,
38    transport: &mut T,
39    oid: Option<ObjectIdentifier>,
40    device_instance: u32,
41) -> Result<(), Error> {
42    let Some(oid) = oid else {
43        return Ok(());
44    };
45    let address = match transport.bip_port() {
46        Some(port) => registration::bip_address(&port, transport.bip_broadcast_endpoint())?,
47        None if transport.bip6_port().is_some() => {
48            return prepare_bip6(db, transport, oid, device_instance);
49        }
50        // A B/IP link that reports no mode is configured both ways.
51        None if transport.bip_broadcast_endpoint().is_some() => {
52            return Err(Error::Encoding(
53                "registered Network Port refused: a BBMD that also registers as a foreign \
54                 device has no single B/IP mode"
55                    .into(),
56            ));
57        }
58        None => {
59            return Err(Error::Encoding(
60                "registered port requires B/IP or B/IPv6".into(),
61            ));
62        }
63    };
64    let (_, lease) = db.reserve_bip_port_internal(oid, address.ip().octets(), address.port())?;
65    transport.retain_network_port_lease_internal(lease)
66}
67
68/// [`prepare`] for a B/IPv6 link: its configured address must be one
69/// concrete unicast address, as a B/IP link's must, and its VMAC the served
70/// Device's (#2057).
71fn prepare_bip6<T: TransportPort>(
72    db: &mut ObjectDatabase,
73    transport: &mut T,
74    oid: ObjectIdentifier,
75    device_instance: u32,
76) -> Result<(), Error> {
77    let port = transport
78        .bip6_port()
79        .ok_or_else(|| Error::Encoding("registered port requires B/IP or B/IPv6".into()))?;
80    let address = registration::bip6_address(&port, Some(device_instance))?;
81    let (_, lease) = db.reserve_bip6_port_internal(oid, address.ip().octets(), address.port())?;
82    transport.retain_network_port_lease_internal(lease)
83}
84
85pub(super) fn publish<T: TransportPort + 'static>(
86    db: &mut ObjectDatabase,
87    network: &NetworkLayer<T>,
88    oid: Option<ObjectIdentifier>,
89    activation: bool,
90) -> Result<(), Error> {
91    let Some(oid) = oid else {
92        return Ok(());
93    };
94    let transport = network.transport();
95    let capacity = transport.local_receive_apdu_capacity() as u32;
96    match transport.bip_port() {
97        Some(port) => {
98            registration::publish_bip(db, oid, port, capacity)?;
99        }
100        _ => {
101            let port = transport.bip6_port().ok_or_else(|| {
102                Error::Encoding("registered B/IP mode changed during bind".into())
103            })?;
104            registration::publish_bip6(db, oid, port, capacity)?;
105        }
106    }
107    // Written values activate only through a ReinitializeDevice handler.
108    db.set_network_port_activation_internal(oid, activation)?;
109    // The port's configured number is the local network number from the
110    // start; the control worker copies every later change from the port.
111    if let Some(state) = db.network_number_internal(oid, None) {
112        network.local_network_number().publish(state);
113    }
114    Ok(())
115}
116
117/// Failed/cancelled startup still owns cleanup. Capturing the runtime at creation
118/// lets Drop transfer the network even when its caller drops the future elsewhere.
119pub(super) struct StartingNetwork<T: TransportPort + 'static> {
120    network: Option<NetworkLayer<T>>,
121    runtime: tokio::runtime::Handle,
122}
123impl<T: TransportPort + 'static> StartingNetwork<T> {
124    pub(super) fn new(transport: T) -> Self {
125        Self {
126            network: Some(NetworkLayer::new(transport)),
127            runtime: tokio::runtime::Handle::current(),
128        }
129    }
130    pub(super) fn from_network(network: NetworkLayer<T>) -> Self {
131        Self {
132            network: Some(network),
133            runtime: tokio::runtime::Handle::current(),
134        }
135    }
136    pub(super) fn network(&mut self) -> &mut NetworkLayer<T> {
137        self.network.as_mut().expect("startup network owned")
138    }
139    pub(super) fn finish(mut self) -> NetworkLayer<T> {
140        self.network.take().expect("startup network owned")
141    }
142    pub(super) async fn cleanup(mut self) -> Result<(), Error> {
143        let mut network = self.network.take().expect("startup network owned");
144        self.runtime
145            .spawn(async move { network.stop().await })
146            .await
147            .map_err(|error| Error::Encoding(format!("startup cleanup failed: {error}")))?
148    }
149}
150impl<T: TransportPort + 'static> Drop for StartingNetwork<T> {
151    fn drop(&mut self) {
152        if let Some(mut network) = self.network.take() {
153            self.runtime.spawn(async move {
154                if let Err(error) = network.stop().await {
155                    tracing::warn!(%error, "cancelled startup cleanup failed");
156                }
157            });
158        }
159    }
160}
161
162/// Bind, validate the selected port and finalize Audit routes under one cleanup owner.
163pub(super) async fn start<T: TransportPort + 'static>(
164    db: &mut ObjectDatabase,
165    config: &ServerConfig,
166    mut transport: T,
167    audit_routes: super::audit_recipient_routes::AuditRoutes,
168) -> Result<
169    (
170        NetworkLayer<T>,
171        bacnet_network::layer::ApduReceiver,
172        Arc<super::audit_recipient_routes::AuditRoutes>,
173        Option<mpsc::Receiver<bacnet_network::layer::ReceivedNetworkControl>>,
174    ),
175    Error,
176> {
177    prepare(
178        db,
179        &mut transport,
180        config.registered_network_port,
181        config.device.instance(),
182    )?;
183    let mut starting = StartingNetwork::new(transport);
184    let started = async {
185        let controls = if starting
186            .network()
187            .transport()
188            .supports_local_nonrouter_number_controls()
189        {
190            Some(starting.network().enable_network_control_receiver()?)
191        } else {
192            None
193        };
194        let apdu_rx = starting.network().start().await?;
195        publish(
196            db,
197            starting.network(),
198            config.registered_network_port,
199            config.on_reinitialize.is_some(),
200        )?;
201        let routes = audit_routes.finish(db, config, starting.network()).await?;
202        let controls = starting
203            .network()
204            .transport()
205            .supports_local_nonrouter_number_controls()
206            .then_some(controls)
207            .flatten();
208        Ok::<_, Error>((apdu_rx, routes, controls))
209    }
210    .await;
211    match started {
212        Ok((apdu_rx, routes, controls)) => Ok((starting.finish(), apdu_rx, routes, controls)),
213        Err(error) => {
214            starting.cleanup().await?;
215            Err(error)
216        }
217    }
218}
219
220/// Own the Number controls. When one changes the published number, target
221/// Audit re-evaluates its recipient's route (#1460).
222pub(super) fn spawn_number_worker<T: TransportPort + 'static>(
223    network: &Arc<NetworkLayer<T>>,
224    db: &Arc<RwLock<ObjectDatabase>>,
225    selected: Option<ObjectIdentifier>,
226    mut controls: mpsc::Receiver<bacnet_network::layer::ReceivedNetworkControl>,
227    audit: Option<std::sync::Weak<super::audit_recipient::TargetAudit<T>>>,
228) -> JoinHandle<()> {
229    let network = Arc::clone(network);
230    let mut owner =
231        crate::network_number::NetworkNumberOwner::new(selected.map(|oid| (Arc::clone(db), oid)))
232            .publishing_to(network.local_network_number().clone());
233    super::heap_futures::spawn_boxed(move || async move {
234        while let Some(control) = controls.recv().await {
235            let before = network.local_network_number().get();
236            let reply = owner.handle(control).await;
237            if network.local_network_number().get() != before
238                && let Some(audit) = audit.as_ref().and_then(std::sync::Weak::upgrade)
239            {
240                audit.number_changed();
241            }
242            if let Some(npdu) = reply
243                && let Err(error) = network.transport().send_broadcast(&npdu).await
244            {
245                tracing::debug!(%error, "Network-Number-Is broadcast failed");
246            }
247        }
248    })
249}
250
251/// Resolve the Device's Max_APDU_Length_Accepted against the transport's
252/// receive capacity, once, at start.
253///
254/// The result stays fixed while the server runs, even if the transport's
255/// capacity later falls (a BACnet/SC hub failover, a router port whose path
256/// shrinks; #2093). Peers read the value once, from I-Am or their address
257/// binding, and cache it, so changing the property afterwards would not
258/// reach them. Replies stay safe without it: each response is sized to its
259/// own route's budget (`ResponseRoute::max_apdu_length`, the requester's
260/// declared limit included), so a smaller path still gets replies that fit.
261pub(super) fn validate_apdu_capacity<T: TransportPort>(
262    config: &mut ServerConfig,
263    transport: &T,
264) -> Result<(), Error> {
265    config.device = config
266        .device
267        .resolve(transport.local_receive_apdu_capacity())?;
268    Ok(())
269}
bacnet_server/server/network_port.rs

bacnet_server/server/
network_port.rs
