bacnet_client/client/builder_build.rs
1//! Shared transport preparation for concrete and transport-erased clients.
2use super::*;
3use bacnet_transport::{any::AnyTransport, mstp::SerialPort};
4
5impl BipClientBuilder {
6 fn prepare(self) -> Result<(ClientConfig, BipTransport, ClientOptions), Error> {
7 let mut transport = BipTransport::new(
8 self.config.interface,
9 self.config.port,
10 self.config.broadcast_address,
11 );
12 transport.set_share_port_by_address(self.config.share_port_by_address);
13 transport.set_mode(self.mode);
14 transport.set_multicast(self.multicast)?;
15 Ok((self.config, transport, self.options))
16 }
17
18 /// Build and start the client, constructing a BipTransport from the config.
19 pub async fn build(self) -> Result<BACnetClient<BipTransport>, Error> {
20 Box::pin(async move {
21 let (config, transport, options) = self.prepare()?;
22 Box::pin(BACnetClient::start_with_options(config, transport, options)).await
23 })
24 .await
25 }
26
27 /// Build and start a configured B/IP client with a transport-erased type.
28 ///
29 /// Uses the same settings and validation as [`Self::build`]. Choose
30 /// [`NoSerial`](bacnet_transport::mstp::NoSerial) when MS/TP is not needed;
31 /// no serial-port value is constructed or required.
32 ///
33 /// ```no_run
34 /// use bacnet_client::client::BACnetClient;
35 /// use bacnet_transport::{any::AnyTransport, mstp::NoSerial};
36 /// # async fn example() -> Result<(), bacnet_types::error::Error> {
37 /// let mut client: BACnetClient<AnyTransport<NoSerial>> = BACnetClient::bip_builder()
38 /// .port(0)
39 /// .apdu_retries(2)
40 /// .build_any::<NoSerial>().await?;
41 /// client.stop().await?;
42 /// # Ok(()) }
43 /// ```
44 pub async fn build_any<S: SerialPort>(self) -> Result<BACnetClient<AnyTransport<S>>, Error> {
45 Box::pin(async move {
46 let (config, transport, options) = self.prepare()?;
47 Box::pin(BACnetClient::start_with_options(
48 config,
49 transport.into(),
50 options,
51 ))
52 .await
53 })
54 .await
55 }
56}
57
58#[cfg(feature = "ipv6")]
59impl Bip6ClientBuilder {
60 fn prepare(self) -> (ClientConfig, Bip6Transport, ClientOptions) {
61 let transport = Bip6Transport::new(self.interface, self.config.port, self.device_instance);
62 (self.config, transport, self.options)
63 }
64
65 /// Build and start the client, constructing a Bip6Transport from the config.
66 pub async fn build(self) -> Result<BACnetClient<Bip6Transport>, Error> {
67 Box::pin(async move {
68 let (config, transport, options) = self.prepare();
69 Box::pin(BACnetClient::start_with_options(config, transport, options)).await
70 })
71 .await
72 }
73
74 /// Build and start a configured B/IPv6 client with a transport-erased type.
75 ///
76 /// Uses the same interface, device identity and client options as
77 /// [`Self::build`]. No serial-port value is required.
78 ///
79 /// ```no_run
80 /// use bacnet_client::client::BACnetClient;
81 /// use bacnet_transport::{any::AnyTransport, mstp::NoSerial};
82 /// # async fn example() -> Result<(), bacnet_types::error::Error> {
83 /// let mut client: BACnetClient<AnyTransport<NoSerial>> = BACnetClient::bip6_builder()
84 /// .device_instance(123)
85 /// .build_any::<NoSerial>().await?;
86 /// client.stop().await?;
87 /// # Ok(()) }
88 /// ```
89 pub async fn build_any<S: SerialPort>(self) -> Result<BACnetClient<AnyTransport<S>>, Error> {
90 Box::pin(async move {
91 let (config, transport, options) = self.prepare();
92 Box::pin(BACnetClient::start_with_options(
93 config,
94 transport.into(),
95 options,
96 ))
97 .await
98 })
99 .await
100 }
101}
102
103#[cfg(feature = "sc-tls")]
104impl ScClientBuilder {
105 pub(super) fn validate(&self) -> Result<(), Error> {
106 self.npdu_admission_policy.validate()?;
107 if let Some(config) = &self.reconnect {
108 config.validate()?;
109 }
110 if self.failover_hub_url.is_some() && self.reconnect.is_none() {
111 // Without reconnection a lost connection is never re-established,
112 // so the failover hub would never be reached or left again.
113 return Err(Error::Encoding(
114 "SC client builder: failover_hub_url requires reconnect".into(),
115 ));
116 }
117 self.validate_identity()?;
118 self.options.validate()?;
119 self.config.validate_request_settings()?;
120 self.config.validate_window()
121 }
122
123 async fn prepare(
124 self,
125 ) -> Result<
126 (
127 ClientConfig,
128 bacnet_transport::sc::ScTransport<bacnet_transport::sc_tls::TlsWebSocket>,
129 ClientOptions,
130 ),
131 Error,
132 > {
133 self.validate()?;
134 let tls_config =
135 self.tls_config.as_ref().cloned().ok_or_else(|| {
136 Error::Encoding("SC client builder: tls_config is required".into())
137 })?;
138 use bacnet_transport::sc::ScActiveHub;
139 use bacnet_transport::sc_tls::TlsWebSocket;
140 TlsWebSocket::validate_url(&self.hub_url)?;
141 if let Some(url) = &self.failover_hub_url {
142 TlsWebSocket::validate_url(url)?;
143 }
144 // An unreachable primary starts the client on the failover hub; the
145 // transport restores the primary once it answers (Annex AB.5.2).
146 // Bounded like the transport's own redials: a blackholed or stalled
147 // hub must not hold up build() or the failover fallback.
148 let limit = std::time::Duration::from_millis(self.connect_timeout_ms);
149 let dial = |url: &str| {
150 let tls = tls_config.clone();
151 let url = url.to_owned();
152 async move {
153 bacnet_transport::sc::dial_within(limit, Box::pin(TlsWebSocket::connect(&url, tls)))
154 .await
155 }
156 };
157 let (ws, hub) = match dial(&self.hub_url).await {
158 Ok(ws) => (ws, ScActiveHub::Primary),
159 Err(primary) => match &self.failover_hub_url {
160 Some(url) => {
161 // No transport (or event stream) exists yet to report it.
162 tracing::warn!(error = %primary, "BACnet/SC primary hub dial failed; trying the failover hub");
163 match dial(url).await {
164 Ok(ws) => (ws, ScActiveHub::Failover),
165 Err(_) => return Err(primary),
166 }
167 }
168 None => return Err(primary),
169 },
170 };
171 let mut transport = self.sc_transport(ws).with_initial_hub(hub);
172 if let Some(failover_url) = self.failover_hub_url {
173 let tls_config = tls_config.clone();
174 transport = transport.with_failover_connector(move || {
175 let failover_url = failover_url.clone();
176 let tls_config = tls_config.clone();
177 async move {
178 bacnet_transport::sc_tls::TlsWebSocket::connect(&failover_url, tls_config).await
179 }
180 });
181 }
182 if let Some(rc) = self.reconnect {
183 let hub_url = self.hub_url;
184 transport = transport
185 .with_connector(move || {
186 let hub_url = hub_url.clone();
187 let tls_config = tls_config.clone();
188 async move {
189 bacnet_transport::sc_tls::TlsWebSocket::connect(&hub_url, tls_config).await
190 }
191 })
192 .with_reconnect(rc);
193 }
194 Ok((self.config, transport, self.options))
195 }
196
197 /// Connect to the hub and start the client.
198 ///
199 /// NPDU admission and reconnect configuration are validated before TLS lookup or dialing, and
200 /// again when the transport starts. An error still consumes this builder
201 /// and drops its inputs; this does not promise generic endpoint rollback.
202 ///
203 /// The client's [`transport()`](BACnetClient::transport) lends the hub
204 /// connection-state watch, hub-connector events and link status, accepted
205 /// Hub certificate observation and NPDU drop counts:
206 ///
207 /// ```no_run
208 /// use bacnet_client::client::BACnetClient;
209 /// use bacnet_transport::sc::{ScConnectionState, ScEvent, ScReconnectConfig};
210 /// use bacnet_transport::sc_tls::ScNodeTlsConfig;
211 ///
212 /// async fn watch_hub(tls: ScNodeTlsConfig) -> Result<(), bacnet_types::error::Error> {
213 /// let client = BACnetClient::sc_builder()
214 /// .hub_url("wss://hub.example.com/bacnet")
215 /// .failover_hub_url("wss://failover.example.com/bacnet")
216 /// .reconnect(ScReconnectConfig::default())
217 /// .tls_config(tls)
218 /// .vmac([0x02, 0, 0, 0, 0, 0x01])
219 /// .device_uuid([0x42; 16])
220 /// .build()
221 /// .await?;
222 /// // Owned receivers: they can live in their own tasks.
223 /// let mut state = client.transport().connection_state_changes();
224 /// tokio::spawn(async move {
225 /// while state.changed().await.is_ok() {
226 /// let _connected = *state.borrow_and_update() == ScConnectionState::Connected;
227 /// }
228 /// });
229 /// let mut events = client.transport().connection_events();
230 /// tokio::spawn(async move {
231 /// use tokio::sync::broadcast::error::RecvError;
232 /// loop {
233 /// match events.recv().await {
234 /// Ok(event) => {
235 /// let _failover = matches!(event, ScEvent::FailoverEngaged);
236 /// }
237 /// // Fell behind: the oldest events were dropped, not waited for.
238 /// Err(RecvError::Lagged(_)) => continue,
239 /// Err(RecvError::Closed) => break,
240 /// }
241 /// }
242 /// });
243 /// // Active hub, hub VMAC and heartbeat timing: poll when redrawing.
244 /// let _status = client.transport().connection_status();
245 /// // Owned historical TLS data, present only for an accepted active Hub.
246 /// let _peer = client.transport().hub_peer_info().await;
247 /// // A snapshot: poll it as often as needed.
248 /// let _full_queue_drops = client.transport().npdu_drop_counts().full_drops;
249 /// Ok(())
250 /// }
251 /// ```
252 pub async fn build(
253 self,
254 ) -> Result<
255 BACnetClient<bacnet_transport::sc::ScTransport<bacnet_transport::sc_tls::TlsWebSocket>>,
256 Error,
257 > {
258 Box::pin(async move {
259 let (config, transport, options) = Box::pin(self.prepare()).await?;
260 Box::pin(BACnetClient::start_with_options(config, transport, options)).await
261 })
262 .await
263 }
264
265 /// Connect to the hub and start the client, as [`build`](Self::build)
266 /// does, and also return a hub-connector event receiver subscribed
267 /// before the client started.
268 ///
269 /// The first [`ScEvent::Connected`](bacnet_transport::sc::ScEvent::Connected)
270 /// goes out while `build` runs, before a caller of
271 /// `transport().connection_events()` can subscribe. This receiver is
272 /// already subscribed then, so it holds that event, followed by every
273 /// later one, with the same lag rules as `connection_events()`.
274 ///
275 /// ```no_run
276 /// use bacnet_client::client::BACnetClient;
277 /// use bacnet_transport::sc::ScEvent;
278 /// use bacnet_transport::sc_tls::ScNodeTlsConfig;
279 ///
280 /// async fn first_event(tls: ScNodeTlsConfig) -> Result<(), bacnet_types::error::Error> {
281 /// let (client, mut events) = BACnetClient::sc_builder()
282 /// .hub_url("wss://hub.example.com/bacnet")
283 /// .tls_config(tls)
284 /// .vmac([0x02, 0, 0, 0, 0, 0x01])
285 /// .device_uuid([0x42; 16])
286 /// .build_with_events()
287 /// .await?;
288 /// let first = events.recv().await;
289 /// assert!(matches!(first, Ok(ScEvent::Connected { .. })));
290 /// drop(client);
291 /// Ok(())
292 /// }
293 /// ```
294 pub async fn build_with_events(
295 self,
296 ) -> Result<
297 (
298 BACnetClient<bacnet_transport::sc::ScTransport<bacnet_transport::sc_tls::TlsWebSocket>>,
299 tokio::sync::broadcast::Receiver<bacnet_transport::sc::ScEvent>,
300 ),
301 Error,
302 > {
303 Box::pin(async move {
304 let (config, transport, options, events) = Box::pin(self.prepare_subscribed()).await?;
305 let client =
306 Box::pin(BACnetClient::start_with_options(config, transport, options)).await?;
307 Ok((client, events))
308 })
309 .await
310 }
311
312 /// [`prepare`](Self::prepare), then subscribe to the transport's events
313 /// before anything starts it. The first `Connected` goes out while the
314 /// client starts, so every `*_with_events` build goes through here.
315 async fn prepare_subscribed(
316 self,
317 ) -> Result<
318 (
319 ClientConfig,
320 bacnet_transport::sc::ScTransport<bacnet_transport::sc_tls::TlsWebSocket>,
321 ClientOptions,
322 tokio::sync::broadcast::Receiver<bacnet_transport::sc::ScEvent>,
323 ),
324 Error,
325 > {
326 let (config, transport, options) = Box::pin(self.prepare()).await?;
327 let events = transport.connection_events();
328 Ok((config, transport, options, events))
329 }
330
331 /// Connect to the hub and start a transport-erased BACnet/SC client.
332 ///
333 /// Shares [`Self::build`]'s validation, TLS policy, identity, heartbeat and
334 /// optional reconnect configuration. Identity, NPDU admission, reconnect and client options
335 /// are checked before dialing; heartbeat timing is checked when the transport
336 /// starts. No serial-port value is required.
337 ///
338 /// ```no_run
339 /// use bacnet_client::client::BACnetClient;
340 /// use bacnet_transport::{any::AnyTransport, mstp::NoSerial, sc::ScReconnectConfig};
341 /// use bacnet_transport::sc_tls::ScNodeTlsConfig;
342 /// async fn connect(tls: ScNodeTlsConfig) -> Result<BACnetClient<AnyTransport<NoSerial>>,
343 /// bacnet_types::error::Error> {
344 /// BACnetClient::sc_builder()
345 /// .hub_url("wss://hub.example.com/bacnet")
346 /// .tls_config(tls)
347 /// .vmac([2, 0, 0, 0, 0, 1])
348 /// .device_uuid([0x42; 16])
349 /// .reconnect(ScReconnectConfig::default())
350 /// .build_any::<NoSerial>().await
351 /// }
352 /// ```
353 pub async fn build_any<S: SerialPort>(self) -> Result<BACnetClient<AnyTransport<S>>, Error> {
354 Box::pin(async move {
355 let (config, transport, options) = Box::pin(self.prepare()).await?;
356 Box::pin(BACnetClient::start_with_options(
357 config,
358 transport.into(),
359 options,
360 ))
361 .await
362 })
363 .await
364 }
365
366 /// Connect to the hub and start a transport-erased BACnet/SC client, as
367 /// [`build_any`](Self::build_any) does, and also return a hub-connector
368 /// event receiver subscribed before the client started (#1909).
369 ///
370 /// As with [`build_with_events`](Self::build_with_events), the receiver
371 /// holds the first
372 /// [`ScEvent::Connected`](bacnet_transport::sc::ScEvent::Connected),
373 /// which a later `connection_events()` call misses.
374 ///
375 /// ```no_run
376 /// use bacnet_client::client::BACnetClient;
377 /// use bacnet_transport::{mstp::NoSerial, sc::ScEvent};
378 /// use bacnet_transport::sc_tls::ScNodeTlsConfig;
379 ///
380 /// async fn first_event(tls: ScNodeTlsConfig) -> Result<(), bacnet_types::error::Error> {
381 /// let (client, mut events) = BACnetClient::sc_builder()
382 /// .hub_url("wss://hub.example.com/bacnet")
383 /// .tls_config(tls)
384 /// .vmac([0x02, 0, 0, 0, 0, 0x01])
385 /// .device_uuid([0x42; 16])
386 /// .build_any_with_events::<NoSerial>()
387 /// .await?;
388 /// assert!(matches!(events.recv().await, Ok(ScEvent::Connected { .. })));
389 /// drop(client);
390 /// Ok(())
391 /// }
392 /// ```
393 pub async fn build_any_with_events<S: SerialPort>(
394 self,
395 ) -> Result<
396 (
397 BACnetClient<AnyTransport<S>>,
398 tokio::sync::broadcast::Receiver<bacnet_transport::sc::ScEvent>,
399 ),
400 Error,
401 > {
402 Box::pin(async move {
403 let (config, transport, options, events) = Box::pin(self.prepare_subscribed()).await?;
404 let client = Box::pin(BACnetClient::start_with_options(
405 config,
406 transport.into(),
407 options,
408 ))
409 .await?;
410 Ok((client, events))
411 })
412 .await
413 }
414}