netlink_socket2/
multicast.rs1use std::{io, os::fd::AsRawFd, sync::Arc};
2
3use netlink_bindings::utils;
4
5use crate::{NetlinkReplyInner, NetlinkSocket, ReplyError, Socket, RECV_BUF_SIZE};
6
7#[derive(Debug, Clone)]
8pub struct MulticastRecv {
9 pub multicast_group: u32,
10 pub message_type: u16,
11}
12
13pub struct MulticastSocketRaw {
14 buf: Arc<[u8; RECV_BUF_SIZE]>,
15 sock: Socket,
16 reply: NetlinkReplyInner,
17 last_group: Option<u32>,
18}
19
20impl MulticastSocketRaw {
21 pub fn new(protonum: u16) -> io::Result<Self> {
22 let sock = NetlinkSocket::get_socket_new(protonum)?;
23
24 let res = unsafe {
26 libc::setsockopt(
27 sock.as_raw_fd(),
28 libc::SOL_NETLINK,
29 libc::NETLINK_PKTINFO,
30 &1u32 as *const u32 as *const libc::c_void,
31 4,
32 )
33 };
34 if res < 0 {
35 return Err(io::Error::from_raw_os_error(-res));
36 }
37
38 let mut buf: libc::sockaddr_nl = unsafe { std::mem::zeroed() };
39 buf.nl_family = libc::AF_NETLINK as u16;
40 buf.nl_groups = 0;
41
42 let res = unsafe {
43 libc::bind(
44 sock.as_raw_fd(),
45 &buf as *const _ as *const libc::sockaddr,
46 std::mem::size_of_val(&buf) as libc::socklen_t,
47 )
48 };
49 if res < 0 {
50 return Err(io::Error::last_os_error());
51 }
52
53 Ok(Self {
54 buf: Arc::new([0u8; RECV_BUF_SIZE]),
55 sock,
56 reply: NetlinkReplyInner {
57 buf_offset: 0,
58 buf_read: 0,
59 },
60 last_group: None,
61 })
62 }
63
64 pub fn listen(&mut self, group_id: u32) -> io::Result<()> {
65 let res = unsafe {
66 libc::setsockopt(
67 self.sock.as_raw_fd(),
68 libc::SOL_NETLINK,
69 libc::NETLINK_ADD_MEMBERSHIP,
70 &group_id as *const u32 as *const libc::c_void,
71 4,
72 )
73 };
74 if res < 0 {
75 return Err(io::Error::last_os_error());
76 }
77
78 Ok(())
79 }
80
81 #[cfg_attr(not(feature = "async"), maybe_async::maybe_async)]
82 pub async fn recv(&mut self) -> Result<(MulticastRecv, &[u8]), ReplyError> {
83 let buf = Arc::make_mut(&mut self.buf);
84
85 loop {
86 if self.reply.buf_offset == self.reply.buf_read {
87 let read = Self::read_buf(&mut self.sock, buf, &mut self.last_group).await?;
88 self.reply.buf_read = read;
89 self.reply.buf_offset = 0;
90 }
91
92 match self.reply.parse_next(buf).await {
93 Err(io_err) => {
94 return Err(io_err.into());
95 }
96 Ok((seq, message_type, res)) => {
97 if seq != 0 {
98 continue;
99 }
100
101 let Some(multicast_group) = self.last_group else {
102 continue;
103 };
104
105 match res {
106 Ok((l, r)) => {
107 return Ok((
108 MulticastRecv {
109 multicast_group,
110 message_type,
111 },
112 &self.buf[l..r],
113 ));
114 }
115 Err(mut err) => {
116 if err.code.raw_os_error().unwrap() == 0 {
117 continue;
118 }
119
120 if err.has_context() {
121 err.reply_buf = Some(self.buf.clone());
123 }
124
125 return Err(err);
126 }
127 };
128 }
129 };
130 }
131 }
132
133 #[cfg_attr(not(feature = "async"), maybe_async::maybe_async)]
134 async fn read_buf(
135 sock: &Socket,
136 buf: &mut [u8],
137 last_group: &mut Option<u32>,
138 ) -> Result<usize, ReplyError> {
139 loop {
140 #[cfg(feature = "async")]
141 sock.readable().await?;
142
143 let mut addr: libc::sockaddr_nl = unsafe { std::mem::zeroed() };
144 let mut iov = libc::iovec {
145 iov_base: buf.as_mut_ptr() as *mut libc::c_void,
146 iov_len: buf.len(),
147 };
148
149 let mut control_buf = [0u8; 128];
150
151 let mut msghdr = libc::msghdr {
152 msg_name: &mut addr as *mut libc::sockaddr_nl as *mut libc::c_void,
153 msg_namelen: std::mem::size_of_val(&addr) as u32,
154 msg_iov: &mut iov as *mut libc::iovec,
155 msg_iovlen: 1,
156 msg_control: control_buf.as_mut_ptr() as *mut libc::c_void,
157 msg_controllen: control_buf.len(),
158 msg_flags: 0,
159 };
160
161 let read = unsafe { libc::recvmsg(sock.as_raw_fd(), &mut msghdr, 0) };
162 if read < 0 {
163 let err = io::Error::last_os_error();
164 #[cfg(feature = "async")]
165 if err.kind() == std::io::ErrorKind::WouldBlock {
166 continue;
167 }
168 return Err(err.into());
169 }
170
171 *last_group = None;
172 unsafe {
173 let msghdr_ptr = &msghdr as *const libc::msghdr;
174 let mut cmsg_ptr: *const libc::cmsghdr = libc::CMSG_FIRSTHDR(msghdr_ptr);
175 while !cmsg_ptr.is_null() {
176 let libc::cmsghdr {
177 cmsg_len,
178 cmsg_level,
179 cmsg_type,
180 } = *cmsg_ptr;
181
182 match (cmsg_level, cmsg_type) {
183 (libc::SOL_NETLINK, libc::NETLINK_PKTINFO) => {
184 let data = std::slice::from_raw_parts(
185 libc::CMSG_DATA(cmsg_ptr),
186 cmsg_len - libc::CMSG_LEN(0) as usize,
187 );
188 *last_group = Some(utils::parse_u32(&data[..4]).unwrap());
189 }
190 _ => {}
191 }
192
193 cmsg_ptr = libc::CMSG_NXTHDR(msghdr_ptr, cmsg_ptr);
194 }
195 }
196
197 return Ok(read as usize);
198 }
199 }
200}