rust_ethernet_ip/
tag_group.rs

1//! Named groups of tags polled as one logical stream.
2
3use crate::PlcValue;
4use crate::subscription::try_send_drop_oldest;
5use std::sync::Arc;
6use std::sync::atomic::{AtomicBool, Ordering};
7use std::time::SystemTime;
8use tokio::sync::{Mutex, mpsc};
9
10/// Defines a named tag group with a polling interval.
11#[derive(Debug, Clone)]
12pub struct TagGroupConfig {
13    /// Stable group name used in snapshots and events.
14    pub name: String,
15    /// Fully qualified symbolic tags to read.
16    pub tags: Vec<String>,
17    /// Polling interval in milliseconds.
18    pub update_rate_ms: u32,
19}
20
21/// Per-tag result in a group snapshot.
22#[derive(Debug, Clone)]
23pub struct TagGroupValueResult {
24    /// Tag associated with this result.
25    pub tag_name: String,
26    /// Decoded value when the read succeeded.
27    pub value: Option<PlcValue>,
28    /// Human-readable error when the read failed.
29    pub error: Option<String>,
30}
31
32/// Snapshot of one polling cycle for a group.
33#[derive(Debug, Clone)]
34pub struct TagGroupSnapshot {
35    /// Group that produced the snapshot.
36    pub group_name: String,
37    /// Time at which the group was sampled.
38    pub sampled_at: SystemTime,
39    /// One result for each configured tag.
40    pub values: Vec<TagGroupValueResult>,
41}
42
43/// High-level classification for tag-group polling events.
44#[derive(Debug, Clone, Copy, PartialEq, Eq)]
45#[non_exhaustive]
46pub enum TagGroupEventKind {
47    /// Every tag read succeeded.
48    Data,
49    /// The polling cycle returned both values and per-tag errors.
50    PartialError,
51    /// The polling cycle failed before per-tag results were available.
52    ReadFailure,
53}
54
55/// Structured category for tag-group polling failures.
56#[derive(Debug, Clone, Copy, PartialEq, Eq)]
57#[non_exhaustive]
58pub enum TagGroupFailureCategory {
59    /// Socket or connection failure.
60    Network,
61    /// Request deadline exceeded.
62    Timeout,
63    /// Controller returned a CIP status code.
64    PlcStatus,
65    /// Malformed or unsupported protocol data.
66    Protocol,
67    /// Access was denied.
68    Permission,
69    /// Tag or symbolic path problem.
70    Tag,
71    /// Value type or encoding problem.
72    Data,
73    /// Failure did not fit another category.
74    Other,
75}
76
77/// Structured diagnostics for read failures during tag-group polling.
78#[derive(Debug, Clone, PartialEq, Eq)]
79pub struct TagGroupFailureDiagnostic {
80    /// Stable failure category.
81    pub category: TagGroupFailureCategory,
82    /// Whether retrying the same request may succeed.
83    pub retriable: bool,
84    /// CIP general status code when supplied by the controller.
85    pub status_code: Option<u8>,
86}
87
88impl TagGroupFailureDiagnostic {
89    /// Converts a library error into wrapper-friendly failure details.
90    pub fn from_error(error: &crate::EtherNetIpError) -> Self {
91        use crate::EtherNetIpError;
92
93        let (category, status_code) = match error {
94            EtherNetIpError::Timeout(_) => (TagGroupFailureCategory::Timeout, None),
95            EtherNetIpError::Io(_)
96            | EtherNetIpError::Connection(_)
97            | EtherNetIpError::ConnectionLost(_) => (TagGroupFailureCategory::Network, None),
98            EtherNetIpError::CipError { code, .. } => {
99                (TagGroupFailureCategory::PlcStatus, Some(*code))
100            }
101            EtherNetIpError::ReadError { status, .. }
102            | EtherNetIpError::WriteError { status, .. } => {
103                (TagGroupFailureCategory::PlcStatus, Some(*status))
104            }
105            EtherNetIpError::Permission(_) => (TagGroupFailureCategory::Permission, None),
106            EtherNetIpError::TagNotFound(_) | EtherNetIpError::Tag(_) => {
107                (TagGroupFailureCategory::Tag, None)
108            }
109            EtherNetIpError::DataTypeMismatch { .. }
110            | EtherNetIpError::Udt(_)
111            | EtherNetIpError::StringTooLong { .. }
112            | EtherNetIpError::InvalidString { .. } => (TagGroupFailureCategory::Data, None),
113            EtherNetIpError::Protocol(_)
114            | EtherNetIpError::InvalidResponse { .. }
115            | EtherNetIpError::Subscription(_)
116            | EtherNetIpError::Utf8(_)
117            | EtherNetIpError::Unsupported { .. } => (TagGroupFailureCategory::Protocol, None),
118            EtherNetIpError::Other(_) => (TagGroupFailureCategory::Other, None),
119        };
120
121        Self {
122            category,
123            retriable: error.is_retriable(),
124            status_code,
125        }
126    }
127}
128
129/// Event emitted by background tag-group polling.
130#[derive(Debug, Clone)]
131pub struct TagGroupEvent {
132    /// Overall result of the polling cycle.
133    pub kind: TagGroupEventKind,
134    /// Snapshot produced by the cycle; may contain per-tag errors.
135    pub snapshot: TagGroupSnapshot,
136    /// Cycle-level error message, when no normal snapshot was possible.
137    pub error: Option<String>,
138    /// Structured cycle-level failure details.
139    pub failure: Option<TagGroupFailureDiagnostic>,
140}
141
142/// Live subscription to a tag group polling stream.
143#[derive(Debug, Clone)]
144pub struct TagGroupSubscription {
145    /// Subscribed group name.
146    pub group_name: String,
147    /// Polling interval in milliseconds.
148    pub update_rate_ms: u32,
149    is_active: Arc<AtomicBool>,
150    sender: Arc<Mutex<mpsc::Sender<TagGroupEvent>>>,
151    receiver: Arc<Mutex<mpsc::Receiver<TagGroupEvent>>>,
152}
153
154impl TagGroupSubscription {
155    /// Creates an active subscription with a bounded event queue.
156    pub fn new(group_name: String, update_rate_ms: u32) -> Self {
157        let (sender, receiver) = mpsc::channel(64);
158        Self {
159            group_name,
160            update_rate_ms,
161            is_active: Arc::new(AtomicBool::new(true)),
162            sender: Arc::new(Mutex::new(sender)),
163            receiver: Arc::new(Mutex::new(receiver)),
164        }
165    }
166
167    /// Returns whether polling should continue.
168    pub fn is_active(&self) -> bool {
169        self.is_active.load(Ordering::Relaxed)
170    }
171
172    /// Marks the subscription inactive.
173    pub fn stop(&self) {
174        self.is_active.store(false, Ordering::Relaxed);
175    }
176
177    /// Classifies and publishes a polling snapshot.
178    pub async fn publish(&self, snapshot: TagGroupSnapshot) -> Result<(), String> {
179        let event = TagGroupEvent {
180            kind: if snapshot.values.iter().any(|v| v.error.is_some()) {
181                TagGroupEventKind::PartialError
182            } else {
183                TagGroupEventKind::Data
184            },
185            snapshot,
186            error: None,
187            failure: None,
188        };
189        self.publish_event(event).await
190    }
191
192    /// Publishes an event without blocking the polling task.
193    ///
194    /// If the bounded channel is full, the oldest queued event is dropped
195    /// where possible so a slow or abandoned consumer cannot wedge polling.
196    pub async fn publish_event(&self, event: TagGroupEvent) -> Result<(), String> {
197        try_send_drop_oldest(&self.sender, &self.receiver, event).await
198    }
199
200    /// Waits for the next queued event, or `None` when all senders close.
201    pub async fn wait_for_update(&self) -> Option<TagGroupEvent> {
202        let mut receiver = self.receiver.lock().await;
203        let next_event = receiver.recv().await;
204        drop(receiver);
205        next_event
206    }
207}
208
209#[cfg(test)]
210mod tests {
211    use super::*;
212    use crate::EtherNetIpError;
213
214    #[test]
215    fn maps_cip_status_failure_diagnostic() {
216        let diagnostic = TagGroupFailureDiagnostic::from_error(&EtherNetIpError::CipError {
217            code: 0x05,
218            message: "Path destination unknown".to_string(),
219        });
220        assert_eq!(diagnostic.category, TagGroupFailureCategory::PlcStatus);
221        assert_eq!(diagnostic.status_code, Some(0x05));
222        assert!(!diagnostic.retriable);
223    }
224
225    #[test]
226    fn maps_timeout_failure_diagnostic_as_retriable() {
227        let diagnostic = TagGroupFailureDiagnostic::from_error(&EtherNetIpError::Timeout(
228            std::time::Duration::from_secs(2),
229        ));
230        assert_eq!(diagnostic.category, TagGroupFailureCategory::Timeout);
231        assert_eq!(diagnostic.status_code, None);
232        assert!(diagnostic.retriable);
233    }
234
235    #[tokio::test]
236    async fn publish_assigns_partial_error_kind() {
237        let sub = TagGroupSubscription::new("group".to_string(), 100);
238        let snapshot = TagGroupSnapshot {
239            group_name: "group".to_string(),
240            sampled_at: SystemTime::now(),
241            values: vec![TagGroupValueResult {
242                tag_name: "Tag1".to_string(),
243                value: None,
244                error: Some("Read failed".to_string()),
245            }],
246        };
247
248        sub.publish(snapshot).await.expect("publish should succeed");
249        let event = sub.wait_for_update().await.expect("event should exist");
250
251        assert_eq!(event.kind, TagGroupEventKind::PartialError);
252        assert!(event.error.is_none());
253        assert!(event.failure.is_none());
254    }
255
256    #[tokio::test]
257    async fn publish_assigns_data_kind_when_all_values_are_ok() {
258        let sub = TagGroupSubscription::new("group".to_string(), 100);
259        let snapshot = TagGroupSnapshot {
260            group_name: "group".to_string(),
261            sampled_at: SystemTime::now(),
262            values: vec![TagGroupValueResult {
263                tag_name: "Tag1".to_string(),
264                value: Some(crate::PlcValue::Dint(42)),
265                error: None,
266            }],
267        };
268
269        sub.publish(snapshot).await.expect("publish should succeed");
270        let event = sub.wait_for_update().await.expect("event should exist");
271
272        assert_eq!(event.kind, TagGroupEventKind::Data);
273        assert!(event.error.is_none());
274        assert!(event.failure.is_none());
275    }
276
277    #[tokio::test]
278    async fn publish_event_preserves_read_failure_diagnostics() {
279        let sub = TagGroupSubscription::new("group".to_string(), 100);
280        let event = TagGroupEvent {
281            kind: TagGroupEventKind::ReadFailure,
282            snapshot: TagGroupSnapshot {
283                group_name: "group".to_string(),
284                sampled_at: SystemTime::now(),
285                values: Vec::new(),
286            },
287            error: Some("timeout while reading tag group".to_string()),
288            failure: Some(TagGroupFailureDiagnostic {
289                category: TagGroupFailureCategory::Timeout,
290                retriable: true,
291                status_code: None,
292            }),
293        };
294
295        sub.publish_event(event.clone())
296            .await
297            .expect("publish_event should succeed");
298        let received = sub.wait_for_update().await.expect("event should exist");
299
300        assert_eq!(received.kind, TagGroupEventKind::ReadFailure);
301        assert_eq!(received.error, event.error);
302        assert_eq!(received.failure, event.failure);
303    }
304
305    #[test]
306    fn stop_marks_subscription_inactive() {
307        let sub = TagGroupSubscription::new("group".to_string(), 100);
308        assert!(sub.is_active());
309        sub.stop();
310        assert!(!sub.is_active());
311    }
312}
rust_ethernet_ip/tag_group.rs

rust_ethernet_ip/
tag_group.rs
