bacnet_server/server/
network_port.rs

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

bacnet_server/server/
network_port.rs
