v0.1.5 (#40)

Reviewed-on: wrenn/wrenn#40

envd-rs/src/port/scanner.rs

@@ -0,0 +1,79 @@
+use std::sync::{Arc, RwLock};
+use std::time::Duration;
+
+use tokio::sync::mpsc;
+use tokio_util::sync::CancellationToken;
+
+use super::conn::{ConnStat, read_tcp_connections};
+
+pub struct ScannerFilter {
+    pub ips: Vec<String>,
+    pub state: String,
+}
+
+impl ScannerFilter {
+    pub fn matches(&self, conn: &ConnStat) -> bool {
+        if self.state.is_empty() && self.ips.is_empty() {
+            return false;
+        }
+        self.ips.contains(&conn.local_ip) && self.state == conn.status
+    }
+}
+
+pub struct ScannerSubscriber {
+    pub tx: mpsc::Sender<Vec<ConnStat>>,
+    pub filter: Option<ScannerFilter>,
+}
+
+pub struct Scanner {
+    period: Duration,
+    subs: RwLock<Vec<(String, Arc<ScannerSubscriber>)>>,
+}
+
+impl Scanner {
+    pub fn new(period: Duration) -> Self {
+        Self {
+            period,
+            subs: RwLock::new(Vec::new()),
+        }
+    }
+
+    pub fn add_subscriber(
+        &self,
+        id: &str,
+        filter: Option<ScannerFilter>,
+    ) -> mpsc::Receiver<Vec<ConnStat>> {
+        let (tx, rx) = mpsc::channel(4);
+        let sub = Arc::new(ScannerSubscriber { tx, filter });
+        let mut subs = self.subs.write().unwrap();
+        subs.push((id.to_string(), sub));
+        rx
+    }
+
+    pub fn remove_subscriber(&self, id: &str) {
+        let mut subs = self.subs.write().unwrap();
+        subs.retain(|(sid, _)| sid != id);
+    }
+
+    pub async fn scan_and_broadcast(&self, cancel: CancellationToken) {
+        loop {
+            let conns = read_tcp_connections();
+
+            {
+                let subs = self.subs.read().unwrap();
+                for (_, sub) in subs.iter() {
+                    let payload = match &sub.filter {
+                        Some(f) => conns.iter().filter(|c| f.matches(c)).cloned().collect(),
+                        None => conns.clone(),
+                    };
+                    let _ = sub.tx.try_send(payload);
+                }
+            }
+
+            tokio::select! {
+                _ = cancel.cancelled() => return,
+                _ = tokio::time::sleep(self.period) => {}
+            }
+        }
+    }
+}