rust_ethernet_ip/
subscription.rs

1use crate::error::{EtherNetIpError, Result};
2use crate::PlcValue;
3use std::sync::atomic::AtomicBool;
4use std::sync::Arc;
5use tokio::sync::{mpsc, Mutex};
6
7/// Configuration options for tag subscriptions
8#[derive(Debug, Clone)]
9pub struct SubscriptionOptions {
10    /// Update rate in milliseconds
11    pub update_rate: u32,
12    /// Change threshold for numeric values
13    pub change_threshold: f32,
14    /// Timeout in milliseconds
15    pub timeout: u32,
16}
17
18impl Default for SubscriptionOptions {
19    fn default() -> Self {
20        Self {
21            update_rate: 100,        // 100ms default update rate
22            change_threshold: 0.001, // 0.1% change threshold
23            timeout: 5000,           // 5 second timeout
24        }
25    }
26}
27
28/// Represents a subscription to a PLC tag
29#[derive(Debug, Clone)]
30pub struct TagSubscription {
31    /// The path of the subscribed tag
32    pub tag_path: String,
33    /// Subscription configuration
34    pub options: SubscriptionOptions,
35    /// Last received value
36    pub last_value: Arc<Mutex<Option<PlcValue>>>,
37    /// Channel sender for value updates
38    pub sender: Arc<Mutex<mpsc::Sender<PlcValue>>>,
39    /// Channel receiver for value updates
40    pub receiver: Arc<Mutex<mpsc::Receiver<PlcValue>>>,
41    /// Whether the subscription is active
42    pub is_active: Arc<AtomicBool>,
43}
44
45impl TagSubscription {
46    /// Creates a new tag subscription
47    pub fn new(tag_name: String, options: SubscriptionOptions) -> Self {
48        let (sender, receiver) = mpsc::channel(100); // Buffer size of 100
49        Self {
50            tag_path: tag_name,
51            options,
52            last_value: Arc::new(Mutex::new(None)),
53            sender: Arc::new(Mutex::new(sender)),
54            receiver: Arc::new(Mutex::new(receiver)),
55            is_active: Arc::new(AtomicBool::new(true)),
56        }
57    }
58
59    /// Checks if the subscription is active
60    pub fn is_active(&self) -> bool {
61        self.is_active.load(std::sync::atomic::Ordering::Relaxed)
62    }
63
64    /// Stops the subscription
65    pub fn stop(&self) {
66        self.is_active
67            .store(false, std::sync::atomic::Ordering::Relaxed);
68    }
69
70    /// Updates the subscription value
71    pub async fn update_value(&self, value: &PlcValue) -> Result<()> {
72        let mut last_value = self.last_value.lock().await;
73
74        // Check if value has changed significantly
75        if let (Some(PlcValue::Real(old)), PlcValue::Real(new)) = (last_value.as_ref(), value) {
76            if (*new - *old).abs() < self.options.change_threshold {
77                return Ok(());
78            }
79        }
80
81        // Update value and send notification
82        *last_value = Some(value.clone());
83        let sender = self.sender.lock().await;
84        sender
85            .send(value.clone())
86            .await
87            .map_err(|e| EtherNetIpError::Subscription(format!("Failed to send update: {}", e)))?;
88
89        Ok(())
90    }
91
92    /// Waits for the next value update
93    pub async fn wait_for_update(&self) -> Result<PlcValue> {
94        let mut receiver = self.receiver.lock().await;
95        receiver
96            .recv()
97            .await
98            .ok_or_else(|| EtherNetIpError::Subscription("Channel closed".to_string()))
99    }
100
101    /// Gets the last value received
102    pub async fn get_last_value(&self) -> Option<PlcValue> {
103        self.last_value.lock().await.clone()
104    }
105}
106
107/// Manages multiple tag subscriptions
108#[derive(Debug, Clone)]
109pub struct SubscriptionManager {
110    subscriptions: Arc<Mutex<Vec<TagSubscription>>>,
111}
112
113impl Default for SubscriptionManager {
114    fn default() -> Self {
115        Self::new()
116    }
117}
118
119impl SubscriptionManager {
120    /// Creates a new subscription manager
121    pub fn new() -> Self {
122        Self {
123            subscriptions: Arc::new(Mutex::new(Vec::new())),
124        }
125    }
126
127    /// Adds a new subscription
128    pub async fn add_subscription(&self, subscription: TagSubscription) {
129        let mut subscriptions = self.subscriptions.lock().await;
130        subscriptions.push(subscription);
131    }
132
133    /// Removes a subscription
134    pub async fn remove_subscription(&self, tag_name: &str) {
135        let mut subscriptions = self.subscriptions.lock().await;
136        subscriptions.retain(|sub| sub.tag_path != tag_name);
137    }
138
139    /// Updates a value for all matching subscriptions
140    pub async fn update_value(&self, tag_name: &str, value: &PlcValue) -> Result<()> {
141        let subscriptions = self.subscriptions.lock().await;
142        for subscription in subscriptions.iter() {
143            if subscription.tag_path == tag_name && subscription.is_active() {
144                subscription.update_value(value).await?;
145            }
146        }
147        Ok(())
148    }
149
150    /// Gets all active subscriptions
151    pub async fn get_subscriptions(&self) -> Vec<TagSubscription> {
152        let subscriptions = self.subscriptions.lock().await;
153        subscriptions.clone()
154    }
155
156    /// Gets a specific subscription by tag name
157    pub async fn get_subscription(&self, tag_name: &str) -> Option<TagSubscription> {
158        let subscriptions = self.subscriptions.lock().await;
159        subscriptions
160            .iter()
161            .find(|sub| sub.tag_path == tag_name)
162            .cloned()
163    }
164}
rust_ethernet_ip/subscription.rs

rust_ethernet_ip/
subscription.rs
