bacnet_server/server/
network_port.rs

1//! Explicit single B/IP port admission and cancellation-owned startup cleanup.
2//!
3//! The registered port may run in NORMAL, FOREIGN or BBMD mode. Startup
4//! publishes the mode with the actual bind; in BBMD mode the Network Port
5//! object then reads the transport's own tables on each property read.
6use super::*;
7use bacnet_network::layer::ReceivedApdu;
8
9impl<T: TransportPort + 'static> ServerBuilder<T> {
10    /// Select the concrete built-in Network Port for this owned B/IP link, in
11    /// whichever B/IP mode the transport runs. The object must already
12    /// describe the concrete interface and configured UDP port (zero is
13    /// allowed). Startup validates, then publishes the actual bind and mode.
14    pub fn registered_network_port(mut self, oid: ObjectIdentifier) -> Self {
15        self.config.registered_network_port = Some(oid);
16        self
17    }
18}
19impl BipServerBuilder {
20    /// Select the built-in Network Port that represents this owned B/IP link.
21    pub fn registered_network_port(mut self, oid: ObjectIdentifier) -> Self {
22        self.config.registered_network_port = Some(oid);
23        self
24    }
25}
26
27pub(super) fn prepare<T: TransportPort>(
28    db: &mut ObjectDatabase,
29    transport: &mut T,
30    oid: Option<ObjectIdentifier>,
31) -> Result<(), Error> {
32    let Some(oid) = oid else {
33        return Ok(());
34    };
35    let address = match transport.bip_port() {
36        Some(port) => port.endpoint,
37        // A B/IP link that reports no mode is configured both ways.
38        None if transport.bip_broadcast_endpoint().is_some() => {
39            return Err(Error::Encoding(
40                "registered Network Port refused: a BBMD that also registers as a foreign \
41                 device has no single B/IP mode"
42                    .into(),
43            ))
44        }
45        None => return Err(Error::Encoding("registered port requires B/IP".into())),
46    };
47    let ip = *address.ip();
48    if ip.is_unspecified()
49        || ip.is_multicast()
50        || ip.is_broadcast()
51        || transport
52            .bip_broadcast_endpoint()
53            .is_some_and(|broadcast| *broadcast.ip() == ip)
54    {
55        return Err(Error::Encoding(
56            "registered B/IP requires a concrete unicast interface".into(),
57        ));
58    }
59    let (_, lease) = db.reserve_bip_port_internal(oid, ip.octets(), address.port())?;
60    transport.retain_network_port_lease_internal(lease)
61}
62
63pub(super) fn publish<T: TransportPort + 'static>(
64    db: &mut ObjectDatabase,
65    network: &NetworkLayer<T>,
66    oid: Option<ObjectIdentifier>,
67) -> Result<(), Error> {
68    let Some(oid) = oid else {
69        return Ok(());
70    };
71    let port = network
72        .transport()
73        .bip_port()
74        .ok_or_else(|| Error::Encoding("registered B/IP mode changed during bind".into()))?;
75    db.publish_bip_port_internal(
76        oid,
77        port.endpoint.ip().octets(),
78        port.endpoint.port(),
79        network.transport().local_receive_apdu_capacity() as u32,
80        port.mode,
81    )?;
82    // The port's configured number is the local network number from the
83    // start; the control worker copies every later change from the port.
84    if let Some(state) = db.network_number_internal(oid, None) {
85        network.local_network_number().publish(state);
86    }
87    Ok(())
88}
89
90/// Failed/cancelled startup still owns cleanup. Capturing the runtime at creation
91/// lets Drop transfer the network even when its caller drops the future elsewhere.
92pub(super) struct StartingNetwork<T: TransportPort + 'static> {
93    network: Option<NetworkLayer<T>>,
94    runtime: tokio::runtime::Handle,
95}
96impl<T: TransportPort + 'static> StartingNetwork<T> {
97    pub(super) fn new(transport: T) -> Self {
98        Self {
99            network: Some(NetworkLayer::new(transport)),
100            runtime: tokio::runtime::Handle::current(),
101        }
102    }
103    pub(super) fn from_network(network: NetworkLayer<T>) -> Self {
104        Self {
105            network: Some(network),
106            runtime: tokio::runtime::Handle::current(),
107        }
108    }
109    pub(super) fn network(&mut self) -> &mut NetworkLayer<T> {
110        self.network.as_mut().expect("startup network owned")
111    }
112    pub(super) fn finish(mut self) -> NetworkLayer<T> {
113        self.network.take().expect("startup network owned")
114    }
115    pub(super) async fn cleanup(mut self) -> Result<(), Error> {
116        let mut network = self.network.take().expect("startup network owned");
117        self.runtime
118            .spawn(async move { network.stop().await })
119            .await
120            .map_err(|error| Error::Encoding(format!("startup cleanup failed: {error}")))?
121    }
122}
123impl<T: TransportPort + 'static> Drop for StartingNetwork<T> {
124    fn drop(&mut self) {
125        if let Some(mut network) = self.network.take() {
126            self.runtime.spawn(async move {
127                if let Err(error) = network.stop().await {
128                    tracing::warn!(%error, "cancelled startup cleanup failed");
129                }
130            });
131        }
132    }
133}
134
135/// Bind, validate the selected port and finalize Audit routes under one cleanup owner.
136pub(super) async fn start<T: TransportPort + 'static>(
137    db: &mut ObjectDatabase,
138    config: &ServerConfig,
139    mut transport: T,
140    audit_routes: super::audit_recipient_routes::AuditRoutes,
141) -> Result<
142    (
143        NetworkLayer<T>,
144        mpsc::Receiver<ReceivedApdu>,
145        Arc<super::audit_recipient_routes::AuditRoutes>,
146        Option<mpsc::Receiver<bacnet_network::layer::ReceivedNetworkControl>>,
147    ),
148    Error,
149> {
150    prepare(db, &mut transport, config.registered_network_port)?;
151    let mut starting = StartingNetwork::new(transport);
152    let started = async {
153        let controls = if starting
154            .network()
155            .transport()
156            .supports_local_nonrouter_number_controls()
157        {
158            Some(starting.network().enable_network_control_receiver()?)
159        } else {
160            None
161        };
162        let apdu_rx = starting.network().start().await?;
163        publish(db, starting.network(), config.registered_network_port)?;
164        let routes = audit_routes.finish(db, config, starting.network()).await?;
165        let controls = starting
166            .network()
167            .transport()
168            .supports_local_nonrouter_number_controls()
169            .then_some(controls)
170            .flatten();
171        Ok::<_, Error>((apdu_rx, routes, controls))
172    }
173    .await;
174    match started {
175        Ok((apdu_rx, routes, controls)) => Ok((starting.finish(), apdu_rx, routes, controls)),
176        Err(error) => {
177            starting.cleanup().await?;
178            Err(error)
179        }
180    }
181}
182
183/// Own the Number controls. When one changes the published number, target
184/// Audit re-evaluates its recipient's route (#1460).
185pub(super) fn spawn_number_worker<T: TransportPort + 'static>(
186    network: &Arc<NetworkLayer<T>>,
187    db: &Arc<RwLock<ObjectDatabase>>,
188    selected: Option<ObjectIdentifier>,
189    mut controls: mpsc::Receiver<bacnet_network::layer::ReceivedNetworkControl>,
190    audit: Option<std::sync::Weak<super::audit_recipient::TargetAudit<T>>>,
191) -> JoinHandle<()> {
192    let network = Arc::clone(network);
193    let mut owner =
194        crate::network_number::NetworkNumberOwner::new(selected.map(|oid| (Arc::clone(db), oid)))
195            .publishing_to(network.local_network_number().clone());
196    super::heap_futures::spawn_boxed(move || async move {
197        while let Some(control) = controls.recv().await {
198            let before = network.local_network_number().get();
199            let reply = owner.handle(control).await;
200            if network.local_network_number().get() != before {
201                if let Some(audit) = audit.as_ref().and_then(std::sync::Weak::upgrade) {
202                    audit.number_changed();
203                }
204            }
205            if let Some(npdu) = reply {
206                if let Err(error) = network.transport().send_broadcast(&npdu).await {
207                    tracing::debug!(%error, "Network-Number-Is broadcast failed");
208                }
209            }
210        }
211    })
212}
213
214pub(super) fn validate_apdu_capacity<T: TransportPort>(
215    config: &mut ServerConfig,
216    transport: &T,
217) -> Result<(), Error> {
218    let transport_max = transport.local_receive_apdu_capacity() as u32;
219    config.max_apdu_length = config.max_apdu_length.min(transport_max);
220    if config.max_apdu_length < u32::from(apdu::MINIMUM_MESSAGE_SIZE) {
221        return Err(Error::Encoding(format!(
222            "local max_apdu_length {} is below 50",
223            config.max_apdu_length
224        )));
225    }
226    Ok(())
227}
bacnet_server/server/network_port.rs

bacnet_server/server/
network_port.rs
