implemented listeners for transfered event [CI SKIP]

Merge branch 'timeout' into new_events
added transfered event [CI SKIP]
2019-04-04 21:18:21 +02:00 · 2019-04-04 20:39:00 +02:00 · 2019-04-04 20:34:54 +02:00 · 2019-04-04 18:59:17 +02:00 · 2019-04-03 17:54:53 +02:00
6 changed files with 99 additions and 130 deletions
--- a/6
+++ b/6
@@ -2,12 +2,6 @@ REPO := repo.shimun.net/shimun/wireguard-user

 build:
 	docker build . -t ${REPO}
-
-.ONESHELL:	
-update:
-	cd boringtun
-	git checkout master
-	git pull	
 	
 push: build
 	docker push ${REPO}     
--- a/handler.sh
+++ b/handler.sh
@@ -24,10 +24,13 @@ case "$EVENT" in
        PREV_ENDPOINT=$8
 ;;
 "added")
+
 ;;
 "removed")
    
 ;;
+"transfered")
+
 *)
 exit 1
 ;;
--- a/wg-event-gen/Cargo.lock
+++ b/wg-event-gen/Cargo.lock
@@ -133,12 +133,9 @@ version = "0.2.47"
 source = "registry+https://github.com/rust-lang/crates.io-index"

 [[package]]
-name = "num_cpus"
-version = "1.10.0"
+name = "number_prefix"
+version = "0.3.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-dependencies = [
- "libc 0.2.47 (registry+https://github.com/rust-lang/crates.io-index)",
-]

 [[package]]
 name = "proc-macro2"
@@ -222,14 +219,6 @@ dependencies = [
 "unicode-width 0.1.5 (registry+https://github.com/rust-lang/crates.io-index)",
 ]

-[[package]]
-name = "threadpool"
-version = "1.7.1"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-dependencies = [
- "num_cpus 1.10.0 (registry+https://github.com/rust-lang/crates.io-index)",
-]
-
 [[package]]
 name = "time"
 version = "0.1.42"
@@ -267,9 +256,9 @@ dependencies = [
 "base64 0.10.0 (registry+https://github.com/rust-lang/crates.io-index)",
 "derive_builder 0.7.1 (registry+https://github.com/rust-lang/crates.io-index)",
 "hex 0.3.2 (registry+https://github.com/rust-lang/crates.io-index)",
+ "number_prefix 0.3.0 (registry+https://github.com/rust-lang/crates.io-index)",
 "structopt 0.2.14 (registry+https://github.com/rust-lang/crates.io-index)",
 "structopt-derive 0.2.14 (registry+https://github.com/rust-lang/crates.io-index)",
- "threadpool 1.7.1 (registry+https://github.com/rust-lang/crates.io-index)",
 "time 0.1.42 (registry+https://github.com/rust-lang/crates.io-index)",
 ]

@@ -309,7 +298,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 "checksum hex 0.3.2 (registry+https://github.com/rust-lang/crates.io-index)" = "805026a5d0141ffc30abb3be3173848ad46a1b1664fe632428479619a3644d77"
 "checksum ident_case 1.0.1 (registry+https://github.com/rust-lang/crates.io-index)" = "b9e0384b61958566e926dc50660321d12159025e767c18e043daf26b70104c39"
 "checksum libc 0.2.47 (registry+https://github.com/rust-lang/crates.io-index)" = "48450664a984b25d5b479554c29cc04e3150c97aa4c01da5604a2d4ed9151476"
-"checksum num_cpus 1.10.0 (registry+https://github.com/rust-lang/crates.io-index)" = "1a23f0ed30a54abaa0c7e83b1d2d87ada7c3c23078d1d87815af3e3b6385fbba"
+"checksum number_prefix 0.3.0 (registry+https://github.com/rust-lang/crates.io-index)" = "17b02fc0ff9a9e4b35b3342880f48e896ebf69f2967921fe8646bf5b7125956a"
 "checksum proc-macro2 0.4.27 (registry+https://github.com/rust-lang/crates.io-index)" = "4d317f9caece796be1980837fd5cb3dfec5613ebdb04ad0956deea83ce168915"
 "checksum quote 0.6.10 (registry+https://github.com/rust-lang/crates.io-index)" = "53fa22a1994bd0f9372d7a816207d8a2677ad0325b073f5c5332760f0fb62b5c"
 "checksum redox_syscall 0.1.50 (registry+https://github.com/rust-lang/crates.io-index)" = "52ee9a534dc1301776eff45b4fa92d2c39b1d8c3d3357e6eb593e0d795506fc2"
