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

bacnet_server/server/
network_port.rs
