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