rust_ethernet_ip/
fleet.rs

1use crate::client::{Client, ConnectionEvent};
2use crate::error::Result;
3use crate::route::RoutePath;
4use std::collections::HashMap;
5use std::hash::Hash;
6use tokio::sync::broadcast;
7use tokio::task::JoinHandle;
8
9/// Fleet-level connection event annotated with the PLC identifier.
10#[derive(Debug, Clone, PartialEq, Eq)]
11pub struct FleetEvent<PlcId> {
12    /// Application-defined controller identifier.
13    pub plc_id: PlcId,
14    /// Connection lifecycle event from that controller's client.
15    pub event: ConnectionEvent,
16}
17
18/// Multi-PLC pool built from actor-backed [`Client`] handles.
19#[derive(Debug)]
20pub struct Fleet<PlcId> {
21    clients: HashMap<PlcId, Client>,
22    forwarders: HashMap<PlcId, JoinHandle<()>>,
23    events: broadcast::Sender<FleetEvent<PlcId>>,
24}
25
26impl<PlcId> Default for Fleet<PlcId>
27where
28    PlcId: Clone + Eq + Hash + Send + Sync + 'static,
29{
30    fn default() -> Self {
31        Self::new()
32    }
33}
34
35impl<PlcId> Fleet<PlcId>
36where
37    PlcId: Clone + Eq + Hash + Send + Sync + 'static,
38{
39    /// Creates an empty fleet.
40    #[must_use]
41    pub fn new() -> Self {
42        let (events, _) = broadcast::channel(128);
43        Self {
44            clients: HashMap::new(),
45            forwarders: HashMap::new(),
46            events,
47        }
48    }
49
50    /// Connects and adds one PLC by address.
51    pub async fn connect(&mut self, plc_id: PlcId, addr: &str) -> Result<Client> {
52        let client = Client::connect(addr).await?;
53        self.insert_client(plc_id, client.clone());
54        Ok(client)
55    }
56
57    /// Connects and adds one routed PLC by address and route path.
58    pub async fn connect_with_route(
59        &mut self,
60        plc_id: PlcId,
61        addr: &str,
62        route: RoutePath,
63    ) -> Result<Client> {
64        let client = Client::with_route_path(addr, route).await?;
65        self.insert_client(plc_id, client.clone());
66        Ok(client)
67    }
68
69    /// Inserts an existing actor client into the fleet.
70    pub fn insert_client(&mut self, plc_id: PlcId, client: Client) -> Option<Client> {
71        if let Some(forwarder) = self.forwarders.remove(&plc_id) {
72            forwarder.abort();
73        }
74
75        let previous = self.clients.insert(plc_id.clone(), client.clone());
76        let _ = self.events.send(FleetEvent {
77            plc_id: plc_id.clone(),
78            event: ConnectionEvent::Connected,
79        });
80        let forwarder = self.forward_events(plc_id.clone(), client);
81        self.forwarders.insert(plc_id, forwarder);
82        previous
83    }
84
85    /// Returns a cloneable client handle for one PLC.
86    #[must_use]
87    pub fn client(&self, plc_id: &PlcId) -> Option<Client> {
88        self.clients.get(plc_id).cloned()
89    }
90
91    /// Subscribes to fleet-level connection events.
92    pub fn events(&self) -> broadcast::Receiver<FleetEvent<PlcId>> {
93        self.events.subscribe()
94    }
95
96    /// Performs a health check against every PLC currently in the fleet.
97    pub async fn check_health(&self) -> HashMap<PlcId, Result<bool>> {
98        let mut health = HashMap::with_capacity(self.clients.len());
99        for (plc_id, client) in &self.clients {
100            health.insert(plc_id.clone(), client.check_health().await);
101        }
102        health
103    }
104
105    /// Returns the number of PLCs in the fleet.
106    #[must_use]
107    pub fn len(&self) -> usize {
108        self.clients.len()
109    }
110
111    /// Returns true when the fleet has no PLC clients.
112    #[must_use]
113    pub fn is_empty(&self) -> bool {
114        self.clients.is_empty()
115    }
116
117    fn forward_events(&self, plc_id: PlcId, client: Client) -> JoinHandle<()> {
118        let events = self.events.clone();
119        tokio::spawn(async move {
120            forward_events_loop(plc_id, client.events(), events).await;
121        })
122    }
123}
124
125impl<PlcId> Drop for Fleet<PlcId> {
126    fn drop(&mut self) {
127        for (_, forwarder) in self.forwarders.drain() {
128            forwarder.abort();
129        }
130    }
131}
132
133async fn forward_events_loop<PlcId>(
134    plc_id: PlcId,
135    mut client_events: broadcast::Receiver<ConnectionEvent>,
136    events: broadcast::Sender<FleetEvent<PlcId>>,
137) where
138    PlcId: Clone,
139{
140    loop {
141        match client_events.recv().await {
142            Ok(ConnectionEvent::Connected) => continue,
143            Ok(event) => {
144                let _ = events.send(FleetEvent {
145                    plc_id: plc_id.clone(),
146                    event,
147                });
148            }
149            Err(broadcast::error::RecvError::Lagged(_)) => continue,
150            Err(broadcast::error::RecvError::Closed) => break,
151        }
152    }
153}
154
155#[cfg(test)]
156mod tests {
157    use super::*;
158
159    #[tokio::test]
160    async fn forward_events_loop_continues_after_lagged_source() {
161        let (source_tx, source_rx) = broadcast::channel(1);
162        let (fleet_tx, mut fleet_rx) = broadcast::channel(8);
163        let task = tokio::spawn(forward_events_loop("plc-a", source_rx, fleet_tx));
164
165        let _ = source_tx.send(ConnectionEvent::Connected);
166        let _ = source_tx.send(ConnectionEvent::Disconnected);
167        let _ = source_tx.send(ConnectionEvent::WorkerStopped);
168
169        let forwarded = fleet_rx.recv().await.expect("event after lag");
170        assert_eq!(forwarded.plc_id, "plc-a");
171        assert_eq!(forwarded.event, ConnectionEvent::WorkerStopped);
172
173        drop(source_tx);
174        task.await.expect("forwarder exits after source closes");
175    }
176}
rust_ethernet_ip/fleet.rs

rust_ethernet_ip/
fleet.rs
