rust_ethernet_ip/
tag_group.rs

1use crate::PlcValue;
2use crate::subscription::try_send_drop_oldest;
3use std::sync::Arc;
4use std::sync::atomic::{AtomicBool, Ordering};
5use std::time::SystemTime;
6use tokio::sync::{Mutex, mpsc};
7
8/// Defines a named tag group with a polling interval.
9#[derive(Debug, Clone)]
10pub struct TagGroupConfig {
11    pub name: String,
12    pub tags: Vec<String>,
13    pub update_rate_ms: u32,
14}
15
16/// Per-tag result in a group snapshot.
17#[derive(Debug, Clone)]
18pub struct TagGroupValueResult {
19    pub tag_name: String,
20    pub value: Option<PlcValue>,
21    pub error: Option<String>,
22}
23
24/// Snapshot of one polling cycle for a group.
25#[derive(Debug, Clone)]
26pub struct TagGroupSnapshot {
27    pub group_name: String,
28    pub sampled_at: SystemTime,
29    pub values: Vec<TagGroupValueResult>,
30}
31
32/// High-level classification for tag-group polling events.
33#[derive(Debug, Clone, Copy, PartialEq, Eq)]
34#[non_exhaustive]
35pub enum TagGroupEventKind {
36    Data,
37    PartialError,
38    ReadFailure,
39}
40
41/// Structured category for tag-group polling failures.
42#[derive(Debug, Clone, Copy, PartialEq, Eq)]
43#[non_exhaustive]
44pub enum TagGroupFailureCategory {
45    Network,
46    Timeout,
47    PlcStatus,
48    Protocol,
49    Permission,
50    Tag,
51    Data,
52    Other,
53}
54
55/// Structured diagnostics for read failures during tag-group polling.
56#[derive(Debug, Clone, PartialEq, Eq)]
57pub struct TagGroupFailureDiagnostic {
58    pub category: TagGroupFailureCategory,
59    pub retriable: bool,
60    pub status_code: Option<u8>,
61}
62
63impl TagGroupFailureDiagnostic {
64    pub fn from_error(error: &crate::EtherNetIpError) -> Self {
65        use crate::EtherNetIpError;
66
67        let (category, status_code) = match error {
68            EtherNetIpError::Timeout(_) => (TagGroupFailureCategory::Timeout, None),
69            EtherNetIpError::Io(_)
70            | EtherNetIpError::Connection(_)
71            | EtherNetIpError::ConnectionLost(_) => (TagGroupFailureCategory::Network, None),
72            EtherNetIpError::CipError { code, .. } => {
73                (TagGroupFailureCategory::PlcStatus, Some(*code))
74            }
75            EtherNetIpError::ReadError { status, .. }
76            | EtherNetIpError::WriteError { status, .. } => {
77                (TagGroupFailureCategory::PlcStatus, Some(*status))
78            }
79            EtherNetIpError::Permission(_) => (TagGroupFailureCategory::Permission, None),
80            EtherNetIpError::TagNotFound(_) | EtherNetIpError::Tag(_) => {
81                (TagGroupFailureCategory::Tag, None)
82            }
83            EtherNetIpError::DataTypeMismatch { .. }
84            | EtherNetIpError::Udt(_)
85            | EtherNetIpError::StringTooLong { .. }
86            | EtherNetIpError::InvalidString { .. } => (TagGroupFailureCategory::Data, None),
87            EtherNetIpError::Protocol(_)
88            | EtherNetIpError::InvalidResponse { .. }
89            | EtherNetIpError::Subscription(_)
90            | EtherNetIpError::Utf8(_)
91            | EtherNetIpError::Unsupported { .. } => (TagGroupFailureCategory::Protocol, None),
92            EtherNetIpError::Other(_) => (TagGroupFailureCategory::Other, None),
93        };
94
95        Self {
96            category,
97            retriable: error.is_retriable(),
98            status_code,
99        }
100    }
101}
102
103/// Event emitted by background tag-group polling.
104#[derive(Debug, Clone)]
105pub struct TagGroupEvent {
106    pub kind: TagGroupEventKind,
107    pub snapshot: TagGroupSnapshot,
108    pub error: Option<String>,
109    pub failure: Option<TagGroupFailureDiagnostic>,
110}
111
112/// Live subscription to a tag group polling stream.
113#[derive(Debug, Clone)]
114pub struct TagGroupSubscription {
115    pub group_name: String,
116    pub update_rate_ms: u32,
117    is_active: Arc<AtomicBool>,
118    sender: Arc<Mutex<mpsc::Sender<TagGroupEvent>>>,
119    receiver: Arc<Mutex<mpsc::Receiver<TagGroupEvent>>>,
120}
121
122impl TagGroupSubscription {
123    pub fn new(group_name: String, update_rate_ms: u32) -> Self {
124        let (sender, receiver) = mpsc::channel(64);
125        Self {
126            group_name,
127            update_rate_ms,
128            is_active: Arc::new(AtomicBool::new(true)),
129            sender: Arc::new(Mutex::new(sender)),
130            receiver: Arc::new(Mutex::new(receiver)),
131        }
132    }
133
134    pub fn is_active(&self) -> bool {
135        self.is_active.load(Ordering::Relaxed)
136    }
137
138    pub fn stop(&self) {
139        self.is_active.store(false, Ordering::Relaxed);
140    }
141
142    pub async fn publish(&self, snapshot: TagGroupSnapshot) -> Result<(), String> {
143        let event = TagGroupEvent {
144            kind: if snapshot.values.iter().any(|v| v.error.is_some()) {
145                TagGroupEventKind::PartialError
146            } else {
147                TagGroupEventKind::Data
148            },
149            snapshot,
150            error: None,
151            failure: None,
152        };
153        self.publish_event(event).await
154    }
155
156    /// Publishes an event without blocking the polling task.
157    ///
158    /// If the bounded channel is full, the oldest queued event is dropped
159    /// where possible so a slow or abandoned consumer cannot wedge polling.
160    pub async fn publish_event(&self, event: TagGroupEvent) -> Result<(), String> {
161        try_send_drop_oldest(&self.sender, &self.receiver, event).await
162    }
163
164    pub async fn wait_for_update(&self) -> Option<TagGroupEvent> {
165        let mut receiver = self.receiver.lock().await;
166        let next_event = receiver.recv().await;
167        drop(receiver);
168        next_event
169    }
170}
171
172#[cfg(test)]
173mod tests {
174    use super::*;
175    use crate::EtherNetIpError;
176
177    #[test]
178    fn maps_cip_status_failure_diagnostic() {
179        let diagnostic = TagGroupFailureDiagnostic::from_error(&EtherNetIpError::CipError {
180            code: 0x05,
181            message: "Path destination unknown".to_string(),
182        });
183        assert_eq!(diagnostic.category, TagGroupFailureCategory::PlcStatus);
184        assert_eq!(diagnostic.status_code, Some(0x05));
185        assert!(!diagnostic.retriable);
186    }
187
188    #[test]
189    fn maps_timeout_failure_diagnostic_as_retriable() {
190        let diagnostic = TagGroupFailureDiagnostic::from_error(&EtherNetIpError::Timeout(
191            std::time::Duration::from_secs(2),
192        ));
193        assert_eq!(diagnostic.category, TagGroupFailureCategory::Timeout);
194        assert_eq!(diagnostic.status_code, None);
195        assert!(diagnostic.retriable);
196    }
197
198    #[tokio::test]
199    async fn publish_assigns_partial_error_kind() {
200        let sub = TagGroupSubscription::new("group".to_string(), 100);
201        let snapshot = TagGroupSnapshot {
202            group_name: "group".to_string(),
203            sampled_at: SystemTime::now(),
204            values: vec![TagGroupValueResult {
205                tag_name: "Tag1".to_string(),
206                value: None,
207                error: Some("Read failed".to_string()),
208            }],
209        };
210
211        sub.publish(snapshot).await.expect("publish should succeed");
212        let event = sub.wait_for_update().await.expect("event should exist");
213
214        assert_eq!(event.kind, TagGroupEventKind::PartialError);
215        assert!(event.error.is_none());
216        assert!(event.failure.is_none());
217    }
218
219    #[tokio::test]
220    async fn publish_assigns_data_kind_when_all_values_are_ok() {
221        let sub = TagGroupSubscription::new("group".to_string(), 100);
222        let snapshot = TagGroupSnapshot {
223            group_name: "group".to_string(),
224            sampled_at: SystemTime::now(),
225            values: vec![TagGroupValueResult {
226                tag_name: "Tag1".to_string(),
227                value: Some(crate::PlcValue::Dint(42)),
228                error: None,
229            }],
230        };
231
232        sub.publish(snapshot).await.expect("publish should succeed");
233        let event = sub.wait_for_update().await.expect("event should exist");
234
235        assert_eq!(event.kind, TagGroupEventKind::Data);
236        assert!(event.error.is_none());
237        assert!(event.failure.is_none());
238    }
239
240    #[tokio::test]
241    async fn publish_event_preserves_read_failure_diagnostics() {
242        let sub = TagGroupSubscription::new("group".to_string(), 100);
243        let event = TagGroupEvent {
244            kind: TagGroupEventKind::ReadFailure,
245            snapshot: TagGroupSnapshot {
246                group_name: "group".to_string(),
247                sampled_at: SystemTime::now(),
248                values: Vec::new(),
249            },
250            error: Some("timeout while reading tag group".to_string()),
251            failure: Some(TagGroupFailureDiagnostic {
252                category: TagGroupFailureCategory::Timeout,
253                retriable: true,
254                status_code: None,
255            }),
256        };
257
258        sub.publish_event(event.clone())
259            .await
260            .expect("publish_event should succeed");
261        let received = sub.wait_for_update().await.expect("event should exist");
262
263        assert_eq!(received.kind, TagGroupEventKind::ReadFailure);
264        assert_eq!(received.error, event.error);
265        assert_eq!(received.failure, event.failure);
266    }
267
268    #[test]
269    fn stop_marks_subscription_inactive() {
270        let sub = TagGroupSubscription::new("group".to_string(), 100);
271        assert!(sub.is_active());
272        sub.stop();
273        assert!(!sub.is_active());
274    }
275}
rust_ethernet_ip/tag_group.rs

rust_ethernet_ip/
tag_group.rs
