feat: use watchable instead of callbacks for netmon changes

netwatch/Cargo.toml

@@ -19,6 +19,7 @@ workspace = true
 atomic-waker = "1.1.2"
 bytes = "1.7"
 n0-future = "0.1.3"
+n0-watcher = "0.1"
 nested_enum_utils = "0.2.0"
 snafu = "0.8.5"
 time = "0.3.20"

netwatch/src/interfaces.rs

@@ -29,10 +29,14 @@ use self::bsd::default_route;
 use self::linux::default_route;
 #[cfg(target_os = "windows")]
 use self::windows::default_route;
+#[cfg(not(wasm_browser))]
+use crate::ip::is_link_local;
 use crate::ip::{is_private_v6, is_up};
+#[cfg(not(wasm_browser))]
+use crate::netmon::is_interesting_interface;
 
 /// Represents a network interface.
-#[derive(Debug)]
+#[derive(Debug, Clone)]
 pub struct Interface {
     iface: netdev::interface::Interface,
 }
@@ -61,12 +65,12 @@ impl Eq for Interface {}
 
 impl Interface {
     /// Is this interface up?
-    pub(crate) fn is_up(&self) -> bool {
+    pub fn is_up(&self) -> bool {
         is_up(&self.iface)
     }
 
     /// The name of the interface.
-    pub(crate) fn name(&self) -> &str {
+    pub fn name(&self) -> &str {
         &self.iface.name
     }
 
@@ -153,7 +157,7 @@ impl IpNet {
 
 /// Intended to store the state of the machine's network interfaces, routing table, and
 /// other network configuration. For now it's pretty basic.
-#[derive(Debug, PartialEq, Eq)]
+#[derive(Debug, PartialEq, Eq, Clone)]
 pub struct State {
     /// Maps from an interface name interface.
     pub interfaces: HashMap<String, Interface>,
@@ -165,22 +169,16 @@ pub struct State {
     /// Whether the machine has some non-localhost, non-link-local IPv4 address.
     pub have_v4: bool,
 
-    //// Whether the current network interface is considered "expensive", which currently means LTE/etc
+    /// Whether the current network interface is considered "expensive", which currently means LTE/etc
     /// instead of Wifi. This field is not populated by `get_state`.
-    pub(crate) is_expensive: bool,
+    pub is_expensive: bool,
 
     /// The interface name for the machine's default route.
     ///
     /// It is not yet populated on all OSes.
     ///
     /// When set, its value is the map key into `interface` and `interface_ips`.
-    pub(crate) default_route_interface: Option<String>,
-
-    /// The HTTP proxy to use, if any.
-    pub(crate) http_proxy: Option<String>,
-
-    /// The URL to the Proxy Autoconfig URL, if applicable.
-    pub(crate) pac: Option<String>,
+    pub default_route_interface: Option<String>,
 }
 
 impl fmt::Display for State {
@@ -241,8 +239,6 @@ impl State {
             have_v6,
             is_expensive: false,
             default_route_interface,
-            http_proxy: None,
-            pac: None,
         }
     }
 
@@ -258,10 +254,42 @@ impl State {
             have_v4: true,
             is_expensive: false,
             default_route_interface: Some(ifname),
-            http_proxy: None,
-            pac: None,
         }
     }
+
+    /// Is this a major change compared to the `old` one?.
+    #[cfg(wasm_browser)]
+    pub fn is_major_change(&self, old: &State) -> bool {
+        // All changes are major.
+        // In the browser, there only are changes from online to offline
+        self != old
+    }
+
+    /// Is this a major change compared to the `old` one?.
+    #[cfg(not(wasm_browser))]
+    pub fn is_major_change(&self, old: &State) -> bool {
+        if self.have_v6 != old.have_v6
+            || self.have_v4 != old.have_v4
+            || self.is_expensive != old.is_expensive
+            || self.default_route_interface != old.default_route_interface
+        {
+            return true;
+        }
+
+        for (iname, i) in &old.interfaces {
+            if !is_interesting_interface(i.name()) {
+                continue;
+            }
+            let Some(i2) = self.interfaces.get(iname) else {
+                return true;
+            };
+            if i != i2 || !prefixes_major_equal(i.addrs(), i2.addrs()) {
+                return true;
+            }
+        }
+
+        false
+    }
 }
 
 /// Reports whether ip is a usable IPv4 address which should have Internet connectivity.
@@ -373,6 +401,30 @@ impl HomeRouter {
     }
 }
 
+/// Checks whether `a` and `b` are equal after ignoring uninteresting
+/// things, like link-local, loopback and multicast addresses.
+#[cfg(not(wasm_browser))]
+fn prefixes_major_equal(a: impl Iterator<Item = IpNet>, b: impl Iterator<Item = IpNet>) -> bool {
+    fn is_interesting(p: &IpNet) -> bool {
+        let a = p.addr();
+        if is_link_local(a) || a.is_loopback() || a.is_multicast() {
+            return false;
+        }
+        true
+    }
+
+    let a = a.filter(is_interesting);
+    let b = b.filter(is_interesting);
+
+    for (a, b) in a.zip(b) {
+        if a != b {
+            return false;
+        }
+    }
+
+    true
+}
+
 #[cfg(test)]
 mod tests {
     use std::net::Ipv6Addr;

netwatch/src/interfaces/wasm_browser.rs

@@ -5,7 +5,7 @@ use js_sys::{JsString, Reflect};
 pub const BROWSER_INTERFACE: &str = "browserif";
 
 /// Represents a network interface.
-#[derive(Debug, PartialEq, Eq)]
+#[derive(Clone, Debug, PartialEq, Eq)]
 pub struct Interface {
     is_up: bool,
 }
@@ -45,7 +45,7 @@ impl Interface {
 
 /// Intended to store the state of the machine's network interfaces, routing table, and
 /// other network configuration. For now it's pretty basic.
-#[derive(Debug, PartialEq, Eq)]
+#[derive(Clone, Debug, PartialEq, Eq)]
 pub struct State {
     /// Maps from an interface name interface.
     pub interfaces: HashMap<String, Interface>,

netwatch/src/netmon.rs

@@ -1,9 +1,7 @@
 //! Monitoring of networking interfaces and route changes.
 
-use n0_future::{
-    boxed::BoxFuture,
-    task::{self, AbortOnDropHandle},
-};
+use n0_future::task::{self, AbortOnDropHandle};
+use n0_watcher::Watchable;
 use nested_enum_utils::common_fields;
 use snafu::{Backtrace, ResultExt, Snafu};
 use tokio::sync::{mpsc, oneshot};
@@ -26,15 +24,18 @@ mod wasm_browser;
 #[cfg(target_os = "windows")]
 mod windows;
 
-pub use self::actor::CallbackToken;
+#[cfg(not(wasm_browser))]
+pub(crate) use self::actor::is_interesting_interface;
 use self::actor::{Actor, ActorMessage};
+pub use crate::interfaces::State;
 
 /// Monitors networking interface and route changes.
 #[derive(Debug)]
 pub struct Monitor {
     /// Task handle for the monitor task.
     _handle: AbortOnDropHandle<()>,
     actor_tx: mpsc::Sender<ActorMessage>,
+    interface_state: Watchable<State>,
 }
 
 #[common_fields({
@@ -66,6 +67,7 @@ impl Monitor {
     pub async fn new() -> Result<Self, Error> {
         let actor = Actor::new().await.context(ActorSnafu)?;
         let actor_tx = actor.subscribe();
+        let interface_state = actor.state().clone();
 
         let handle = task::spawn(async move {
             actor.run().await;
@@ -74,30 +76,13 @@ impl Monitor {
         Ok(Monitor {
             _handle: AbortOnDropHandle::new(handle),
             actor_tx,
+            interface_state,
         })
     }
 
     /// Subscribe to network changes.
-    pub async fn subscribe<F>(&self, callback: F) -> Result<CallbackToken, Error>
-    where
-        F: Fn(bool) -> BoxFuture<()> + 'static + Sync + Send,
-    {
-        let (s, r) = oneshot::channel();
-        self.actor_tx
-            .send(ActorMessage::Subscribe(Box::new(callback), s))
-            .await?;
-        let token = r.await?;
-        Ok(token)
-    }
-
-    /// Unsubscribe a callback from network changes, using the provided token.
-    pub async fn unsubscribe(&self, token: CallbackToken) -> Result<(), Error> {
-        let (s, r) = oneshot::channel();
-        self.actor_tx
-            .send(ActorMessage::Unsubscribe(token, s))
-            .await?;
-        r.await?;
-        Ok(())
+    pub fn interface_state(&self) -> n0_watcher::Direct<State> {
+        self.interface_state.watch()
     }
 
     /// Potential change detected outside
@@ -109,23 +94,16 @@ impl Monitor {
 
 #[cfg(test)]
 mod tests {
-    use n0_future::future::FutureExt;
+    use n0_watcher::Watcher as _;
 
     use super::*;
 
     #[tokio::test]
     async fn test_smoke_monitor() {
         let mon = Monitor::new().await.unwrap();
-        let _token = mon
-            .subscribe(|is_major| {
-                async move {
-                    println!("CHANGE DETECTED: {}", is_major);
-                }
-                .boxed()
-            })
-            .await
-            .unwrap();
-
-        tokio::time::sleep(std::time::Duration::from_secs(15)).await;
+        let sub = mon.interface_state();
+
+        let current = sub.get().unwrap();
+        println!("current state: {}", current);
     }
 }