@@ -320,7 +309,6 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 "checksum syn 0.15.26 (registry+https://github.com/rust-lang/crates.io-index)" = "f92e629aa1d9c827b2bb8297046c1ccffc57c99b947a680d3ccff1f136a3bee9"
 "checksum termion 1.5.1 (registry+https://github.com/rust-lang/crates.io-index)" = "689a3bdfaab439fd92bc87df5c4c78417d3cbe537487274e9b0b2dce76e92096"
 "checksum textwrap 0.10.0 (registry+https://github.com/rust-lang/crates.io-index)" = "307686869c93e71f94da64286f9a9524c0f308a9e1c87a583de8e9c9039ad3f6"
-"checksum threadpool 1.7.1 (registry+https://github.com/rust-lang/crates.io-index)" = "e2f0c90a5f3459330ac8bc0d2f879c693bb7a2f59689c1083fc4ef83834da865"
 "checksum time 0.1.42 (registry+https://github.com/rust-lang/crates.io-index)" = "db8dcfca086c1143c9270ac42a2bbd8a7ee477b78ac8e45b19abfb0cbede4b6f"
 "checksum unicode-segmentation 1.2.1 (registry+https://github.com/rust-lang/crates.io-index)" = "aa6024fc12ddfd1c6dbc14a80fa2324d4568849869b779f6bd37e5e4c03344d1"
 "checksum unicode-width 0.1.5 (registry+https://github.com/rust-lang/crates.io-index)" = "882386231c45df4700b275c7ff55b6f3698780a650026380e72dabe76fa46526"
--- a/wg-event-gen/Cargo.toml
+++ b/wg-event-gen/Cargo.toml
@@ -11,7 +11,7 @@ time = "0.1.42"
 structopt = "0.2.14"
 structopt-derive = "0.2.14"
 derive_builder = "0.7.1"
-threadpool = "1.7.1"
+number_prefix = "0.3.0"

 [profile.release]
 lto = true
