ensure rtnetlink handle is not dropped before reading reply by wrapping it0.4.0

7 files changed, 555 insertions, 524 deletions

diff --git a/Cargo.toml b/Cargo.toml
index e028539..c0218ad 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -1,6 +1,6 @@
 [package]
 name = "rsdsl_netlinklib"
-version = "0.3.0"
+version = "0.4.0"
 edition = "2021"
 
 # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
@@ -9,6 +9,7 @@ edition = "2021"
 futures = { version = "0.3.11", default-features = false, features = ["std"] }
 libc = "0.2.150"
 netlink-packet-route = "^0.17"
+netlink-proto = "^0.11"
 rtnetlink = { version = "0.13.1" }
 thiserror = "1.0"
 tokio = { version = "1.0", features = ["time"] }
diff --git a/src/addr.rs b/src/addr.rs
index 36efe1a..97e87d4 100644
--- a/src/addr.rs
+++ b/src/addr.rs
@@ -1,6 +1,6 @@
 //! Simple functions to add and delete IP addresses.
 
-use crate::{Error, Result};
+use crate::{Connection, Error, Result};
 
 use std::net::{IpAddr, Ipv4Addr, Ipv6Addr};
 
@@ -9,216 +9,220 @@ use netlink_packet_route::{
     address::Nla, AddressMessage, AF_INET, AF_INET6, RT_SCOPE_LINK, RT_SCOPE_UNIVERSE,
 };
 
