transports/tcp: simplify IfWatcher integration

transports/tcp/Cargo.toml

@@ -14,9 +14,7 @@ categories = ["network-programming", "asynchronous"]
 async-io-crate = { package = "async-io", version = "1.2.0", optional = true }
 futures = "0.3.8"
 futures-timer = "3.0"
-if-watch = { version = "1.1.1", optional = true }
-if-addrs = { version = "0.7.0", optional = true }
-ipnet = "2.0.0"
+if-watch = { version = "2.0.0", git = "https://github.com/mxinden/if-watch.git" }
 libc = "0.2.80"
 libp2p-core = { version = "0.35.0", path = "../../core", default-features = false  }
 log = "0.4.11"
@@ -25,10 +23,10 @@ tokio-crate = { package = "tokio", version = "1.19.0", default-features = false,
 
 [features]
 default = ["async-io"]
-tokio = ["tokio-crate", "if-addrs"]
-async-io = ["async-io-crate", "if-watch"]
+tokio = ["tokio-crate"]
+async-io = ["async-io-crate"]
 
 [dev-dependencies]
 async-std = { version = "1.6.5", features = ["attributes"] }
-tokio-crate = { package = "tokio", version = "1.0.1", default-features = false, features = ["net", "rt"] }
+tokio-crate = { package = "tokio", version = "1.0.1", default-features = false, features = ["net", "rt", "macros"] }
 env_logger = "0.9.0"

transports/tcp/src/lib.rs

@@ -28,6 +28,7 @@
 
 mod provider;
 
+use if_watch::{IfEvent, IfWatcher};
 #[cfg(feature = "async-io")]
 pub use provider::async_io;
 
@@ -43,9 +44,8 @@ pub use provider::tokio;
 pub type TokioTcpTransport = GenTcpTransport<tokio::Tcp>;
 
 use futures::{
-    future::{self, BoxFuture, Ready},
+    future::{self, Ready},
     prelude::*,
-    ready,
 };
 use futures_timer::Delay;
 use libp2p_core::{
@@ -64,7 +64,7 @@ use std::{
     time::Duration,
 };
 
-use provider::{IfEvent, Provider};
+use provider::Provider;
 
 /// The configuration for a TCP/IP transport capability for libp2p.
 #[derive(Clone, Debug)]
@@ -243,6 +243,9 @@ impl GenTcpConfig {
     /// # use libp2p_core::transport::{ListenerId, TransportEvent};
     /// # use libp2p_core::{Multiaddr, Transport};
     /// # use std::pin::Pin;
+    /// # #[cfg(not(feature = "async-io"))]
+    /// # fn main() {}
+    /// #
     /// #[cfg(feature = "async-io")]
     /// #[async_std::main]
     /// async fn main() -> std::io::Result<()> {
@@ -398,7 +401,6 @@ impl<T> Transport for GenTcpTransport<T>
 where
     T: Provider + Send + 'static,
     T::Listener: Unpin,
-    T::IfWatcher: Unpin,
     T::Stream: Unpin,
 {
     type Output = T::Stream;
@@ -605,23 +607,12 @@ pub enum TcpListenerEvent<S> {
     Error(io::Error),
 }
 
-enum IfWatch<TIfWatcher> {
-    Pending(BoxFuture<'static, io::Result<TIfWatcher>>),
-    Ready(TIfWatcher),
-}
-
 /// The listening addresses of a [`TcpListenStream`].
-enum InAddr<TIfWatcher> {
+enum InAddr {
     /// The stream accepts connections on a single interface.
-    One {
-        addr: IpAddr,
-        out: Option<Multiaddr>,
-    },
+    One(Option<Multiaddr>),
     /// The stream accepts connections on all interfaces.
-    Any {
-        addrs: HashSet<IpAddr>,
-        if_watch: IfWatch<TIfWatcher>,
-    },
+    Any(IfWatcher),
 }
 
 /// A stream of incoming connections on one or more interfaces.
@@ -642,7 +633,7 @@ where
     ///
     /// If the listen socket listens on all interfaces, these may change over
     /// time as interfaces become available or unavailable.
-    in_addr: InAddr<T::IfWatcher>,
+    in_addr: InAddr,
     /// The port reuse configuration for outgoing connections.
     ///
     /// If enabled, all IP addresses on which this listening stream
@@ -676,15 +667,9 @@ where
         } {
             // The `addrs` are populated via `if_watch` when the
             // `TcpListenStream` is polled.
-            InAddr::Any {
-                addrs: HashSet::new(),
-                if_watch: IfWatch::Pending(T::if_watcher()),
-            }
+            InAddr::Any(IfWatcher::new()?)
         } else {
-            InAddr::One {
-                out: Some(ip_to_multiaddr(listen_addr.ip(), listen_addr.port())),
-                addr: listen_addr.ip(),
-            }
+            InAddr::One(Some(ip_to_multiaddr(listen_addr.ip(), listen_addr.port())))
         };
 
         let listener = T::new_listener(listener)?;
@@ -708,12 +693,14 @@ where
     /// Has no effect if port reuse is disabled.
     fn disable_port_reuse(&mut self) {
         match &self.in_addr {
-            InAddr::One { addr, .. } => {
-                self.port_reuse.unregister(*addr, self.listen_addr.port());
+            InAddr::One(_) => {
+                self.port_reuse
+                    .unregister(self.listen_addr.ip(), self.listen_addr.port());
             }
-            InAddr::Any { addrs, .. } => {
-                for addr in addrs {
-                    self.port_reuse.unregister(*addr, self.listen_addr.port());
+            InAddr::Any(if_watcher) => {
+                for ip_net in if_watcher.iter() {
+                    self.port_reuse
+                        .unregister(ip_net.addr(), self.listen_addr.port());
                 }
             }
         }
@@ -734,7 +721,6 @@ where
     T: Provider,
     T::Listener: Unpin,
     T::Stream: Unpin,
-    T::IfWatcher: Unpin,
 {
     type Item = Result<TcpListenerEvent<T::Stream>, io::Error>;
 
@@ -743,68 +729,46 @@ where
 
         loop {
             match &mut me.in_addr {
-                InAddr::Any { if_watch, addrs } => match if_watch {
-                    // If we listen on all interfaces, wait for `if-watch` to be ready.
-                    IfWatch::Pending(f) => match ready!(Pin::new(f).poll(cx)) {
-                        Ok(w) => {
-                            *if_watch = IfWatch::Ready(w);
-                            continue;
-                        }
-                        Err(err) => {
-                            log::debug! {
-                                "Failed to begin observing interfaces: {:?}. Scheduling retry.",
-                                err
-                            };
-                            *if_watch = IfWatch::Pending(T::if_watcher());
-                            me.pause = Some(Delay::new(me.sleep_on_error));
-                            return Poll::Ready(Some(Ok(TcpListenerEvent::Error(err))));
-                        }
-                    },
-                    // Consume all events for up/down interface changes.
-                    IfWatch::Ready(watch) => {
-                        while let Poll::Ready(ev) = T::poll_interfaces(watch, cx) {
-                            match ev {
-                                Ok(IfEvent::Up(inet)) => {
-                                    let ip = inet.addr();
-                                    if me.listen_addr.is_ipv4() == ip.is_ipv4() && addrs.insert(ip)
-                                    {
-                                        let ma = ip_to_multiaddr(ip, me.listen_addr.port());
-                                        log::debug!("New listen address: {}", ma);
-                                        me.port_reuse.register(ip, me.listen_addr.port());
-                                        return Poll::Ready(Some(Ok(
-                                            TcpListenerEvent::NewAddress(ma),
-                                        )));
-                                    }
-                                }
-                                Ok(IfEvent::Down(inet)) => {
-                                    let ip = inet.addr();
-                                    if me.listen_addr.is_ipv4() == ip.is_ipv4() && addrs.remove(&ip)
-                                    {
-                                        let ma = ip_to_multiaddr(ip, me.listen_addr.port());
-                                        log::debug!("Expired listen address: {}", ma);
-                                        me.port_reuse.unregister(ip, me.listen_addr.port());
-                                        return Poll::Ready(Some(Ok(
-                                            TcpListenerEvent::AddressExpired(ma),
-                                        )));
-                                    }
+                InAddr::Any(if_watcher) => {
+                    while let Poll::Ready(ev) = IfWatcher::poll_next(Pin::new(if_watcher), cx) {
+                        match ev {
+                            Ok(IfEvent::Up(inet)) => {
+                                let ip = inet.addr();
+                                if me.listen_addr.is_ipv4() == ip.is_ipv4() {
+                                    let ma = ip_to_multiaddr(ip, me.listen_addr.port());
+                                    log::debug!("New listen address: {}", ma);
+                                    me.port_reuse.register(ip, me.listen_addr.port());
+                                    return Poll::Ready(Some(Ok(TcpListenerEvent::NewAddress(ma))));
                                 }
-                                Err(err) => {
-                                    log::debug! {
-                                        "Failure polling interfaces: {:?}. Scheduling retry.",
-                                        err
-                                    };
-                                    me.pause = Some(Delay::new(me.sleep_on_error));
-                                    return Poll::Ready(Some(Ok(TcpListenerEvent::Error(err))));
+                            }
+                            Ok(IfEvent::Down(inet)) => {
+                                let ip = inet.addr();
+                                if me.listen_addr.is_ipv4() == ip.is_ipv4() {
+                                    let ma = ip_to_multiaddr(ip, me.listen_addr.port());
+                                    log::debug!("Expired listen address: {}", ma);
+                                    me.port_reuse.unregister(ip, me.listen_addr.port());
+                                    return Poll::Ready(Some(Ok(
+                                        TcpListenerEvent::AddressExpired(ma),
+                                    )));
                                 }
                             }
+                            Err(err) => {
+                                log::debug! {
+                                    "Failure polling interfaces: {:?}. Scheduling retry.",
+                                    err
+                                };
+                                me.pause = Some(Delay::new(me.sleep_on_error));
+                                return Poll::Ready(Some(Ok(TcpListenerEvent::Error(err))));
+                            }
                         }
                     }
-                },
+                }
                 // If the listener is bound to a single interface, make sure the
                 // address is registered for port reuse and reported once.
-                InAddr::One { addr, out } => {
+                InAddr::One(out) => {
                     if let Some(multiaddr) = out.take() {
-                        me.port_reuse.register(*addr, me.listen_addr.port());
+                        me.port_reuse
+                            .register(me.listen_addr.ip(), me.listen_addr.port());
                         return Poll::Ready(Some(Ok(TcpListenerEvent::NewAddress(multiaddr))));
                     }
                 }

transports/tcp/src/provider.rs

@@ -28,18 +28,10 @@ pub mod tokio;
 
 use futures::future::BoxFuture;
 use futures::io::{AsyncRead, AsyncWrite};
-use ipnet::IpNet;
 use std::net::{SocketAddr, TcpListener, TcpStream};
 use std::task::{Context, Poll};
 use std::{fmt, io};
 
-/// An event relating to a change of availability of an address
-/// on a network interface.
-pub enum IfEvent {
-    Up(IpNet),
-    Down(IpNet),
-}
-
 /// An incoming connection returned from [`Provider::poll_accept()`].
 pub struct Incoming<S> {
     pub stream: S,
@@ -54,12 +46,6 @@ pub trait Provider: Clone + Send + 'static {
     type Stream: AsyncRead + AsyncWrite + Send + Unpin + fmt::Debug;
     /// The type of TCP listeners obtained from [`Provider::new_listener`].
     type Listener: Send + Unpin;
-    /// The type of network interface observers obtained from [`Provider::if_watcher`].
-    type IfWatcher: Send + Unpin;
-
-    /// Creates an instance of [`Self::IfWatcher`] that can be polled for
-    /// network interface changes via [`Self::poll_interfaces`].
-    fn if_watcher() -> BoxFuture<'static, io::Result<Self::IfWatcher>>;
 
     /// Creates a new listener wrapping the given [`TcpListener`] that
     /// can be polled for incoming connections via [`Self::poll_accept()`].
@@ -77,8 +63,4 @@ pub trait Provider: Clone + Send + 'static {
         _: &mut Self::Listener,
         _: &mut Context<'_>,
     ) -> Poll<io::Result<Incoming<Self::Stream>>>;
-
-    /// Polls a [`Self::IfWatcher`] for network interface changes, ensuring a task wakeup,
-    /// if necessary.
-    fn poll_interfaces(_: &mut Self::IfWatcher, _: &mut Context<'_>) -> Poll<io::Result<IfEvent>>;
 }

transports/tcp/src/provider/async_io.rs

@@ -18,7 +18,7 @@
 // FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
 // DEALINGS IN THE SOFTWARE.
 
-use super::{IfEvent, Incoming, Provider};
+use super::{Incoming, Provider};
 
 use async_io_crate::Async;
 use futures::future::{BoxFuture, FutureExt};
@@ -32,11 +32,6 @@ pub enum Tcp {}
 impl Provider for Tcp {
     type Stream = Async<net::TcpStream>;
     type Listener = Async<net::TcpListener>;
-    type IfWatcher = if_watch::IfWatcher;
-
-    fn if_watcher() -> BoxFuture<'static, io::Result<Self::IfWatcher>> {
-        if_watch::IfWatcher::new().boxed()
-    }
 
     fn new_listener(l: net::TcpListener) -> io::Result<Self::Listener> {
         Async::new(l)
@@ -87,11 +82,4 @@ impl Provider for Tcp {
             remote_addr,
         }))
     }
-
-    fn poll_interfaces(w: &mut Self::IfWatcher, cx: &mut Context<'_>) -> Poll<io::Result<IfEvent>> {
-        w.poll_unpin(cx).map_ok(|e| match e {
-            if_watch::IfEvent::Up(a) => IfEvent::Up(a),
-            if_watch::IfEvent::Down(a) => IfEvent::Down(a),
-        })
-    }
 }