--- a/wg-event-gen/src/gen.rs
+++ b/wg-event-gen/src/gen.rs
@@ -26,36 +26,40 @@ pub(crate) fn gen_events(
            (Some(prev), Some(cur)) => {
                if let (Some(prev_addr), Some(cur_addr)) = (prev.endpoint, cur.endpoint) {
                    if prev_addr != cur_addr {
-                        listeners.roaming(&cur, prev_addr);
+                        listeners.fire(Event::Roamed(&cur, prev_addr));
                    }
                }

-                let timedout_now = || {
-                    if let Some(shake) = cur.last_handshake {
-                        if let Ok(el) = shake.elapsed() {
-                            return el > timeout && el + poll_interval < timeout;
-                        }
-                    }
-                    true
-                };
+                let timedout_now = || !cur.last_handshake.is_some();

-                let timedout_prev = || {
-                    if let Some(shake) = prev.last_handshake {
-                        if let Ok(el) = shake.elapsed().map(|el| el - poll_interval) {
-                            return el > timeout && el < timeout + poll_interval;
-                        }
-                    }
-                    true
-                };
+                let timedout_prev = || !prev.last_handshake.is_some();

-                match (timedout_prev(), timedout_now()) {
-                    (false, true) => listeners.disconnected(&cur),
-                    (true, false) => listeners.connected(&cur),
-                    other => (),
+                match (
+                    timedout_prev(),
+                    timedout_now(),
+                    prev.last_handshake.and_then(|p_shake| {
+                        cur.last_handshake
+                            .and_then(|c_shake| c_shake.duration_since(p_shake).ok())
+                    }),
+                ) {
+                    (false, true, _) => listeners.fire(Event::Disconnected(&cur)),
+                    (true, false, _) => listeners.fire(Event::Connected(&cur)),
+                    other => {
+                        //dbg!(other);
+                    }
+                }
+                if prev.traffic != cur.traffic {
+                    if let ((p_tx, p_rx), (c_tx, c_rx)) = (prev.traffic, cur.traffic) {
+                        listeners.fire(Event::Transfered {
+                            peer: &cur,
+                            tx: c_tx - p_tx,
+                            rx: c_rx - p_rx,
+                        });
+                    }
                }
            }
-            (None, Some(cur)) => listeners.added(&cur),
-            (Some(prev), None) => listeners.removed(&prev),
+            (None, Some(cur)) => listeners.fire(Event::Added(&cur)),
+            (Some(prev), None) => listeners.fire(Event::Removed(&prev)),
            fail => {
                println!("{:?}", fail);
                unreachable!()
--- a/wg-event-gen/src/listener.rs
+++ b/wg-event-gen/src/listener.rs
@@ -1,56 +1,34 @@
 use crate::Peer;
+use number_prefix::{NumberPrefix, Prefixed, Standalone};
 use std::net::SocketAddr;
 use std::path::PathBuf;
 use std::process::Command;
 use std::thread;
 use std::time::SystemTime;
-use threadpool::ThreadPool;
+
+//#[non_exhaustive]
+#[derive(Debug, Clone)]
+pub enum Event<'a> {
+    Added(&'a Peer),
+    Removed(&'a Peer),
+    Connected(&'a Peer),
+    Disconnected(&'a Peer),
+    Roamed(&'a Peer, SocketAddr),
+    Transfered { peer: &'a Peer, tx: u64, rx: u64 },
+}

 pub trait EventListener {
    fn name(&self) -> &'static str;

-    fn added<'a>(&self, peer: &'a Peer) {
-        self.connected(peer);
-    }
-
-    fn connected<'a>(&self, peer: &'a Peer);
-
-    fn disconnected<'a>(&self, peer: &'a Peer);
-
-    fn removed<'a>(&self, peer: &'a Peer) {
-        self.disconnected(peer)
-    }
-
-    fn roaming<'a>(&self, peer: &'a Peer, previous_addr: SocketAddr);
+    fn fire<'a>(&self, event: Event<'a>);
 }

 impl EventListener for Vec<Box<EventListener>> {
    fn name(&self) -> &'static str {
        "List of Listeners"
    }
-
-    fn added<'a>(&self, peer: &'a Peer) {
-        if cfg!(feature = "addrem") || cfg!(test) {
-            self.iter().for_each(|l| l.added(&peer));
-        }
-    }
-
-    fn connected<'a>(&self, peer: &'a Peer) {
-        self.iter().for_each(|l| l.connected(&peer));
-    }
-
-    fn disconnected<'a>(&self, peer: &'a Peer) {
-        self.iter().for_each(|l| l.disconnected(&peer));
-    }
-
-    fn removed<'a>(&self, peer: &'a Peer) {
-        if cfg!(feature = "addrem") || cfg!(test) {
-            self.iter().for_each(|l| l.removed(&peer));
-        }
-    }
-
-    fn roaming<'a>(&self, peer: &'a Peer, previous_addr: SocketAddr) {
-        self.iter().for_each(|l| l.roaming(&peer, previous_addr));
+    fn fire<'a>(&self, event: Event<'a>) {
+        self.iter().for_each(|l| l.fire(event.clone()));
    }
 }

@@ -60,41 +38,44 @@ impl EventListener for LogListener {
    fn name(&self) -> &'static str {
        "Log"
    }
-    fn connected<'a>(&self, peer: &'a Peer) {
-        println!("{} connected!", peer.key);
-    }

