Support creating multicast netlink connection

examples/ip_monitor.rs

@@ -1,35 +1,8 @@
 // SPDX-License-Identifier: MIT
 
 use futures::stream::StreamExt;
-use netlink_sys::{AsyncSocket, SocketAddr};
-use rtnetlink::new_connection;
+use rtnetlink::{new_multicast_connection, MulticastGroup};
 
-const RTNLGRP_LINK: u32 = 1;
-const RTNLGRP_NEIGH: u32 = 3;
-const RTNLGRP_IPV4_IFADDR: u32 = 5;
-const RTNLGRP_IPV4_MROUTE: u32 = 6;
-const RTNLGRP_IPV4_ROUTE: u32 = 7;
-const RTNLGRP_IPV4_RULE: u32 = 8;
-const RTNLGRP_IPV6_IFADDR: u32 = 9;
-const RTNLGRP_IPV6_MROUTE: u32 = 10;
-const RTNLGRP_IPV6_ROUTE: u32 = 11;
-const RTNLGRP_IPV6_RULE: u32 = 19;
-const RTNLGRP_IPV4_NETCONF: u32 = 24;
-const RTNLGRP_IPV6_NETCONF: u32 = 25;
-const RTNLGRP_MPLS_ROUTE: u32 = 27;
-const RTNLGRP_NSID: u32 = 28;
-const RTNLGRP_MPLS_NETCONF: u32 = 29;
-
-const fn nl_mgrp(group: u32) -> u32 {
-    if group > 31 {
-        panic!("use netlink_sys::Socket::add_membership() for this group");
-    }
-    if group == 0 {
-        0
-    } else {
-        1 << (group - 1)
-    }
-}
 #[tokio::main]
 async fn main() -> Result<(), String> {
     // conn - `Connection` that has a netlink socket which is a `Future` that
@@ -39,42 +12,28 @@ async fn main() -> Result<(), String> {
     // messages.
     //
     // messages - A channel receiver.
-    let (mut conn, mut _handle, mut messages) =
-        new_connection().map_err(|e| format!("{e}"))?;
-
-    // These flags specify what kinds of broadcast messages we want to listen
-    // for.
-    let groups = nl_mgrp(RTNLGRP_LINK)
-        | nl_mgrp(RTNLGRP_IPV4_IFADDR)
-        | nl_mgrp(RTNLGRP_IPV6_IFADDR)
-        | nl_mgrp(RTNLGRP_IPV4_ROUTE)
-        | nl_mgrp(RTNLGRP_IPV6_ROUTE)
-        | nl_mgrp(RTNLGRP_MPLS_ROUTE)
-        | nl_mgrp(RTNLGRP_IPV4_MROUTE)
-        | nl_mgrp(RTNLGRP_IPV6_MROUTE)
-        | nl_mgrp(RTNLGRP_NEIGH)
-        | nl_mgrp(RTNLGRP_IPV4_NETCONF)
-        | nl_mgrp(RTNLGRP_IPV6_NETCONF)
-        | nl_mgrp(RTNLGRP_IPV4_RULE)
-        | nl_mgrp(RTNLGRP_IPV6_RULE)
-        | nl_mgrp(RTNLGRP_NSID)
-        | nl_mgrp(RTNLGRP_MPLS_NETCONF);
-
-    let addr = SocketAddr::new(0, groups);
-    conn.socket_mut()
-        .socket_mut()
-        .bind(&addr)
-        .expect("Failed to bind");
+    let (conn, mut _handle, mut messages) = new_multicast_connection(&[
+        MulticastGroup::Link,
+        MulticastGroup::Ipv4Ifaddr,
+        MulticastGroup::Ipv6Ifaddr,
+        MulticastGroup::Ipv4Route,
+        MulticastGroup::Ipv6Route,
+        MulticastGroup::MplsRoute,
+        MulticastGroup::Ipv4Mroute,
+        MulticastGroup::Ipv6Mroute,
+        MulticastGroup::Neigh,
+        MulticastGroup::Ipv4Netconf,
+        MulticastGroup::Ipv6Netconf,
+        MulticastGroup::Ipv4Rule,
+        MulticastGroup::Ipv6Rule,
+        MulticastGroup::Nsid,
+        MulticastGroup::MplsNetconf,
+    ])
+    .map_err(|e| format!("{e}"))?;
 
     // Spawn `Connection` to start polling netlink socket.
     tokio::spawn(conn);
 
-    // Use `Handle` to send request to kernel to start multicasting rtnetlink
-    // events.
-    tokio::spawn(async move {
-        // Create message to enable
-    });
-
     // Start receiving events through `messages` channel.
     while let Some((message, _)) = messages.next().await {
         let payload = message.payload;

src/connection.rs

@@ -8,7 +8,7 @@ use netlink_packet_route::RouteNetlinkMessage;
 use netlink_proto::Connection;
 use netlink_sys::{protocols::NETLINK_ROUTE, AsyncSocket, SocketAddr};
 
-use crate::Handle;
+use crate::{Handle, MulticastGroup};
 
 #[cfg(feature = "tokio_socket")]
 #[allow(clippy::type_complexity)]
@@ -20,6 +20,19 @@ pub fn new_connection() -> io::Result<(
     new_connection_with_socket()
 }
 
+/// Equal to `ip monitor` command
+#[cfg(feature = "tokio_socket")]
+#[allow(clippy::type_complexity)]
+pub fn new_multicast_connection(
+    groups: &[MulticastGroup],
+) -> io::Result<(
+    Connection<RouteNetlinkMessage>,
+    Handle,
+    UnboundedReceiver<(NetlinkMessage<RouteNetlinkMessage>, SocketAddr)>,
+)> {
+    new_multicast_connection_with_socket(groups)
+}
+
 #[allow(clippy::type_complexity)]
 pub fn new_connection_with_socket<S>() -> io::Result<(
     Connection<RouteNetlinkMessage, S>,
@@ -34,6 +47,39 @@ where
     Ok((conn, Handle::new(handle), messages))
 }
 
+/// Equal to `ip monitor` command
+#[allow(clippy::type_complexity)]
+pub fn new_multicast_connection_with_socket<S>(
+    groups: &[MulticastGroup],
+) -> io::Result<(
+    Connection<RouteNetlinkMessage, S>,
+    Handle,
+    UnboundedReceiver<(NetlinkMessage<RouteNetlinkMessage>, SocketAddr)>,
+)>
+where
+    S: AsyncSocket,
+{
+    let (mut conn, handle, messages) =
+        netlink_proto::new_connection_with_socket::<RouteNetlinkMessage, S>(
+            NETLINK_ROUTE,
+        )?;
+    let mut all_groups: u32 = 0;
+    for group in groups.iter().filter(|g| !g.need_via_add_membership()) {
+        all_groups |= 1 << (*group as u32 - 1);
+    }
+
+    let addr = SocketAddr::new(0, all_groups);
+    conn.socket_mut().socket_mut().bind(&addr)?;
+
+    for group in groups.iter().filter(|g| g.need_via_add_membership()) {
+        conn.socket_mut()
+            .socket_mut()
+            .add_membership(*group as u32)?;
+    }
+
+    Ok((conn, Handle::new(handle), messages))
+}
+
 #[allow(clippy::type_complexity)]
 pub fn from_socket<S>(
     socket: S,

src/lib.rs

@@ -17,6 +17,7 @@ mod errors;
 mod handle;
 mod link;
 mod macros;
+mod multicast;
 mod neighbour;
 #[cfg(not(target_os = "freebsd"))]
 mod ns;
@@ -26,7 +27,7 @@ mod rule;
 mod traffic_control;
 
 #[cfg(feature = "tokio_socket")]
-pub use crate::connection::new_connection;
+pub use crate::connection::{new_connection, new_multicast_connection};
 #[cfg(not(target_os = "freebsd"))]
 pub use crate::ns::{NetworkNamespace, NETNS_PATH, NONE_FS, SELF_NS_PATH};
 #[cfg(not(target_os = "freebsd"))]
@@ -41,7 +42,10 @@ pub use crate::{
         AddressAddRequest, AddressDelRequest, AddressGetRequest, AddressHandle,
         AddressMessageBuilder,
     },
-    connection::{from_socket, new_connection_with_socket},
+    connection::{
+        from_socket, new_connection_with_socket,
+        new_multicast_connection_with_socket,
+    },
     errors::Error,
     handle::Handle,
     link::{
@@ -51,6 +55,7 @@ pub use crate::{
         LinkUnspec, LinkVeth, LinkVlan, LinkVrf, LinkVxlan, LinkWireguard,
         LinkXfrm, QosMapping,
     },
+    multicast::MulticastGroup,
     neighbour::{
         NeighbourAddRequest, NeighbourDelRequest, NeighbourGetRequest,
         NeighbourHandle,

src/multicast.rs

@@ -0,0 +1,92 @@
+// SPDX-License-Identifier: MIT
+
+// const RTNLGRP_NONE: u32 = 0;
+const RTNLGRP_LINK: u32 = 1;
+const RTNLGRP_NOTIFY: u32 = 2;
+const RTNLGRP_NEIGH: u32 = 3;
+const RTNLGRP_TC: u32 = 4;
+const RTNLGRP_IPV4_IFADDR: u32 = 5;
+const RTNLGRP_IPV4_MROUTE: u32 = 6;
+const RTNLGRP_IPV4_ROUTE: u32 = 7;
+const RTNLGRP_IPV4_RULE: u32 = 8;
+const RTNLGRP_IPV6_IFADDR: u32 = 9;
+const RTNLGRP_IPV6_MROUTE: u32 = 10;
+const RTNLGRP_IPV6_ROUTE: u32 = 11;
+const RTNLGRP_IPV6_IFINFO: u32 = 12;
+const RTNLGRP_DECNET_IFADDR: u32 = 13;
+// const RTNLGRP_NOP2: u32 = 14
+const RTNLGRP_DECNET_ROUTE: u32 = 15;
+const RTNLGRP_DECNET_RULE: u32 = 16;
+// const RTNLGRP_NOP4: u32 = 17;
+const RTNLGRP_IPV6_PREFIX: u32 = 18;
+const RTNLGRP_IPV6_RULE: u32 = 19;
+const RTNLGRP_ND_USEROPT: u32 = 20;
+const RTNLGRP_PHONET_IFADDR: u32 = 21;
+const RTNLGRP_PHONET_ROUTE: u32 = 22;
+const RTNLGRP_DCB: u32 = 23;
+const RTNLGRP_IPV4_NETCONF: u32 = 24;
+const RTNLGRP_IPV6_NETCONF: u32 = 25;
+const RTNLGRP_MDB: u32 = 26;
+const RTNLGRP_MPLS_ROUTE: u32 = 27;
+const RTNLGRP_NSID: u32 = 28;
+const RTNLGRP_MPLS_NETCONF: u32 = 29;
+const RTNLGRP_IPV4_MROUTE_R: u32 = 30;
+const RTNLGRP_IPV6_MROUTE_R: u32 = 31;
+const RTNLGRP_NEXTHOP: u32 = 32;
+const RTNLGRP_BRVLAN: u32 = 33;
+const RTNLGRP_MCTP_IFADDR: u32 = 34;
+const RTNLGRP_TUNNEL: u32 = 35;
+const RTNLGRP_STATS: u32 = 36;
+const RTNLGRP_IPV4_MCADDR: u32 = 37;
+const RTNLGRP_IPV6_MCADDR: u32 = 38;
+const RTNLGRP_IPV6_ACADDR: u32 = 39;
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash)]
+#[repr(u32)]
+#[non_exhaustive]
+pub enum MulticastGroup {
+    Link = RTNLGRP_LINK,
+    Notify = RTNLGRP_NOTIFY,
+    Neigh = RTNLGRP_NEIGH,
+    Tc = RTNLGRP_TC,
+    Ipv4Ifaddr = RTNLGRP_IPV4_IFADDR,
+    Ipv4Mroute = RTNLGRP_IPV4_MROUTE,
+    Ipv4Route = RTNLGRP_IPV4_ROUTE,
+    Ipv4Rule = RTNLGRP_IPV4_RULE,
+    Ipv6Ifaddr = RTNLGRP_IPV6_IFADDR,
+    Ipv6Mroute = RTNLGRP_IPV6_MROUTE,
+    Ipv6Route = RTNLGRP_IPV6_ROUTE,
+    Ipv6Ifinfo = RTNLGRP_IPV6_IFINFO,
+    DecnetIfaddr = RTNLGRP_DECNET_IFADDR,
+    DecnetRoute = RTNLGRP_DECNET_ROUTE,
+    DecnetRule = RTNLGRP_DECNET_RULE,
+    Ipv6Prefix = RTNLGRP_IPV6_PREFIX,
+    Ipv6Rule = RTNLGRP_IPV6_RULE,
+    NdUseropt = RTNLGRP_ND_USEROPT,
+    PhonetIfaddr = RTNLGRP_PHONET_IFADDR,
+    PhonetRoute = RTNLGRP_PHONET_ROUTE,
+    Dcb = RTNLGRP_DCB,
+    Ipv4Netconf = RTNLGRP_IPV4_NETCONF,
+    Ipv6Netconf = RTNLGRP_IPV6_NETCONF,
+    Mdb = RTNLGRP_MDB,
+    MplsRoute = RTNLGRP_MPLS_ROUTE,
+    Nsid = RTNLGRP_NSID,
+    MplsNetconf = RTNLGRP_MPLS_NETCONF,
+    Ipv4MrouteR = RTNLGRP_IPV4_MROUTE_R,
+    Ipv6MrouteR = RTNLGRP_IPV6_MROUTE_R,
+    Nexthop = RTNLGRP_NEXTHOP,
+    Brvlan = RTNLGRP_BRVLAN,
+    MctpIfaddr = RTNLGRP_MCTP_IFADDR,
+    Tunnel = RTNLGRP_TUNNEL,
+    Stats = RTNLGRP_STATS,
+    Ipv4Mcaddr = RTNLGRP_IPV4_MCADDR,
+    Ipv6Mcaddr = RTNLGRP_IPV6_MCADDR,
+    Ipv6Acaddr = RTNLGRP_IPV6_ACADDR,
+}
+
+impl MulticastGroup {
+    /// Whether need to use `netlink_sys::Socket::add_membership()`.
+    pub fn need_via_add_membership(self) -> bool {
+        self as u32 > 31
+    }
+}