rust_ethernet_ip/client/
subscriptions.rs

1use super::EipClient;
2use crate::error::{EtherNetIpError, Result};
3use crate::subscription::{SubscriptionOptions, TagSubscription};
4use crate::tag_group::{
5    TagGroupConfig, TagGroupEvent, TagGroupEventKind, TagGroupFailureDiagnostic, TagGroupSnapshot,
6    TagGroupSubscription, TagGroupValueResult,
7};
8use crate::types::PlcValue;
9
10const CONNECTION_LOST_FAILURE_LIMIT: usize = 3;
11const RETRIABLE_ERROR_BACKOFF_MS: u64 = 250;
12
13impl EipClient {
14    /// Subscribes to a tag for real-time updates.
15    ///
16    /// The returned [`TagSubscription`] can be used to:
17    /// - [`wait_for_update()`](TagSubscription::wait_for_update) for the next value
18    /// - [`get_last_value()`](TagSubscription::get_last_value) for the latest cached value
19    /// - [`into_stream()`](TagSubscription::into_stream) for an async `Stream` of updates
20    ///
21    /// This API validates the tag with an initial read before returning so invalid or
22    /// inaccessible tags fail fast instead of surfacing only through background polling logs.
23    ///
24    /// The background poll loop uses [`SubscriptionOptions::update_rate`] (milliseconds) between reads.
25    pub async fn subscribe_to_tag(
26        &self,
27        tag_path: &str,
28        options: SubscriptionOptions,
29    ) -> Result<TagSubscription> {
30        let subscription = TagSubscription::new(tag_path.to_string(), options.clone());
31        let mut validation_client = self.clone();
32        let initial_value = validation_client.read_tag(tag_path).await?;
33        subscription.update_value(&initial_value).await?;
34
35        let mut subscriptions = self.subscriptions.lock().await;
36        let update_rate_ms = options.update_rate;
37        subscriptions.push(subscription.clone());
38        drop(subscriptions);
39
40        let tag_path = tag_path.to_string();
41        let mut client = self.clone();
42        let subscription_task = subscription.clone();
43        tokio::spawn(async move {
44            let interval = std::time::Duration::from_millis(update_rate_ms as u64);
45            let retry_backoff = std::time::Duration::from_millis(RETRIABLE_ERROR_BACKOFF_MS);
46            let mut consecutive_connection_lost = 0usize;
47            while subscription_task.is_active() {
48                match client.read_tag(&tag_path).await {
49                    Ok(value) => {
50                        consecutive_connection_lost = 0;
51                        if let Err(e) = client.update_subscription(&tag_path, &value).await {
52                            let _ = subscription_task.publish_error(&e, true).await;
53                            tracing::error!("Error updating subscription: {}", e);
54                            break;
55                        }
56                    }
57                    Err(e) => {
58                        tracing::error!("Error reading tag {}: {}", tag_path, e);
59                        let is_connection_lost = matches!(e, EtherNetIpError::ConnectionLost(_));
60                        if is_connection_lost {
61                            consecutive_connection_lost += 1;
62                        } else {
63                            consecutive_connection_lost = 0;
64                        }
65
66                        let terminal = !e.is_retriable()
67                            || consecutive_connection_lost >= CONNECTION_LOST_FAILURE_LIMIT;
68                        let _ = subscription_task.publish_error(&e, terminal).await;
69                        if terminal {
70                            break;
71                        }
72
73                        tokio::time::sleep(interval.max(retry_backoff)).await;
74                        continue;
75                    }
76                }
77                if subscription_task.is_active() {
78                    tokio::time::sleep(interval).await;
79                }
80            }
81        });
82        Ok(subscription)
83    }
84
85    /// Stops and removes subscriptions for a tag path.
86    ///
87    /// Returns true when at least one subscription was found. Polling stops at
88    /// the next loop check, normally within one configured update interval.
89    pub async fn unsubscribe(&self, tag_path: &str) -> bool {
90        let mut subscriptions = self.subscriptions.lock().await;
91        let mut found = false;
92        for subscription in subscriptions.iter() {
93            if subscription.tag_path == tag_path {
94                subscription.stop();
95                found = true;
96            }
97        }
98        subscriptions.retain(TagSubscription::is_active);
99        found
100    }
101
102    #[doc(hidden)]
103    pub async fn subscription_count(&self) -> usize {
104        let mut subscriptions = self.subscriptions.lock().await;
105        subscriptions.retain(TagSubscription::is_active);
106        subscriptions.len()
107    }
108
109    /// Subscribes to multiple tags. Returns one [`TagSubscription`] per tag in order.
110    pub async fn subscribe_to_tags(
111        &self,
112        tags: &[(&str, SubscriptionOptions)],
113    ) -> Result<Vec<TagSubscription>> {
114        let mut subs = Vec::with_capacity(tags.len());
115        for (tag_name, options) in tags {
116            subs.push(self.subscribe_to_tag(tag_name, options.clone()).await?);
117        }
118        Ok(subs)
119    }
120
121    /// Registers or replaces a named tag group for grouped polling.
122    ///
123    /// Tag groups are useful for HMI/SCADA-style scan classes where multiple tags
124    /// share a polling interval and should be read together.
125    pub async fn upsert_tag_group(
126        &self,
127        group_name: &str,
128        tags: &[&str],
129        update_rate_ms: u32,
130    ) -> Result<()> {
131        if group_name.trim().is_empty() {
132            return Err(EtherNetIpError::Protocol(
133                "Tag group name cannot be empty".to_string(),
134            ));
135        }
136        if tags.is_empty() {
137            return Err(EtherNetIpError::Protocol(
138                "Tag group must contain at least one tag".to_string(),
139            ));
140        }
141        if update_rate_ms == 0 {
142            return Err(EtherNetIpError::Protocol(
143                "Tag group update rate must be greater than 0 ms".to_string(),
144            ));
145        }
146
147        let config = TagGroupConfig {
148            name: group_name.to_string(),
149            tags: tags.iter().map(|s| (*s).to_string()).collect(),
150            update_rate_ms,
151        };
152
153        let mut groups = self.tag_groups.lock().await;
154        groups.insert(group_name.to_string(), config);
155        Ok(())
156    }
157
158    /// Removes a named tag group.
159    pub async fn remove_tag_group(&self, group_name: &str) -> bool {
160        let mut groups = self.tag_groups.lock().await;
161        groups.remove(group_name).is_some()
162    }
163
164    /// Lists all currently registered tag groups.
165    pub async fn list_tag_groups(&self) -> Vec<TagGroupConfig> {
166        let groups = self.tag_groups.lock().await;
167        groups.values().cloned().collect()
168    }
169
170    /// Reads all tags in a group once and returns a snapshot.
171    pub async fn read_tag_group_once(&self, group_name: &str) -> Result<TagGroupSnapshot> {
172        let config = {
173            let groups = self.tag_groups.lock().await;
174            groups.get(group_name).cloned().ok_or_else(|| {
175                EtherNetIpError::Protocol(format!("Tag group '{}' is not registered", group_name))
176            })?
177        };
178
179        let mut client = self.clone();
180        let tag_refs: Vec<&str> = config.tags.iter().map(String::as_str).collect();
181        let values = client.read_tags_batch(&tag_refs).await?;
182
183        let mapped = values
184            .into_iter()
185            .map(|(tag_name, result)| match result {
186                Ok(value) => TagGroupValueResult {
187                    tag_name,
188                    value: Some(value),
189                    error: None,
190                },
191                Err(e) => TagGroupValueResult {
192                    tag_name,
193                    value: None,
194                    error: Some(e.to_string()),
195                },
196            })
197            .collect();
198
199        Ok(TagGroupSnapshot {
200            group_name: config.name,
201            sampled_at: std::time::SystemTime::now(),
202            values: mapped,
203        })
204    }
205
206    /// Starts background polling for a registered tag group.
207    ///
208    /// Use the returned subscription to await snapshots and to stop polling.
209    pub async fn subscribe_tag_group(&self, group_name: &str) -> Result<TagGroupSubscription> {
210        let config = {
211            let groups = self.tag_groups.lock().await;
212            groups.get(group_name).cloned().ok_or_else(|| {
213                EtherNetIpError::Protocol(format!("Tag group '{}' is not registered", group_name))
214            })?
215        };
216
217        let subscription = TagGroupSubscription::new(config.name.clone(), config.update_rate_ms);
218        let subscription_task = subscription.clone();
219        let mut client = self.clone();
220        let tags = config.tags.clone();
221        let interval = std::time::Duration::from_millis(config.update_rate_ms as u64);
222        let group_name_owned = config.name.clone();
223
224        tokio::spawn(async move {
225            while subscription_task.is_active() {
226                let tag_refs: Vec<&str> = tags.iter().map(String::as_str).collect();
227                match client.read_tags_batch(&tag_refs).await {
228                    Ok(values) => {
229                        let has_errors = values.iter().any(|(_, result)| result.is_err());
230                        let snapshot = TagGroupSnapshot {
231                            group_name: group_name_owned.clone(),
232                            sampled_at: std::time::SystemTime::now(),
233                            values: values
234                                .into_iter()
235                                .map(|(tag_name, result)| match result {
236                                    Ok(value) => TagGroupValueResult {
237                                        tag_name,
238                                        value: Some(value),
239                                        error: None,
240                                    },
241                                    Err(e) => TagGroupValueResult {
242                                        tag_name,
243                                        value: None,
244                                        error: Some(e.to_string()),
245                                    },
246                                })
247                                .collect(),
248                        };
249
250                        let event = TagGroupEvent {
251                            kind: if has_errors {
252                                TagGroupEventKind::PartialError
253                            } else {
254                                TagGroupEventKind::Data
255                            },
256                            snapshot,
257                            error: None,
258                            failure: None,
259                        };
260
261                        if let Err(e) = subscription_task.publish_event(event).await {
262                            tracing::error!(
263                                "Tag group '{}' publish failed: {}",
264                                group_name_owned,
265                                e
266                            );
267                            break;
268                        }
269                    }
270                    Err(e) => {
271                        tracing::error!(
272                            "Tag group '{}' polling read failed: {}",
273                            group_name_owned,
274                            e
275                        );
276                        let failure_event = TagGroupEvent {
277                            kind: TagGroupEventKind::ReadFailure,
278                            snapshot: TagGroupSnapshot {
279                                group_name: group_name_owned.clone(),
280                                sampled_at: std::time::SystemTime::now(),
281                                values: Vec::new(),
282                            },
283                            error: Some(e.to_string()),
284                            failure: Some(TagGroupFailureDiagnostic::from_error(&e)),
285                        };
286                        if let Err(publish_error) =
287                            subscription_task.publish_event(failure_event).await
288                        {
289                            tracing::error!(
290                                "Tag group '{}' failure-event publish failed: {}",
291                                group_name_owned,
292                                publish_error
293                            );
294                            break;
295                        }
296                    }
297                }
298                tokio::time::sleep(interval).await;
299            }
300        });
301
302        Ok(subscription)
303    }
304
305    async fn update_subscription(&self, tag_name: &str, value: &PlcValue) -> Result<()> {
306        let subscriptions = {
307            let mut subscriptions = self.subscriptions.lock().await;
308            subscriptions.retain(TagSubscription::is_active);
309            subscriptions
310                .iter()
311                .filter(|subscription| subscription.tag_path == tag_name)
312                .cloned()
313                .collect::<Vec<_>>()
314        };
315        for subscription in &subscriptions {
316            if subscription.is_active() {
317                subscription.update_value(value).await?;
318            }
319        }
320        Ok(())
321    }
322}
rust_ethernet_ip/client/subscriptions.rs

rust_ethernet_ip/client/
subscriptions.rs