-    fn disconnected<'a>(&self, peer: &'a Peer) {
-        println!("{} disconnected!", peer.key);
-    }
+    fn fire<'a>(&self, event: Event<'a>) {
+        match event {
+            Event::Connected(peer) => println!("{} connected!", peer.key),
+            Event::Disconnected(peer) => println!("{} disconnected!", peer.key),
+            Event::Added(peer) => println!("{} added!", peer.key),
+            Event::Removed(peer) => println!("{} removed!", peer.key),
+            Event::Roamed(peer, previous_addr) => println!(
+                "{} roamed {} -> {}!",
+                peer.key,
+                previous_addr,
+                peer.endpoint.unwrap()
+            ),
+            Event::Transfered { peer, tx, rx } => {
+                let prefix = |cnt: u64| match NumberPrefix::binary(cnt as f64) {
+                    Standalone(bytes) => format!("{}B", bytes),
+                    Prefixed(prefix, n) => format!("{:.0} {}B", n, prefix),
+                };
+                println!(
+                    "{} transfered {} up, {} down",
+                    peer.key,
+                    prefix(tx),
+                    prefix(rx)
+                );
+            }

-    fn added<'a>(&self, peer: &'a Peer) {
-        println!("{} added!", peer.key);
-    }
-
-    fn removed<'a>(&self, peer: &'a Peer) {
-        println!("{} removed!", peer.key);
-    }
-
-    fn roaming<'a>(&self, peer: &'a Peer, previous_addr: SocketAddr) {
-        println!(
-            "{} roamed {} -> {}!",
-            peer.key,
-            previous_addr,
-            peer.endpoint.unwrap()
-        );
+            _ => unimplemented!(),
+        }
    }
 }

 pub struct ScriptListener {
    pub script: PathBuf,
-    pub pool: ThreadPool,
 }

 impl ScriptListener {
    pub fn new(script: PathBuf) -> ScriptListener {
-        let pool = ThreadPool::new(4);
-        ScriptListener { script, pool }
+        ScriptListener { script }
    }

    fn peer_props<'a>(&self, peer: &'a Peer) -> String {
@@ -139,7 +120,7 @@ impl ScriptListener {

    fn call_sub<'a>(&self, args: Vec<&'a str>) {
        let mut cmd = self.mkcmd(args);
-        self.pool.execute(move || {
+        thread::spawn(move || {
            cmd.spawn().expect("Failed to call Script hooḱ!");
        });
    }
@@ -149,27 +130,26 @@ impl EventListener for ScriptListener {
    fn name(&self) -> &'static str {
        "Script"
    }
-    fn connected<'a>(&self, peer: &'a Peer) {
-        self.call_sub(vec!["connected", &self.peer_props(peer)]);
-    }
-
-    fn disconnected<'a>(&self, peer: &'a Peer) {
-        self.call_sub(vec!["disconnected", &self.peer_props(peer)]);
-    }
-
-    fn added<'a>(&self, peer: &'a Peer) {
-        self.call_sub(vec!["added", &self.peer_props(peer)]);
-    }
-
-    fn removed<'a>(&self, peer: &'a Peer) {
-        self.call_sub(vec!["removed", &self.peer_props(peer)]);
-    }
-
-    fn roaming<'a>(&self, peer: &'a Peer, previous_addr: SocketAddr) {
-        self.call_sub(vec![
-            "roaming",
-            &self.peer_props(peer),
-            &previous_addr.to_string(),
-        ]);
+    fn fire<'a>(&self, event: Event<'a>) {
+        match event {
+            Event::Connected(peer) => self.call_sub(vec!["connected", &self.peer_props(peer)]),
+            Event::Disconnected(peer) => {
+                self.call_sub(vec!["disconnected", &self.peer_props(peer)])
+            }
+            Event::Added(peer) => self.call_sub(vec!["added", &self.peer_props(peer)]),
+            Event::Removed(peer) => self.call_sub(vec!["removed", &self.peer_props(peer)]),
+            Event::Roamed(peer, previous_addr) => self.call_sub(vec![
+                "roaming",
+                &self.peer_props(peer),
+                &previous_addr.to_string(),
+            ]),
+            Event::Transfered { peer, tx, rx } => self.call_sub(vec![
+                "transfered",
+                &self.peer_props(peer),
+                &tx.to_string(),
+                &rx.to_string(),
+            ]),
+            _ => unimplemented!(),
+        }
    }
 }
Author	SHA1	Message	Date
shimunn	0301a53ac9	implemented listeners for transfered event [CI SKIP]	2019-04-04 21:18:21 +02:00
shimunn	ceb3d7b77d	Merge branch 'timeout' into new_events	2019-04-04 20:39:00 +02:00
shimunn	f3eeac404c	added transfered event [CI SKIP]	2019-04-04 20:34:54 +02:00
shimunn	a2cda16977	handle timeouts based on the existence of last_handshake	2019-04-04 18:59:17 +02:00
shimunn	ba247655bd	new event system [CI SKIP]	2019-04-03 17:54:53 +02:00