-/// Flushes all addresses of an interface.
-pub async fn flush(link: String) -> Result<()> {
-    let (conn, handle, _) = rtnetlink::new_connection()?;
-    tokio::spawn(conn);
-
-    let link = handle
-        .link()
-        .get()
-        .match_name(link.clone())
-        .execute()
-        .try_next()
-        .await?
-        .ok_or(Error::LinkNotFound(link))?;
-
-    let id = link.header.index;
-
-    let addrs: Vec<AddressMessage> = handle
-        .address()
-        .get()
-        .set_link_index_filter(id)
-        .execute()
-        .try_collect()
-        .await?;
-
-    for addr in addrs {
-        handle.address().del(addr).execute().await?;
+impl Connection {
+    /// Flushes all addresses of an interface.
+    pub async fn address_flush(&self, link: String) -> Result<()> {
+        let link = self
+            .handle()
+            .link()
+            .get()
+            .match_name(link.clone())
+            .execute()
+            .try_next()
+            .await?
+            .ok_or(Error::LinkNotFound(link))?;
+
+        let id = link.header.index;
+
+        let addrs: Vec<AddressMessage> = self
+            .handle()
+            .address()
+            .get()
+            .set_link_index_filter(id)
+            .execute()
+            .try_collect()
+            .await?;
+
+        for addr in addrs {
+            self.handle().address().del(addr).execute().await?;
+        }
+
+        Ok(())
     }
 
-    Ok(())
-}
-
-/// Flushes the IPv4 addresses of an interface.
-pub async fn flush4(link: String) -> Result<()> {
-    let (conn, handle, _) = rtnetlink::new_connection()?;
-    tokio::spawn(conn);
-
-    let link = handle
-        .link()
-        .get()
-        .match_name(link.clone())
-        .execute()
-        .try_next()
-        .await?
-        .ok_or(Error::LinkNotFound(link))?;
-
-    let id = link.header.index;
-
-    let addrs: Vec<AddressMessage> = handle
-        .address()
-        .get()
-        .set_link_index_filter(id)
-        .execute()
-        .try_filter(|addr| future::ready(addr.header.family == AF_INET as u8))
-        .try_collect()
-        .await?;
-
-    for addr in addrs {
-        handle.address().del(addr).execute().await?;
+    /// Flushes the IPv4 addresses of an interface.
+    pub async fn address_flush4(&self, link: String) -> Result<()> {
+        let link = self
+            .handle()
+            .link()
+            .get()
+            .match_name(link.clone())
+            .execute()
+            .try_next()
+            .await?
+            .ok_or(Error::LinkNotFound(link))?;
+
+        let id = link.header.index;
+
+        let addrs: Vec<AddressMessage> = self
+            .handle()
+            .address()
+            .get()
+            .set_link_index_filter(id)
+            .execute()
+            .try_filter(|addr| future::ready(addr.header.family == AF_INET as u8))
+            .try_collect()
+            .await?;
+
+        for addr in addrs {
+            self.handle().address().del(addr).execute().await?;
+        }
+
+        Ok(())
     }
 
-    Ok(())
-}
-
-/// Flushes the IPv6 addresses of an interface.
-pub async fn flush6(link: String) -> Result<()> {
-    let (conn, handle, _) = rtnetlink::new_connection()?;
-    tokio::spawn(conn);
-
-    let link = handle
-        .link()
-        .get()
-        .match_name(link.clone())
-        .execute()
-        .try_next()
-        .await?
-        .ok_or(Error::LinkNotFound(link))?;
-
-    let id = link.header.index;
-
-    let addrs: Vec<AddressMessage> = handle
-        .address()
-        .get()
-        .set_link_index_filter(id)
-        .execute()
-        .try_filter(|addr| future::ready(addr.header.family == AF_INET6 as u8))
-        .try_collect()
-        .await?;
-
-    for addr in addrs {
-        handle.address().del(addr).execute().await?;
+    /// Flushes the IPv6 addresses of an interface.
+    pub async fn address_flush6(&self, link: String) -> Result<()> {
+        let link = self
+            .handle()
+            .link()
+            .get()
+            .match_name(link.clone())
+            .execute()
+            .try_next()
+            .await?
+            .ok_or(Error::LinkNotFound(link))?;
+
+        let id = link.header.index;
+
+        let addrs: Vec<AddressMessage> = self
+            .handle()
+            .address()
+            .get()
+            .set_link_index_filter(id)
+            .execute()
+            .try_filter(|addr| future::ready(addr.header.family == AF_INET6 as u8))
+            .try_collect()
+            .await?;
+
+        for addr in addrs {
+            self.handle().address().del(addr).execute().await?;
+        }
+
+        Ok(())
     }
 
-    Ok(())
-}
-
-/// Flushes all global unicast IPv6 addresses from all interfaces.
-pub async fn flush6_global() -> Result<()> {
-    let (conn, handle, _) = rtnetlink::new_connection()?;
-    tokio::spawn(conn);
-
-    let addrs: Vec<AddressMessage> = handle
-        .address()
-        .get()
-        .execute()
-        .try_filter(|addr| {
-            future::ready(
-                addr.header.family == AF_INET6 as u8 && addr.header.scope == RT_SCOPE_UNIVERSE,
-            )
-        })
-        .try_collect()
-        .await?;
-
-    for addr in addrs {
-        handle.address().del(addr).execute().await?;
+    /// Flushes all global unicast IPv6 addresses from all interfaces.
+    pub async fn address_flush6_global(&self) -> Result<()> {
+        let addrs: Vec<AddressMessage> = self
+            .handle()
+            .address()
+            .get()
+            .execute()
+            .try_filter(|addr| {
+                future::ready(
+                    addr.header.family == AF_INET6 as u8 && addr.header.scope == RT_SCOPE_UNIVERSE,
+                )
+            })
+            .try_collect()
+            .await?;
+
+        for addr in addrs {
+            self.handle().address().del(addr).execute().await?;
+        }
+
+        Ok(())
     }
 
-    Ok(())
-}
-
-/// Adds an IP address to an interface.
-pub async fn add(link: String, addr: IpAddr, prefix_len: u8) -> Result<()> {
-    let (conn, handle, _) = rtnetlink::new_connection()?;
-    tokio::spawn(conn);
-
-    let link = handle
-        .link()
-        .get()
-        .match_name(link.clone())
-        .execute()
-        .try_next()
-        .await?
-        .ok_or(Error::LinkNotFound(link))?;
-
-    let id = link.header.index;
-
-    handle.address().add(id, addr, prefix_len).execute().await?;
-
-    Ok(())
-}
-
-/// Adds a link-scope IP address to an interface.
-/// This is especially useful with IPv6.
-pub async fn add_link_local(link: String, addr: IpAddr, prefix_len: u8) -> Result<()> {
-    let (conn, handle, _) = rtnetlink::new_connection()?;
-    tokio::spawn(conn);
-
-    let link = handle
-        .link()
-        .get()
-        .match_name(link.clone())
-        .execute()
-        .try_next()
-        .await?
-        .ok_or(Error::LinkNotFound(link))?;
-
-    let id = link.header.index;
-
-    let mut req = handle.address().add(id, addr, prefix_len);
-    req.message_mut().header.scope = RT_SCOPE_LINK;
-
-    req.execute().await?;
+    /// Adds an IP address to an interface.
+    pub async fn address_add(&self, link: String, addr: IpAddr, prefix_len: u8) -> Result<()> {
+        let link = self
+            .handle()
+            .link()
+            .get()
+            .match_name(link.clone())
+            .execute()
+            .try_next()
+            .await?
+            .ok_or(Error::LinkNotFound(link))?;
+
+        let id = link.header.index;
+
+        self.handle()
+            .address()
+            .add(id, addr, prefix_len)
+            .execute()
+            .await?;
+
+        Ok(())
+    }
 
-    Ok(())
-}
+    /// Adds a link-scope IP address to an interface.
+    /// This is especially useful with IPv6.
+    pub async fn address_add_link_local(
+        &self,
+        link: String,
+        addr: IpAddr,
+        prefix_len: u8,
+    ) -> Result<()> {
+        let link = self
+            .handle()
+            .link()
+            .get()
+            .match_name(link.clone())
+            .execute()
+            .try_next()
+            .await?
+            .ok_or(Error::LinkNotFound(link))?;
+
+        let id = link.header.index;
+
+        let mut req = self.handle().address().add(id, addr, prefix_len);
+        req.message_mut().header.scope = RT_SCOPE_LINK;
+
+        req.execute().await?;
+
+        Ok(())
+    }
 
-/// Returns an iterator over the IP addresses of an interface.
-pub async fn get(link: String) -> Result<impl TryStream<Ok = IpAddr, Error = Error>> {
-    let (conn, handle, _) = rtnetlink::new_connection()?;
-    tokio::spawn(conn);
-
-    let link = handle
-        .link()
-        .get()
-        .match_name(link.clone())
-        .execute()
-        .try_next()
-        .await?
-        .ok_or(Error::LinkNotFound(link))?;
-
-    Ok(handle
-        .address()
-        .get()
-        .set_link_index_filter(link.header.index)
-        .execute()
-        .err_into::<Error>()
-        .try_filter_map(|msg| {
-            future::ready(Ok(if let Some(Nla::Address(bytes)) = msg.nlas.first() {
-                match msg.header.family as u16 {
-                    AF_INET => {
-                        let octets: [u8; 4] = (*bytes)
-                            .clone()
-                            .try_into()
-                            .expect("nla does not match ipv4 address length");
-                        let ip = IpAddr::from(Ipv4Addr::from(octets));
-
-                        Some(ip)
+    /// Returns an iterator over the IP addresses of an interface.
+    pub async fn address_get(
+        &self,
+        link: String,
+    ) -> Result<impl TryStream<Ok = IpAddr, Error = Error>> {
+        let link = self
+            .handle()
+            .link()
+            .get()
+            .match_name(link.clone())
+            .execute()
+            .try_next()
+            .await?
+            .ok_or(Error::LinkNotFound(link))?;
+
+        Ok(self
+            .handle()
+            .address()
+            .get()
+            .set_link_index_filter(link.header.index)
+            .execute()
+            .err_into::<Error>()
+            .try_filter_map(|msg| {
+                future::ready(Ok(if let Some(Nla::Address(bytes)) = msg.nlas.first() {
+                    match msg.header.family as u16 {
+                        AF_INET => {
+                            let octets: [u8; 4] = (*bytes)
+                                .clone()
+                                .try_into()
+                                .expect("nla does not match ipv4 address length");
+                            let ip = IpAddr::from(Ipv4Addr::from(octets));
+
+                            Some(ip)
+                        }
+                        AF_INET6 => {
+                            let octets: [u8; 16] = (*bytes)
+                                .clone()
+                                .try_into()
+                                .expect("nla does not match ipv6 address length");
+                            let ip = IpAddr::from(Ipv6Addr::from(octets));
+
+                            Some(ip)
+                        }
+                        _ => None,
                     }
-                    AF_INET6 => {
-                        let octets: [u8; 16] = (*bytes)
-                            .clone()
-                            .try_into()
-                            .expect("nla does not match ipv6 address length");
-                        let ip = IpAddr::from(Ipv6Addr::from(octets));
-
-                        Some(ip)
-                    }
-                    _ => None,
-                }
-            } else {
-                None
+                } else {
+                    None
+                }))
             }))
-        }))
+    }
 }
diff --git a/src/blocking.rs b/src/blocking.rs
index f66a096..7e9a92b 100644
--- a/src/blocking.rs
+++ b/src/blocking.rs
@@ -9,72 +9,82 @@
 #[cfg(feature = "tunnel")]
 pub use crate::tunnel;
 
+/// A blocking wrapper around the async [`crate::Connection`].
+#[derive(Debug)]
+pub struct Connection(crate::Connection);
+
 macro_rules! blockify {
-    ($blk:ident, $r:path) => {
-        pub fn $blk() -> crate::Result<()> {
-            tokio::runtime::Runtime::new()?.block_on($r())
+    ($blk:ident) => {
+        pub fn $blk(&self) -> crate::Result<()> {
+            tokio::runtime::Runtime::new()?.block_on(self.0.$blk())
         }
     };
-    ($blk:ident, $r:path, $($v:tt: $t:ty),*) => {
-        pub fn $blk($($v: $t),*) -> crate::Result<()> {
-            tokio::runtime::Runtime::new()?.block_on($r($($v),*))
+    ($blk:ident, $($v:tt: $t:ty),*) => {
+        pub fn $blk(&self, $($v: $t),*) -> crate::Result<()> {
+            tokio::runtime::Runtime::new()?.block_on(self.0.$blk($($v),*))
         }
     };
-    ($blk:ident -> $ret:ty, $r:path, $($v:tt: $t:ty),*) => {
-        pub fn $blk($($v: $t),*) -> crate::Result<$ret> {
-            tokio::runtime::Runtime::new()?.block_on($r($($v),*))
+    ($blk:ident -> $ret:ty, $($v:tt: $t:ty),*) => {
+        pub fn $blk(&self, $($v: $t),*) -> crate::Result<$ret> {
+            tokio::runtime::Runtime::new()?.block_on(self.0.$blk($($v),*))
         }
     };
 }
 
 #[cfg(feature = "addr")]
 pub mod addr {
-    use crate::addr;
+    use super::Connection;
 
     use std::net::IpAddr;
 
     use futures::TryStreamExt;
 
-    blockify!(flush, addr::flush, link: String);
-    blockify!(flush4, addr::flush4, link: String);
-    blockify!(flush6, addr::flush6, link: String);
-    blockify!(flush6_global, addr::flush6_global);
-    blockify!(add, addr::add, link: String, addr: IpAddr, prefix_len: u8);
-    blockify!(add_link_local, addr::add_link_local, link: String, addr: IpAddr, prefix_len: u8);
+    impl Connection {
+        blockify!(address_flush, link: String);
+        blockify!(address_flush4, link: String);
+        blockify!(address_flush6, link: String);
+        blockify!(address_flush6_global);
+        blockify!(address_add, link: String, addr: IpAddr, prefix_len: u8);
+        blockify!(address_add_link_local, link: String, addr: IpAddr, prefix_len: u8);
 
-    pub fn get(link: String) -> crate::Result<Vec<IpAddr>> {
-        tokio::runtime::Runtime::new()?
-            .block_on(async { addr::get(link).await?.try_collect().await })
+        pub fn get(&self, link: String) -> crate::Result<Vec<IpAddr>> {
+            tokio::runtime::Runtime::new()?
+                .block_on(async { self.0.address_get(link).await?.try_collect().await })
+        }
     }
 }
 
 #[cfg(feature = "status")]
 pub mod link {
-    use crate::link;
+    use super::Connection;
 
-    #[cfg(feature = "link")]
-    blockify!(set, link::set, link: String, state: bool);
-    #[cfg(feature = "link")]
-    blockify!(set_mtu, link::set_mtu, link: String, mtu: u32);
-    #[cfg(feature = "link")]
-    blockify!(add_vlan, link::add_vlan, link: String, parent: String, vlan_id: u16);
+    impl Connection {
+        #[cfg(feature = "link")]
+        blockify!(link_set, link: String, state: bool);
+        #[cfg(feature = "link")]
+        blockify!(link_set_mtu, link: String, mtu: u32);
+        #[cfg(feature = "link")]
+        blockify!(link_add_vlan, link: String, parent: String, vlan_id: u16);
 
-    blockify!(is_up -> bool, link::is_up, link: String);
-    blockify!(wait_up, link::wait_up, link: String);
-    blockify!(exists -> bool, link::exists, link: String);
-    blockify!(wait_exists, link::wait_exists, link: String);
-    blockify!(index -> u32, link::index, link: String);
+        blockify!(link_is_up -> bool, link: String);
+        blockify!(link_wait_up, link: String);
+        blockify!(link_exists -> bool, link: String);
+        blockify!(link_wait_exists, link: String);
+        blockify!(link_index -> u32, link: String);
+    }
 }
 
 #[cfg(feature = "route")]
 pub mod route {
-    use crate::route;
+    use super::Connection;
 
     use std::net::{Ipv4Addr, Ipv6Addr};
 
-    blockify!(flush4, route::flush4, link: String);
-    blockify!(flush6, route::flush6, link: String);
-    blockify!(flush, route::flush, link: String);
-    blockify!(add4, route::add4, dst: Ipv4Addr, prefix_len: u8, rtr: Option<Ipv4Addr>, link: String);
-    blockify!(add6, route::add6, dst: Ipv6Addr, prefix_len: u8, rtr: Option<Ipv6Addr>, link: String);
+    impl Connection {
+        blockify!(route_flush4, link: String);
+        blockify!(route_flush6, link: String);
+        blockify!(route_flush, link: String);
+        blockify!(route_add4, dst: Ipv4Addr, prefix_len: u8, rtr: Option<Ipv4Addr>, link: String);
+        blockify!(route_add6, dst: Ipv6Addr, prefix_len: u8, rtr: Option<Ipv6Addr>, link: String);
+    }
 }
diff --git a/src/connection.rs b/src/connection.rs
new file mode 100644
index 0000000..37d31f4
--- /dev/null
+++ b/src/connection.rs
@@ -0,0 +1,23 @@
+use crate::Result;
+
+use rtnetlink::Handle;
+
+#[derive(Debug)]
+pub struct Connection(Handle);
+
+impl Connection {
+    /// Creates a new connection and handle to rtnetlink and spawns the connection task.
+    /// Can be used to interact with rtnetlink by enabling certain crate features
+    /// and calling the methods they provide.
+    pub fn new() -> Result<Self> {
+        let (conn, handle, _) = rtnetlink::new_connection()?;
+        tokio::spawn(conn);
+
+        Ok(Self(handle))
+    }
+
+    /// Returns a reference to the underlying [`rtnetlink::Handle`].
+    pub(crate) fn handle(&self) -> &Handle {
+        &self.0
+    }
+}
diff --git a/src/lib.rs b/src/lib.rs
index bea9dc2..e417659 100644
--- a/src/lib.rs
+++ b/src/lib.rs
@@ -1,3 +1,6 @@
+mod connection;
+pub use connection::*;
+
 mod error;
 pub use error::*;
 
diff --git a/src/link.rs b/src/link.rs
index ffbaa99..30a432b 100644
--- a/src/link.rs
+++ b/src/link.rs
@@ -1,6 +1,6 @@
 //! Simple functions to add, monitor and configure network interfaces.
 
-use crate::{Error, Result};
+use crate::{Connection, Error, Result};
 
 use std::time::Duration;
 
@@ -9,158 +9,148 @@ use tokio::time::sleep;
 use futures::TryStreamExt;
 use netlink_packet_route::rtnl::IFF_UP;
 
-/// Brings an interface up or down.
-#[cfg(feature = "link")]
-pub async fn set(link: String, state: bool) -> Result<()> {
-    let (conn, handle, _) = rtnetlink::new_connection()?;
-    tokio::spawn(conn);
-
-    let link = handle
-        .link()
-        .get()
-        .match_name(link.clone())
-        .execute()
-        .try_next()
-        .await?
-        .ok_or(Error::LinkNotFound(link))?;
-
-    let id = link.header.index;
-
-    match state {
-        true => handle.link().set(id).up(),
-        false => handle.link().set(id).down(),
-    }
-    .execute()
-    .await?;
-
-    Ok(())
-}
-
-/// Reports whether an interface is up.
-///
-/// # Errors
-///
-/// This function fails if the interface doesn't exist
-/// or if any other `rtnetlink` error occurs.
-pub async fn is_up(link: String) -> Result<bool> {
-    let (conn, handle, _) = rtnetlink::new_connection()?;
-    tokio::spawn(conn);
-
-    let link = handle
-        .link()
-        .get()
-        .match_name(link.clone())
-        .execute()
-        .try_next()
-        .await?
-        .ok_or(Error::LinkNotFound(link))?;
-
-    let is_up = link.header.flags & IFF_UP == IFF_UP;
-    Ok(is_up)
-}
-
-/// Sets the MTU of an interface.
-#[cfg(feature = "link")]
-pub async fn set_mtu(link: String, mtu: u32) -> Result<()> {
-    let (conn, handle, _) = rtnetlink::new_connection()?;
-    tokio::spawn(conn);
-
-    let link = handle
-        .link()
-        .get()
-        .match_name(link.clone())
-        .execute()
-        .try_next()
-        .await?
-        .ok_or(Error::LinkNotFound(link))?;
-
-    let id = link.header.index;
-
-    handle.link().set(id).mtu(mtu).execute().await?;
-    Ok(())
-}
-
-/// Creates a VLAN interface on top of a parent interface.
-///
-/// # Arguments
-///
-/// * `link` - The name of the VLAN interface to be created.
-/// * `parent` - The name of the parent interface for the actual traffic.
-/// * `vlan_id` - The VLAN ID for tagging.
-#[cfg(feature = "link")]
-pub async fn add_vlan(link: String, parent: String, vlan_id: u16) -> Result<()> {
-    let (conn, handle, _) = rtnetlink::new_connection()?;
-    tokio::spawn(conn);
-
-    let parent = handle
-        .link()
-        .get()
-        .match_name(parent.clone())
-        .execute()
-        .try_next()
-        .await?
-        .ok_or(Error::LinkNotFound(parent))?;
-
-    let parent_id = parent.header.index;
-
-    handle
-        .link()
-        .add()
-        .vlan(link, parent_id, vlan_id)
+impl Connection {
+    /// Brings an interface up or down.
+    #[cfg(feature = "link")]
+    pub async fn link_set(&self, link: String, state: bool) -> Result<()> {
+        let link = self
+            .handle()
+            .link()
+            .get()
+            .match_name(link.clone())
+            .execute()
+            .try_next()
+            .await?
+            .ok_or(Error::LinkNotFound(link))?;
+
+        let id = link.header.index;
+
+        match state {
+            true => self.handle().link().set(id).up(),
+            false => self.handle().link().set(id).down(),
+        }
         .execute()
         .await?;
 
-    Ok(())
-}
-
-/// Waits for an interface to come up, including waiting for its creation.
-pub async fn wait_up(link: String) -> Result<()> {
-    while !exists(link.clone()).await? || !is_up(link.clone()).await? {
-        sleep(Duration::from_millis(200)).await;
+        Ok(())
     }
 
-    Ok(())
-}
+    /// Reports whether an interface is up.
+    ///
+    /// # Errors
+    ///
+    /// This function fails if the interface doesn't exist
+    /// or if any other `rtnetlink` error occurs.
+    pub async fn link_is_up(&self, link: String) -> Result<bool> {
+        let link = self
+            .handle()
+            .link()
+            .get()
+            .match_name(link.clone())
+            .execute()
+            .try_next()
+            .await?
+            .ok_or(Error::LinkNotFound(link))?;
+
+        let is_up = link.header.flags & IFF_UP == IFF_UP;
+        Ok(is_up)
+    }
 
-/// Reports whether an interface exists.
-pub async fn exists(link: String) -> Result<bool> {
-    let (conn, handle, _) = rtnetlink::new_connection()?;
-    tokio::spawn(conn);
+    /// Sets the MTU of an interface.
+    #[cfg(feature = "link")]
+    pub async fn link_set_mtu(&self, link: String, mtu: u32) -> Result<()> {
+        let link = self
+            .handle()
+            .link()
+            .get()
+            .match_name(link.clone())
+            .execute()
+            .try_next()
+            .await?
+            .ok_or(Error::LinkNotFound(link))?;
+
+        let id = link.header.index;
+
+        self.handle().link().set(id).mtu(mtu).execute().await?;
+        Ok(())
+    }
 
-    let exists = handle
-        .link()
-        .get()
-        .match_name(link)
-        .execute()
-        .try_next()
-        .await
-        .is_ok();
+    /// Creates a VLAN interface on top of a parent interface.
+    ///
+    /// # Arguments
+    ///
+    /// * `link` - The name of the VLAN interface to be created.
+    /// * `parent` - The name of the parent interface for the actual traffic.
+    /// * `vlan_id` - The VLAN ID for tagging.
+    #[cfg(feature = "link")]
+    pub async fn link_add_vlan(&self, link: String, parent: String, vlan_id: u16) -> Result<()> {
+        let parent = self
+            .handle()
+            .link()
+            .get()
+            .match_name(parent.clone())
+            .execute()
+            .try_next()
+            .await?
+            .ok_or(Error::LinkNotFound(parent))?;
+
+        let parent_id = parent.header.index;
+
+        self.handle()
+            .link()
+            .add()
+            .vlan(link, parent_id, vlan_id)
+            .execute()
+            .await?;
+
+        Ok(())
+    }
 
-    Ok(exists)
-}
+    /// Waits for an interface to come up, including waiting for its creation.
+    pub async fn link_wait_up(&self, link: String) -> Result<()> {
+        while !self.link_exists(link.clone()).await? || !self.link_is_up(link.clone()).await? {
+            sleep(Duration::from_millis(200)).await;
+        }
 
-/// Waits until an interface is created.
-pub async fn wait_exists(link: String) -> Result<()> {
-    while !exists(link.clone()).await? {
-        sleep(Duration::from_millis(200)).await;
+        Ok(())
     }
 
-    Ok(())
-}
+    /// Reports whether an interface exists.
+    pub async fn link_exists(&self, link: String) -> Result<bool> {
+        let exists = self
+            .handle()
+            .link()
+            .get()
+            .match_name(link)
+            .execute()
+            .try_next()
+            .await
+            .is_ok();
+
+        Ok(exists)
+    }
 
-/// Returns the index of an interface.
-pub async fn index(link: String) -> Result<u32> {
-    let (conn, handle, _) = rtnetlink::new_connection()?;
-    tokio::spawn(conn);
+    /// Waits until an interface is created.
+    pub async fn link_wait_exists(&self, link: String) -> Result<()> {
+        while !self.link_exists(link.clone()).await? {
+            sleep(Duration::from_millis(200)).await;
+        }
 
-    let link = handle
-        .link()
-        .get()
-        .match_name(link.clone())
-        .execute()
-        .try_next()
-        .await?
-        .ok_or(Error::LinkNotFound(link))?;
+        Ok(())
+    }
 
-    Ok(link.header.index)
+    /// Returns the index of an interface.
+    pub async fn link_index(&self, link: String) -> Result<u32> {
+        let link = self
+            .handle()
+            .link()
+            .get()
+            .match_name(link.clone())
+            .execute()
+            .try_next()
+            .await?
+            .ok_or(Error::LinkNotFound(link))?;
+
+        Ok(link.header.index)
+    }
 }
diff --git a/src/route.rs b/src/route.rs
index 765fd23..6b11054 100644
--- a/src/route.rs
+++ b/src/route.rs
@@ -1,6 +1,6 @@
 //! Simple functions to add and delete routes.
 
-use crate::{Error, Result};
+use crate::{Connection, Error, Result};
 
 use std::net::{Ipv4Addr, Ipv6Addr};
 
@@ -8,160 +8,160 @@ use futures::{future, TryStreamExt};
 use netlink_packet_route::{RouteMessage, RT_SCOPE_LINK};
 use rtnetlink::IpVersion;
 
-/// Flushes all IPv4 routes from an interface.
-pub async fn flush4(link: String) -> Result<()> {
-    let (conn, handle, _) = rtnetlink::new_connection()?;
-    tokio::spawn(conn);
-
-    let link = handle
-        .link()
-        .get()
-        .match_name(link.clone())
-        .execute()
-        .try_next()
-        .await?
-        .ok_or(Error::LinkNotFound(link))?;
-
-    let id = link.header.index;
-
-    let routes: Vec<RouteMessage> = handle
-        .route()
-        .get(IpVersion::V4)
-        .execute()
-        .try_filter(|route| {
-            future::ready(if let Some(ifi) = route.output_interface() {
-                ifi == id
-            } else {
-                false
+impl Connection {
+    /// Flushes all IPv4 routes from an interface.
+    pub async fn route_flush4(&self, link: String) -> Result<()> {
+        let link = self
+            .handle()
+            .link()
+            .get()
+            .match_name(link.clone())
+            .execute()
+            .try_next()
+            .await?
+            .ok_or(Error::LinkNotFound(link))?;
+
+        let id = link.header.index;
+
+        let routes: Vec<RouteMessage> = self
+            .handle()
+            .route()
+            .get(IpVersion::V4)
+            .execute()
+            .try_filter(|route| {
+                future::ready(if let Some(ifi) = route.output_interface() {
+                    ifi == id
+                } else {
+                    false
+                })
             })
-        })
-        .try_collect()
-        .await?;
+            .try_collect()
+            .await?;
 
-    for route in routes {
-        handle.route().del(route).execute().await?;
-    }
+        for route in routes {
+            self.handle().route().del(route).execute().await?;
+        }
 
-    Ok(())
-}
+        Ok(())
+    }
 
-/// Flushes all IPv6 routes from an interface.
-pub async fn flush6(link: String) -> Result<()> {
-    let (conn, handle, _) = rtnetlink::new_connection()?;
-    tokio::spawn(conn);
-
-    let link = handle
-        .link()
-        .get()
-        .match_name(link.clone())
-        .execute()
-        .try_next()
-        .await?
-        .ok_or(Error::LinkNotFound(link))?;
-
-    let id = link.header.index;
-
-    let routes: Vec<RouteMessage> = handle
-        .route()
-        .get(IpVersion::V6)
-        .execute()
-        .try_filter(|route| {
-            future::ready(if let Some(ifi) = route.output_interface() {
-                ifi == id
-            } else {
-                false
+    /// Flushes all IPv6 routes from an interface.
+    pub async fn route_flush6(&self, link: String) -> Result<()> {
+        let link = self
+            .handle()
+            .link()
+            .get()
+            .match_name(link.clone())
+            .execute()
+            .try_next()
+            .await?
+            .ok_or(Error::LinkNotFound(link))?;
+
+        let id = link.header.index;
+
+        let routes: Vec<RouteMessage> = self
+            .handle()
+            .route()
+            .get(IpVersion::V6)
+            .execute()
+            .try_filter(|route| {
+                future::ready(if let Some(ifi) = route.output_interface() {
+                    ifi == id
+                } else {
+                    false
+                })
             })
-        })
-        .try_collect()
-        .await?;
+            .try_collect()
+            .await?;
 
-    for route in routes {
-        handle.route().del(route).execute().await?;
-    }
-
-    Ok(())
-}
+        for route in routes {
+            self.handle().route().del(route).execute().await?;
+        }
 
-/// Flushes all routes from an interface.
-pub async fn flush(link: String) -> Result<()> {
-    flush4(link.clone()).await?;
-    flush6(link).await?;
+        Ok(())
+    }
 
-    Ok(())
-}
+    /// Flushes all routes from an interface.
+    pub async fn route_flush(&self, link: String) -> Result<()> {
+        self.route_flush4(link.clone()).await?;
+        self.route_flush6(link).await?;
 
-/// Adds a simple IPv4 route with an optional gateway.
-pub async fn add4(
-    dst: Ipv4Addr,
-    prefix_len: u8,
-    rtr: Option<Ipv4Addr>,
-    link: String,
-) -> Result<()> {
-    let (conn, handle, _) = rtnetlink::new_connection()?;
-    tokio::spawn(conn);
-
-    let link = handle
-        .link()
-        .get()
-        .match_name(link.clone())
-        .execute()
-        .try_next()
-        .await?
-        .ok_or(Error::LinkNotFound(link))?;
-
-    let id = link.header.index;
-
-    let mut add = handle
-        .route()
-        .add()
-        .v4()
-        .destination_prefix(dst, prefix_len)
-        .output_interface(id);
-
-    if let Some(rtr) = rtr {
-        add = add.gateway(rtr);
-    } else {
-        add = add.scope(RT_SCOPE_LINK);
+        Ok(())
     }
 
-    add.execute().await?;
-    Ok(())
-}
-
-/// Adds a simple IPv6 route with an optional gateway.
-pub async fn add6(
-    dst: Ipv6Addr,
-    prefix_len: u8,
-    rtr: Option<Ipv6Addr>,
-    link: String,
-) -> Result<()> {
-    let (conn, handle, _) = rtnetlink::new_connection()?;
-    tokio::spawn(conn);
-
-    let link = handle
-        .link()
-        .get()
-        .match_name(link.clone())
-        .execute()
-        .try_next()
-        .await?
-        .ok_or(Error::LinkNotFound(link))?;
-
-    let id = link.header.index;
-
-    let mut add = handle
-        .route()
-        .add()
-        .v6()
-        .destination_prefix(dst, prefix_len)
-        .output_interface(id);
-
-    if let Some(rtr) = rtr {
-        add = add.gateway(rtr);
-    } else {
-        add = add.scope(RT_SCOPE_LINK);
+    /// Adds a simple IPv4 route with an optional gateway.
+    pub async fn route_add4(
+        &self,
+        dst: Ipv4Addr,
+        prefix_len: u8,
+        rtr: Option<Ipv4Addr>,
+        link: String,
+    ) -> Result<()> {
+        let link = self
+            .handle()
+            .link()
+            .get()
+            .match_name(link.clone())
+            .execute()
+            .try_next()
+            .await?
+            .ok_or(Error::LinkNotFound(link))?;
+
+        let id = link.header.index;
+
+        let mut add = self
+            .handle()
+            .route()
+            .add()
+            .v4()
+            .destination_prefix(dst, prefix_len)
+            .output_interface(id);
+
+        if let Some(rtr) = rtr {
+            add = add.gateway(rtr);
+        } else {
+            add = add.scope(RT_SCOPE_LINK);
+        }
+
+        add.execute().await?;
+        Ok(())
     }
 
-    add.execute().await?;
-    Ok(())
+    /// Adds a simple IPv6 route with an optional gateway.
+    pub async fn route_add6(
+        &self,
+        dst: Ipv6Addr,
+        prefix_len: u8,
+        rtr: Option<Ipv6Addr>,
+        link: String,
+    ) -> Result<()> {
+        let link = self
+            .handle()
+            .link()
+            .get()
+            .match_name(link.clone())
+            .execute()
+            .try_next()
+            .await?
+            .ok_or(Error::LinkNotFound(link))?;
+
+        let id = link.header.index;
+
+        let mut add = self
+            .handle()
+            .route()
+            .add()
+            .v6()
+            .destination_prefix(dst, prefix_len)
+            .output_interface(id);
+
+        if let Some(rtr) = rtr {
+            add = add.gateway(rtr);
+        } else {
+            add = add.scope(RT_SCOPE_LINK);
+        }
+
+        add.execute().await?;
+        Ok(())
+    }
 }