bacnet_client/client/
lifecycle.rs

1use super::*;
2
3const DEVICE_PURGE_INTERVAL: Duration = Duration::from_secs(300);
4#[cfg(not(test))]
5const DEVICE_MAX_AGE: Duration = Duration::from_secs(600);
6// Unit tests use Tokio's paused clock, while DeviceTable timestamps use
7// std::time::Instant. Use an immediate threshold so the purge timer is
8// observable without constructing an unrepresentable old Instant on Windows.
9#[cfg(test)]
10const DEVICE_MAX_AGE: Duration = Duration::ZERO;
11
12impl<T: TransportPort> BACnetClient<T> {
13    fn abort_dispatch_task(&mut self) -> Option<JoinHandle<()>> {
14        let task = self.dispatch_task.take()?;
15        task.abort();
16        Some(task)
17    }
18}
19
20impl<T: TransportPort + 'static> BACnetClient<T> {
21    /// Start the client: bind transport, start network layer, spawn dispatch.
22    pub async fn start(config: ClientConfig, transport: T) -> Result<Self, Error> {
23        Self::start_with_options(config, transport, ClientOptions::default()).await
24    }
25
26    /// Start the client with additional startup options.
27    pub async fn start_with_options(
28        mut config: ClientConfig,
29        transport: T,
30        options: ClientOptions,
31    ) -> Result<Self, Error> {
32        validate_max_apdu_length(config.max_apdu_length)?;
33        validate_max_segments(config.max_segments)?;
34        config.max_apdu_length =
35            cap_max_apdu_to_transport(config.max_apdu_length, transport.max_apdu_length())?;
36        if !(1..=127).contains(&config.proposed_window_size) {
37            return Err(Error::Encoding(format!(
38                "invalid proposed-window-size {}; expected 1..=127",
39                config.proposed_window_size
40            )));
41        }
42        options.validate()?;
43
44        let mut network = NetworkLayer::new(transport);
45        let mut network_control_rx = network.enable_network_control_receiver()?;
46        let mut apdu_rx = network.start().await?;
47        config.max_apdu_length = cap_max_apdu_to_transport(
48            config.max_apdu_length,
49            network.transport().max_apdu_length(),
50        )?;
51        let local_mac = MacAddr::from_slice(network.local_mac());
52
53        let network = Arc::new(network);
54
55        let coordinator = Arc::new(OutboundTransactionCoordinator::new());
56        let tsm = Arc::new(Mutex::new(new_coordinated_tsm(&config, coordinator)));
57        let tsm_dispatch = Arc::clone(&tsm);
58        let device_table = Arc::new(Mutex::new(DeviceTable::new()));
59        let device_table_dispatch = Arc::clone(&device_table);
60        let network_dispatch = Arc::clone(&network);
61        let (cov_tx, _) =
62            broadcast::channel::<ReceivedCOVNotification>(options.cov_channel_capacity);
63        let cov_tx_dispatch = cov_tx.clone();
64        let confirmed_cov_ack_policy = options.confirmed_cov_notification_ack_policy.clone();
65        let (device_tx, _) = broadcast::channel::<DeviceEvent>(DEVICE_EVENT_CHANNEL_CAPACITY);
66        let device_tx_dispatch = device_tx.clone();
67        let (device_collision_tx, _) =
68            broadcast::channel::<DeviceCollisionEvent>(DEVICE_EVENT_CHANNEL_CAPACITY);
69        let device_collision_tx_dispatch = device_collision_tx.clone();
70        let seg_ack_senders: Arc<Mutex<HashMap<SegKey, SegmentAckRoute>>> =
71            Arc::new(Mutex::new(HashMap::new()));
72        let seg_ack_senders_dispatch = Arc::clone(&seg_ack_senders);
73        let (cleanup_tx, mut cleanup_rx) = mpsc::unbounded_channel::<TransactionCleanup>();
74        #[cfg(test)]
75        let segmented_cleanup = Arc::new(SegmentedCleanupHook::default());
76        #[cfg(test)]
77        let segmented_cleanup_dispatch = Arc::clone(&segmented_cleanup);
78        let response_limits = ResponseLimits::from_config(&config);
79        let routed_path_limits = Arc::new(RoutedPathLimits::new(routed_path_quarantine_horizon(
80            &config,
81        )));
82        let routed_path_limits_dispatch = Arc::clone(&routed_path_limits);
83
84        let dispatch_task = tokio::spawn(async move {
85            let mut seg_state: HashMap<SegKey, SegmentedReceiveState> = HashMap::new();
86            let mut device_purge_interval = tokio::time::interval(DEVICE_PURGE_INTERVAL);
87            let mut network_control_open = true;
88            // Tokio intervals tick immediately; consume that tick so discovery purges
89            // run on the configured cadence instead of racing the first APDU.
90            device_purge_interval.tick().await;
91
92            loop {
93                tokio::select! {
94                    control = network_control_rx.recv(), if network_control_open => {
95                        match control {
96                            Some(control) => {
97                                routed_path_limits_dispatch
98                                    .handle_network_control(&tsm_dispatch, control)
99                                    .await;
100                            }
101                            None => network_control_open = false,
102                        }
103                    }
104                    cleanup = cleanup_rx.recv() => {
105                        let Some(cleanup) = cleanup else {
106                            break;
107                        };
108                        if cleanup.cancel_tsm {
109                            tsm_dispatch.lock().await.cancel_transaction_for_owner(
110                                &cleanup.mac,
111                                cleanup.invoke_id,
112                                &cleanup.owner,
113                            );
114                        }
115                        let key = (cleanup.mac, cleanup.invoke_id);
116                        let removed = seg_state
117                            .get(&key)
118                            .is_some_and(|state| state.owner.same_as(&cleanup.owner));
119                        if removed {
120                            seg_state.remove(&key);
121                        }
122                        if let Some(expected_sender) = cleanup.seg_ack_sender {
123                            let mut senders = seg_ack_senders_dispatch.lock().await;
124                            if senders
125                                .get(&key)
126                                .is_some_and(|route| {
127                                    route.owner.same_as(&cleanup.owner)
128                                        && route.sender.same_channel(&expected_sender)
129                                })
130                            {
131                                senders.remove(&key);
132                            }
133                        }
134                        #[cfg(test)]
135                        segmented_cleanup_dispatch.record_processed(removed);
136                    }
137                    _ = device_purge_interval.tick() => {
138                        let lost_devices = device_table_dispatch
139                            .lock()
140                            .await
141                            .purge_stale_collect(DEVICE_MAX_AGE);
142                        for device in lost_devices {
143                            let _ = device_tx_dispatch.send(DeviceEvent {
144                                kind: DeviceEventKind::Lost,
145                                device,
146                            });
147                        }
148                    }
149                    received = apdu_rx.recv() => {
150                        let Some(received) = received else {
151                            break;
152                        };
153                        match apdu::decode_apdu(received.apdu.clone()) {
154                            Ok(decoded) => {
155                                Self::dispatch_apdu(
156                                    &tsm_dispatch,
157                                    &device_table_dispatch,
158                                    &network_dispatch,
159                                    &cov_tx_dispatch,
160                                    &confirmed_cov_ack_policy,
161                                    &device_tx_dispatch,
162                                    &device_collision_tx_dispatch,
163                                    &mut seg_state,
164                                    &seg_ack_senders_dispatch,
165                                    &received.source_mac,
166                                    &received.source_network,
167                                    received.is_group,
168                                    received.reply_tx,
169                                    decoded,
170                                    response_limits,
171                                )
172                                .await;
173                            }
174                            Err(e) => {
175                                warn!(error = %e, "Failed to decode received APDU");
176                            }
177                        }
178                    }
179                }
180            }
181        });
182
183        Ok(Self {
184            config,
185            network,
186            tsm,
187            device_table,
188            cov_tx,
189            device_tx,
190            device_collision_tx,
191            dispatch_task: Some(dispatch_task),
192            seg_ack_senders,
193            cleanup_tx,
194            #[cfg(test)]
195            segmented_post_wait_cleanup: Arc::new(SegmentedPostWaitCleanupHook::default()),
196            #[cfg(test)]
197            segmented_cleanup,
198            local_mac,
199            routed_path_limits,
200        })
201    }
202    /// Get the client's local MAC address.
203    pub fn local_mac(&self) -> &[u8] {
204        &self.local_mac
205    }
206
207    /// Current BACnet max-APDU bucket advertised by this client.
208    ///
209    /// Returns `0` if the live transport budget is below the smallest BACnet
210    /// max-APDU bucket and a confirmed request would currently fail before
211    /// encoding.
212    pub fn max_apdu_length(&self) -> u16 {
213        max_apdu_bucket_at_or_below(
214            self.config
215                .max_apdu_length
216                .min(self.network.transport().max_apdu_length()),
217        )
218        .unwrap_or(0)
219    }
220
221    /// Current APDU payload budget reported by the underlying transport.
222    ///
223    /// For BACnet/SC this reflects the negotiated hub envelope after connect
224    /// and may be lower than the encoded BACnet max-APDU bucket advertised by
225    /// the client.
226    pub fn transport_max_apdu_length(&self) -> u16 {
227        self.network.transport().max_apdu_length()
228    }
229
230    /// Stop the client, aborting the dispatch task.
231    pub async fn stop(&mut self) -> Result<(), Error> {
232        if let Some(task) = self.abort_dispatch_task() {
233            let _ = task.await;
234        }
235        self.tsm.lock().await.cancel_all_transactions();
236        let network = Arc::get_mut(&mut self.network).ok_or_else(|| {
237            Error::Encoding("cannot stop BACnetClient while network references remain".into())
238        })?;
239        network.stop().await?;
240        Ok(())
241    }
242}
243
244impl<T: TransportPort> Drop for BACnetClient<T> {
245    fn drop(&mut self) {
246        let _ = self.abort_dispatch_task();
247        if let Ok(mut tsm) = self.tsm.try_lock() {
248            tsm.cancel_all_transactions();
249        }
250    }
251}
bacnet_client/client/lifecycle.rs

bacnet_client/client/
lifecycle.rs
