Make watching work on Linux

diff --git a/src/lib.rs b/src/lib.rs
index 7a8298ef647be7e99b178e97dcde113234607661..69de586c46db24fc7cf395107661093d569ed2f5 100644 (file)
--- a/src/lib.rs
+++ b/src/lib.rs
@@ -145,6 +145,6 @@ pub fn list_interfaces() -> Result<HashMap<IfIndex, Interface>, Error> {
 /// 
 /// The callback will fire once immediately with an initial interface list, and a diff as if
 /// there were originally no interfaces present.
 /// 
 /// The callback will fire once immediately with an initial interface list, and a diff as if
 /// there were originally no interfaces present.

-pub fn watch_interfaces<F: FnMut(Update) + 'static>(callback: F) -> Result<WatchHandle, Error> {
+pub fn watch_interfaces<F: FnMut(Update) + Send + 'static>(callback: F) -> Result<WatchHandle, Error> {

     watch::watch_interfaces(callback).map(|handle| WatchHandle { _inner: handle })
 }
     watch::watch_interfaces(callback).map(|handle| WatchHandle { _inner: handle })
 }

diff --git a/src/watch_linux.rs b/src/watch_linux.rs
index 4e4049b8856ba87dece34d25677f1116c49ae7a5..efc3842482207998e40f36e868e259af4bc44c72 100644 (file)
--- a/src/watch_linux.rs
+++ b/src/watch_linux.rs
@@ -1,7 +1,6 @@
 use std::os::fd::AsRawFd;
 use std::os::fd::OwnedFd;
 
 use std::os::fd::AsRawFd;
 use std::os::fd::OwnedFd;
 

-use nix::libc::nlmsghdr;

 use nix::libc::RTMGRP_IPV4_IFADDR;
 use nix::libc::RTMGRP_IPV6_IFADDR;
 use nix::libc::RTMGRP_LINK;
 use nix::libc::RTMGRP_IPV4_IFADDR;
 use nix::libc::RTMGRP_IPV6_IFADDR;
 use nix::libc::RTMGRP_LINK;


@@ -16,21 +15,23 @@ use nix::sys::socket::SockProtocol;
 use nix::sys::socket::SockType;
 
 use crate::Error;
 use nix::sys::socket::SockType;
 
 use crate::Error;

+use crate::List;

 use crate::Update;
 
 pub(crate) struct WatchHandle {
     // PROBLEM: close() doesn't cancel recv() for a netlink socket
 use crate::Update;
 
 pub(crate) struct WatchHandle {
     // PROBLEM: close() doesn't cancel recv() for a netlink socket

+    // SOLUTION: open a pipe() and use poll() inside the thread to watch for cancellation too

     sockfd: OwnedFd,
 }
 
     sockfd: OwnedFd,
 }
 

-pub(crate) fn watch_interfaces<F: FnMut(Update) + 'static>(
+pub(crate) fn watch_interfaces<F: FnMut(Update) + Send + 'static>(

     callback: F,
 ) -> Result<WatchHandle, Error> {
     let sockfd = start_watcher_thread(callback)?;
     Ok(WatchHandle { sockfd })
 }
 
     callback: F,
 ) -> Result<WatchHandle, Error> {
     let sockfd = start_watcher_thread(callback)?;
     Ok(WatchHandle { sockfd })
 }
 

-fn start_watcher_thread<F: FnMut(Update) + 'static>(callback: F) -> Result<OwnedFd, Error> {
+fn start_watcher_thread<F: FnMut(Update) + Send + 'static>(mut callback: F) -> Result<OwnedFd, Error> {

     let sockfd = socket(AddressFamily::Netlink, SockType::Raw, SockFlag::empty(), Some(SockProtocol::NetlinkRoute))
         .map_err(|_| Error::Internal)?; // TODO: proper errors
     let sa_nl = NetlinkAddr::new(0, (RTMGRP_LINK | RTMGRP_IPV4_IFADDR | RTMGRP_IPV6_IFADDR) as u32);
     let sockfd = socket(AddressFamily::Netlink, SockType::Raw, SockFlag::empty(), Some(SockProtocol::NetlinkRoute))
         .map_err(|_| Error::Internal)?; // TODO: proper errors
     let sa_nl = NetlinkAddr::new(0, (RTMGRP_LINK | RTMGRP_IPV4_IFADDR | RTMGRP_IPV6_IFADDR) as u32);


@@ -40,15 +41,30 @@ fn start_watcher_thread<F: FnMut(Update) + 'static>(callback: F) -> Result<Owned
 
     std::thread::spawn(move || {
         println!("watch thread running");
 
     std::thread::spawn(move || {
         println!("watch thread running");

+        let mut prev_list = List::default();

         let mut buf = [0u8; 4096];
         let mut buf = [0u8; 4096];

-        // recvmsg?
+        let mut handle_update = move |new_list: List| {
+            if new_list == prev_list {
+                return;
+            }
+            let update = Update {
+                interfaces: new_list.0.clone(),
+                diff: new_list.diff_from(&prev_list),
+            };
+            (callback)(update);
+            prev_list = new_list;
+        };
+
+        if let Ok(initial) = crate::list::list_interfaces() {
+            handle_update(initial);
+        };
+

         while let Ok(n) = recv(fd, &mut buf, MsgFlags::empty()) {
             println!("something on the netlink socket: {} bytes", n);
         while let Ok(n) = recv(fd, &mut buf, MsgFlags::empty()) {
             println!("something on the netlink socket: {} bytes", n);

-            let nlmsg_ptr = &buf as *const _ as *const nlmsghdr;
-            let nlmsg = unsafe { &*nlmsg_ptr };
-            // Right conventionally there's some trick here involving macros NLMSG_OK
-            // I can presumably do this using NetlinkGeneric too
-            // It's unclear whether this is worse or not - need to know what those macros do
+            let Ok(new_list) = crate::list::list_interfaces() else {
+                continue;
+            };
+            handle_update(new_list);

         }
         println!("netlink recv thread terminating");
     });
         }
         println!("netlink recv thread terminating");
     });

diff --git a/src/watch_win.rs b/src/watch_win.rs
index a225f4a20c16c95f0cae051681345b21f0d96c15..edd2fe38d2f96587ed46cddb7b5fe714826317b0 100644 (file)
--- a/src/watch_win.rs
+++ b/src/watch_win.rs
@@ -23,7 +23,7 @@ struct WatchState {
     /// The last result that we captured, for diffing
     prev_list: List,
     /// User's callback
     /// The last result that we captured, for diffing
     prev_list: List,
     /// User's callback

-    cb: Box<dyn FnMut(Update) + 'static>,
+    cb: Box<dyn FnMut(Update) + Send + 'static>,

 }
 
 pub(crate) struct WatchHandle {
 }
 
 pub(crate) struct WatchHandle {


@@ -39,7 +39,7 @@ impl Drop for WatchHandle {
     }
 }
 
     }
 }
 

-pub(crate) fn watch_interfaces<F: FnMut(Update) + 'static>(
+pub(crate) fn watch_interfaces<F: FnMut(Update) + Send + 'static>(

     callback: F,
 ) -> Result<WatchHandle, Error> {
     let state = Box::pin(Mutex::new(WatchState {
     callback: F,
 ) -> Result<WatchHandle, Error> {
     let state = Box::pin(Mutex::new(WatchState {