linux: Feature gate async runtime, allowing opting between Tokio or smol

CHANGELOG.md

@@ -4,6 +4,13 @@ All notable changes to this project will be documented in this file.
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
+## [3.0.0] - [unreleased]
+
+### Changed
+- Feature gate async runtime, allowing opting between Tokio or smol. For every OS each `IfWatcher` is
+  under the `tokio` or `smol` module. This makes it a breaking change as there
+  is no more a default implementation. See [PR 27](https://github.com/mxinden/if-watch/pull/27).
+
 ## [2.0.0]
 
 ### Changed

Cargo.toml

@@ -1,26 +1,33 @@
 [package]
 name = "if-watch"
-version = "2.0.0"
+version = "3.0.0"
 authors = ["David Craven <[email protected]>", "Parity Technologies Limited <[email protected]>"]
 edition = "2021"
 keywords = ["asynchronous", "routing"]
 license = "MIT OR Apache-2.0"
 description = "crossplatform asynchronous network watcher"
 repository = "https://github.com/mxinden/if-watch"
 
+[features]
+default = ["smol"]
+tokio = ["dep:tokio", "rtnetlink/tokio_socket"]
+smol = ["dep:smol", "rtnetlink/smol_socket"]
+
 [dependencies]
 fnv = "1.0.7"
 futures = "0.3.19"
 ipnet = "2.3.1"
 log = "0.4.14"
 
 [target.'cfg(target_os = "linux")'.dependencies]
-rtnetlink = { version = "0.10.0", default-features = false, features = ["smol_socket"] }
+rtnetlink = { version = "0.10.0", default-features = false }
 
 [target.'cfg(any(target_os = "macos", target_os = "ios"))'.dependencies]
 core-foundation = "0.9.2"
 if-addrs = "0.7.0"
 system-configuration = "0.5.0"
+tokio = { version = "1.21.2", features = ["rt"], optional = true }
+smol = { version = "1.2.5", optional = true }
 
 [target.'cfg(target_os = "windows")'.dependencies]
 if-addrs = "0.7.0"
@@ -32,3 +39,5 @@ if-addrs = "0.7.0"
 
 [dev-dependencies]
 env_logger = "0.9.0"
+smol = "1.2.5"
+tokio = { version = "1.21.2", features = ["rt", "macros"] }

examples/if_watch.rs

@@ -1,9 +1,9 @@
 use futures::StreamExt;
-use if_watch::IfWatcher;
+use if_watch::smol::IfWatcher;
 
 fn main() {
     env_logger::init();
-    futures::executor::block_on(async {
+    smol::block_on(async {
         let mut set = IfWatcher::new().unwrap();
         loop {
             let event = set.select_next_some().await;

src/apple.rs

@@ -4,31 +4,95 @@ use core_foundation::runloop::{kCFRunLoopCommonModes, CFRunLoop};
 use core_foundation::string::CFString;
 use fnv::FnvHashSet;
 use futures::channel::mpsc;
-use futures::stream::Stream;
+use futures::stream::{FusedStream, Stream};
+use futures::Future;
 use if_addrs::IfAddr;
 use std::collections::VecDeque;
 use std::io::Result;
+use std::marker::PhantomData;
 use std::pin::Pin;
 use std::task::{Context, Poll};
 use system_configuration::dynamic_store::{
     SCDynamicStore, SCDynamicStoreBuilder, SCDynamicStoreCallBackContext,
 };
 
+#[cfg(feature = "tokio")]
+pub mod tokio {
+    //! An interface watcher that uses the `tokio` runtime.
+    use futures::Future;
+
+    #[doc(hidden)]
+    pub struct TokioRuntime;
+
+    impl super::Runtime for TokioRuntime {
+        fn spawn<F>(f: F)
+        where
+            F: Future,
+            F: Send + 'static,
+            <F as Future>::Output: Send + 'static,
+        {
+            tokio::spawn(f);
+        }
+    }
+
+    /// Watches for interface changes.
+    pub type IfWatcher = super::IfWatcher<TokioRuntime>;
+}
+
+#[cfg(feature = "smol")]
+pub mod smol {
+    //! An interface watcher that uses the `smol` runtime.
+
+    use futures::Future;
+
+    #[doc(hidden)]
+    pub struct SmolRuntime;
+
+    impl super::Runtime for SmolRuntime {
+        fn spawn<F>(f: F)
+        where
+            F: Future,
+            F: Send + 'static,
+            <F as Future>::Output: Send + 'static,
+        {
+            smol::spawn(f).detach();
+        }
+    }
+
+    /// Watches for interface changes.
+    pub type IfWatcher = super::IfWatcher<SmolRuntime>;
+}
+
 #[derive(Debug)]
-pub struct IfWatcher {
+pub struct IfWatcher<T> {
     addrs: FnvHashSet<IpNet>,
     queue: VecDeque<IfEvent>,
     rx: mpsc::Receiver<()>,
+    runtime: PhantomData<T>,
 }
 
-impl IfWatcher {
+#[doc(hidden)]
+pub trait Runtime {
+    fn spawn<F>(f: F)
+    where
+        F: Future,
+        F: Send + 'static,
+        <F as Future>::Output: Send + 'static;
+}
+
+impl<T> IfWatcher<T>
+where
+    T: Runtime,
+{
+    /// Create a watcher.
     pub fn new() -> Result<Self> {
         let (tx, rx) = mpsc::channel(1);
-        std::thread::spawn(|| background_task(tx));
+        T::spawn(async { background_task(tx) });
         let mut watcher = Self {
             addrs: Default::default(),
             queue: Default::default(),
             rx,
+            runtime: PhantomData,
         };
         watcher.resync()?;
         Ok(watcher)
@@ -55,10 +119,12 @@ impl IfWatcher {
         Ok(())
     }
 
+    /// Iterate over current networks.
     pub fn iter(&self) -> impl Iterator<Item = &IpNet> {
         self.addrs.iter()
     }
 
+    /// Poll for an address change event.
     pub fn poll_if_event(&mut self, cx: &mut Context) -> Poll<Result<IfEvent>> {
         loop {
             if let Some(event) = self.queue.pop_front() {
@@ -74,6 +140,25 @@ impl IfWatcher {
     }
 }
 
+impl<T> Stream for IfWatcher<T>
+where
+    T: Runtime + Unpin,
+{
+    type Item = Result<IfEvent>;
+    fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
+        Pin::into_inner(self).poll_if_event(cx).map(Some)
+    }
+}
+
+impl<T> FusedStream for IfWatcher<T>
+where
+    T: Runtime + Unpin,
+{
+    fn is_terminated(&self) -> bool {
+        false
+    }
+}
+
 fn ifaddr_to_ipnet(addr: IfAddr) -> IpNet {
     match addr {
         IfAddr::V4(ip) => {

src/fallback.rs

@@ -1,6 +1,6 @@
 use crate::IfEvent;
 use async_io::Timer;
-use futures::stream::Stream;
+use futures::stream::{FusedStream, Stream};
 use if_addrs::IfAddr;
 use ipnet::{IpNet, Ipv4Net, Ipv6Net};
 use std::collections::{HashSet, VecDeque};
@@ -10,6 +10,26 @@ use std::pin::Pin;
 use std::task::{Context, Poll};
 use std::time::{Duration, Instant};
 
+#[cfg(feature = "tokio")]
+pub mod tokio {
+    //! An interface watcher.
+    //! **On this platform there is no difference between `tokio` and `smol` features,**
+    //! **this was done to maintain the api compatible with other platforms**.
+
+    /// Watches for interface changes.
+    pub type IfWatcher = super::IfWatcher;
+}
+
+#[cfg(feature = "smol")]
+pub mod smol {
+    //! An interface watcher.
+    //! **On this platform there is no difference between `tokio` and `smol` features,**
+    //! **this was done to maintain the api compatible with other platforms**.
+
+    /// Watches for interface changes.
+    pub type IfWatcher = super::IfWatcher;
+}
+
 /// An address set/watcher
 #[derive(Debug)]
 pub struct IfWatcher {
@@ -19,7 +39,7 @@ pub struct IfWatcher {
 }
 
 impl IfWatcher {
-    /// Create a watcher
+    /// Create a watcher.
     pub fn new() -> Result<Self> {
         Ok(Self {
             addrs: Default::default(),
@@ -45,10 +65,12 @@ impl IfWatcher {
         Ok(())
     }
 
+    /// Iterate over current networks.
     pub fn iter(&self) -> impl Iterator<Item = &IpNet> {
         self.addrs.iter()
     }
 
+    /// Poll for an address change event.
     pub fn poll_if_event(&mut self, cx: &mut Context) -> Poll<Result<IfEvent>> {
         loop {
             if let Some(event) = self.queue.pop_front() {
@@ -64,6 +86,19 @@ impl IfWatcher {
     }
 }
 
+impl Stream for IfWatcher {
+    type Item = Result<IfEvent>;
+    fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
+        Pin::into_inner(self).poll_if_event(cx).map(Some)
+    }
+}
+
+impl FusedStream for IfWatcher {
+    fn is_terminated(&self) -> bool {
+        false
+    }
+}
+
 fn ifaddr_to_ipnet(addr: IfAddr) -> IpNet {
     match addr {
         IfAddr::V4(ip) => {