rust_ethernet_ip/client/
actor.rs

1use super::EipClient;
2use crate::batch::{BatchError, BatchOperation, BatchResult};
3use crate::error::{EtherNetIpError, Result};
4use crate::monitoring::DiagnosticsSnapshot;
5use crate::route::RoutePath;
6use crate::types::PlcValue;
7use std::time::Duration;
8use tokio::sync::{broadcast, mpsc, oneshot};
9
10type BatchReadResults = Vec<(String, std::result::Result<PlcValue, BatchError>)>;
11type BatchWriteResults = Vec<(String, std::result::Result<(), BatchError>)>;
12
13#[derive(Debug, Clone, PartialEq, Eq)]
14pub enum ConnectionEvent {
15    Connected,
16    Disconnected,
17    WorkerStopped,
18}
19
20#[derive(Debug, Clone)]
21pub struct Client {
22    tx: mpsc::Sender<ClientCommand>,
23    events: broadcast::Sender<ConnectionEvent>,
24}
25
26#[derive(Debug, Clone)]
27pub enum Backoff {
28    Constant(Duration),
29    Exponential {
30        initial: Duration,
31        max: Duration,
32        factor: u32,
33    },
34}
35
36#[derive(Debug, Clone)]
37pub struct RetryPolicy {
38    pub max_attempts: usize,
39    pub backoff: Backoff,
40    pub retry_writes: bool,
41}
42
43impl RetryPolicy {
44    pub fn constant(max_attempts: usize, delay: Duration) -> Self {
45        Self {
46            max_attempts: max_attempts.max(1),
47            backoff: Backoff::Constant(delay),
48            retry_writes: false,
49        }
50    }
51
52    pub fn exponential(max_attempts: usize, initial: Duration, max: Duration) -> Self {
53        Self {
54            max_attempts: max_attempts.max(1),
55            backoff: Backoff::Exponential {
56                initial,
57                max,
58                factor: 2,
59            },
60            retry_writes: false,
61        }
62    }
63
64    pub fn retry_writes(mut self, retry_writes: bool) -> Self {
65        self.retry_writes = retry_writes;
66        self
67    }
68
69    fn delay_for_attempt(&self, attempt_index: usize) -> Duration {
70        match self.backoff {
71            Backoff::Constant(delay) => delay,
72            Backoff::Exponential {
73                initial,
74                max,
75                factor,
76            } => {
77                let multiplier = factor.saturating_pow(attempt_index as u32);
78                initial.saturating_mul(multiplier).min(max)
79            }
80        }
81    }
82}
83
84#[derive(Clone)]
85pub struct RetryClient {
86    client: Client,
87    policy: RetryPolicy,
88}
89
90enum ClientCommand {
91    ReadTag {
92        tag_name: String,
93        reply: oneshot::Sender<Result<PlcValue>>,
94    },
95    WriteTag {
96        tag_name: String,
97        value: PlcValue,
98        reply: oneshot::Sender<Result<()>>,
99    },
100    WriteStringTag {
101        tag_name: String,
102        value: String,
103        reply: oneshot::Sender<Result<()>>,
104    },
105    WriteUdtMember {
106        tag_name: String,
107        member_name: String,
108        value: PlcValue,
109        reply: oneshot::Sender<Result<()>>,
110    },
111    ExecuteBatch {
112        operations: Vec<BatchOperation>,
113        reply: oneshot::Sender<Result<Vec<BatchResult>>>,
114    },
115    ReadTagsBatch {
116        tag_names: Vec<String>,
117        reply: oneshot::Sender<Result<BatchReadResults>>,
118    },
119    WriteTagsBatch {
120        tag_values: Vec<(String, PlcValue)>,
121        reply: oneshot::Sender<Result<BatchWriteResults>>,
122    },
123    CheckHealth {
124        reply: oneshot::Sender<bool>,
125    },
126    Diagnostics {
127        verified: bool,
128        reply: oneshot::Sender<Result<DiagnosticsSnapshot>>,
129    },
130}
131
132impl Client {
133    pub async fn connect(addr: &str) -> Result<Self> {
134        Self::from_eip_client(EipClient::connect(addr).await?)
135    }
136
137    pub async fn with_route_path(addr: &str, route: RoutePath) -> Result<Self> {
138        Self::from_eip_client(EipClient::with_route_path(addr, route).await?)
139    }
140
141    pub fn from_eip_client(client: EipClient) -> Result<Self> {
142        let (tx, rx) = mpsc::channel(128);
143        let (events, _) = broadcast::channel(128);
144        let actor = Self {
145            tx,
146            events: events.clone(),
147        };
148
149        tokio::spawn(run_client_actor(client, rx, events));
150        Ok(actor)
151    }
152
153    pub fn events(&self) -> broadcast::Receiver<ConnectionEvent> {
154        self.events.subscribe()
155    }
156
157    pub fn with_retry(&self, policy: RetryPolicy) -> RetryClient {
158        RetryClient {
159            client: self.clone(),
160            policy,
161        }
162    }
163
164    pub async fn read_tag(&self, tag_name: &str) -> Result<PlcValue> {
165        let (reply, rx) = oneshot::channel();
166        self.send(ClientCommand::ReadTag {
167            tag_name: tag_name.to_string(),
168            reply,
169        })
170        .await?;
171        rx.await.unwrap_or_else(|_| actor_stopped())
172    }
173
174    pub async fn write_tag(&self, tag_name: &str, value: PlcValue) -> Result<()> {
175        let (reply, rx) = oneshot::channel();
176        self.send(ClientCommand::WriteTag {
177            tag_name: tag_name.to_string(),
178            value,
179            reply,
180        })
181        .await?;
182        rx.await.unwrap_or_else(|_| actor_stopped())
183    }
184
185    pub async fn write_string_tag(&self, tag_name: &str, value: &str) -> Result<()> {
186        let (reply, rx) = oneshot::channel();
187        self.send(ClientCommand::WriteStringTag {
188            tag_name: tag_name.to_string(),
189            value: value.to_string(),
190            reply,
191        })
192        .await?;
193        rx.await.unwrap_or_else(|_| actor_stopped())
194    }
195
196    pub async fn write_udt_member(
197        &self,
198        udt_tag_name: &str,
199        member_name: &str,
200        value: PlcValue,
201    ) -> Result<()> {
202        let (reply, rx) = oneshot::channel();
203        self.send(ClientCommand::WriteUdtMember {
204            tag_name: udt_tag_name.to_string(),
205            member_name: member_name.to_string(),
206            value,
207            reply,
208        })
209        .await?;
210        rx.await.unwrap_or_else(|_| actor_stopped())
211    }
212
213    pub async fn write_udt_array_member(
214        &self,
215        udt_array_element_path: &str,
216        member_name: &str,
217        value: PlcValue,
218    ) -> Result<()> {
219        self.write_udt_member(udt_array_element_path, member_name, value)
220            .await
221    }
222
223    pub async fn execute_batch(&self, operations: &[BatchOperation]) -> Result<Vec<BatchResult>> {
224        let (reply, rx) = oneshot::channel();
225        self.send(ClientCommand::ExecuteBatch {
226            operations: operations.to_vec(),
227            reply,
228        })
229        .await?;
230        rx.await.unwrap_or_else(|_| actor_stopped())
231    }
232
233    pub async fn read_tags_batch(&self, tag_names: &[&str]) -> Result<BatchReadResults> {
234        let (reply, rx) = oneshot::channel();
235        self.send(ClientCommand::ReadTagsBatch {
236            tag_names: tag_names.iter().map(|name| (*name).to_string()).collect(),
237            reply,
238        })
239        .await?;
240        rx.await.unwrap_or_else(|_| actor_stopped())
241    }
242
243    pub async fn write_tags_batch(
244        &self,
245        tag_values: &[(&str, PlcValue)],
246    ) -> Result<BatchWriteResults> {
247        let (reply, rx) = oneshot::channel();
248        self.send(ClientCommand::WriteTagsBatch {
249            tag_values: tag_values
250                .iter()
251                .map(|(name, value)| ((*name).to_string(), value.clone()))
252                .collect(),
253            reply,
254        })
255        .await?;
256        rx.await.unwrap_or_else(|_| actor_stopped())
257    }
258
259    pub async fn check_health(&self) -> Result<bool> {
260        let (reply, rx) = oneshot::channel();
261        self.send(ClientCommand::CheckHealth { reply }).await?;
262        rx.await.map_err(|_| actor_stopped_error())
263    }
264
265    pub async fn get_diagnostics_snapshot(&self) -> Result<DiagnosticsSnapshot> {
266        self.diagnostics(false).await
267    }
268
269    pub async fn get_diagnostics_snapshot_detailed(&self) -> Result<DiagnosticsSnapshot> {
270        self.diagnostics(true).await
271    }
272
273    async fn diagnostics(&self, verified: bool) -> Result<DiagnosticsSnapshot> {
274        let (reply, rx) = oneshot::channel();
275        self.send(ClientCommand::Diagnostics { verified, reply })
276            .await?;
277        rx.await.unwrap_or_else(|_| actor_stopped())
278    }
279
280    async fn send(&self, command: ClientCommand) -> Result<()> {
281        self.tx
282            .send(command)
283            .await
284            .map_err(|_| actor_stopped_error())
285    }
286}
287
288impl RetryClient {
289    pub async fn read_tag(&self, tag_name: &str) -> Result<PlcValue> {
290        self.retry(|| async { self.client.read_tag(tag_name).await })
291            .await
292    }
293
294    pub async fn write_tag(&self, tag_name: &str, value: PlcValue) -> Result<()> {
295        if !self.policy.retry_writes {
296            return self.client.write_tag(tag_name, value).await;
297        }
298
299        self.retry(|| {
300            let value = value.clone();
301            async move { self.client.write_tag(tag_name, value).await }
302        })
303        .await
304    }
305
306    pub async fn write_string_tag(&self, tag_name: &str, value: &str) -> Result<()> {
307        if !self.policy.retry_writes {
308            return self.client.write_string_tag(tag_name, value).await;
309        }
310
311        self.retry(|| async { self.client.write_string_tag(tag_name, value).await })
312            .await
313    }
314
315    async fn retry<T, Fut, Op>(&self, mut op: Op) -> Result<T>
316    where
317        Fut: std::future::Future<Output = Result<T>>,
318        Op: FnMut() -> Fut,
319    {
320        let mut attempt = 0;
321        loop {
322            match op().await {
323                Ok(value) => return Ok(value),
324                Err(err) if err.is_retriable() && attempt + 1 < self.policy.max_attempts => {
325                    let delay = self.policy.delay_for_attempt(attempt);
326                    attempt += 1;
327                    tokio::time::sleep(delay).await;
328                }
329                Err(err) => return Err(err),
330            }
331        }
332    }
333}
334
335async fn run_client_actor(
336    mut client: EipClient,
337    mut rx: mpsc::Receiver<ClientCommand>,
338    events: broadcast::Sender<ConnectionEvent>,
339) {
340    let _ = events.send(ConnectionEvent::Connected);
341
342    while let Some(command) = rx.recv().await {
343        match command {
344            ClientCommand::ReadTag { tag_name, reply } => {
345                let _ = reply.send(client.read_tag(&tag_name).await);
346            }
347            ClientCommand::WriteTag {
348                tag_name,
349                value,
350                reply,
351            } => {
352                let _ = reply.send(client.write_tag(&tag_name, value).await);
353            }
354            ClientCommand::WriteStringTag {
355                tag_name,
356                value,
357                reply,
358            } => {
359                let _ = reply.send(client.write_string_tag(&tag_name, &value).await);
360            }
361            ClientCommand::WriteUdtMember {
362                tag_name,
363                member_name,
364                value,
365                reply,
366            } => {
367                let _ = reply.send(
368                    client
369                        .write_udt_member(&tag_name, &member_name, value)
370                        .await,
371                );
372            }
373            ClientCommand::ExecuteBatch { operations, reply } => {
374                let _ = reply.send(client.execute_batch(&operations).await);
375            }
376            ClientCommand::ReadTagsBatch { tag_names, reply } => {
377                let refs: Vec<&str> = tag_names.iter().map(String::as_str).collect();
378                let _ = reply.send(client.read_tags_batch(&refs).await);
379            }
380            ClientCommand::WriteTagsBatch { tag_values, reply } => {
381                let refs: Vec<(&str, PlcValue)> = tag_values
382                    .iter()
383                    .map(|(name, value)| (name.as_str(), value.clone()))
384                    .collect();
385                let _ = reply.send(client.write_tags_batch(&refs).await);
386            }
387            ClientCommand::CheckHealth { reply } => {
388                let _ = reply.send(client.check_health().await);
389            }
390            ClientCommand::Diagnostics { verified, reply } => {
391                let result = if verified {
392                    client.get_diagnostics_snapshot_detailed().await
393                } else {
394                    Ok(client.get_diagnostics_snapshot().await)
395                };
396                let _ = reply.send(result);
397            }
398        }
399    }
400
401    let _ = client.unregister_session().await;
402    let _ = events.send(ConnectionEvent::Disconnected);
403    let _ = events.send(ConnectionEvent::WorkerStopped);
404}
405
406fn actor_stopped<T>() -> Result<T> {
407    Err(actor_stopped_error())
408}
409
410fn actor_stopped_error() -> EtherNetIpError {
411    EtherNetIpError::ConnectionLost("client actor stopped".to_string())
412}
rust_ethernet_ip/client/actor.rs

rust_ethernet_ip/client/
actor.rs
