use super::{SocketEvents, SocketType, UnixMessage};
use crate::{message::Message as _, socket::stats};
use libc::msghdr;
use std::os::unix::io::{AsRawFd, RawFd};
impl UnixMessage for msghdr {
#[inline]
fn send<E: SocketEvents>(
fd: RawFd,
entries: &mut [Self],
events: &mut E,
stats: &stats::Sender,
) {
send(&fd, entries, events, stats)
}
#[inline]
fn recv<E: SocketEvents>(
fd: RawFd,
ty: SocketType,
entries: &mut [Self],
events: &mut E,
stats: &stats::Sender,
) {
recv(&fd, ty, entries, events, stats)
}
}
#[inline]
pub fn send<'a, Sock: AsRawFd, P: IntoIterator<Item = &'a mut msghdr>, E: SocketEvents>(
socket: &Sock,
packets: P,
events: &mut E,
stats: &stats::Sender,
) {
for packet in packets {
#[cfg(debug_assertions)]
let prev_msg_control_ptr = packet.msg_control;
#[cfg(any(target_os = "macos", target_os = "ios", test))]
let msg_control = {
let msg_control = packet.msg_control;
if packet.msg_controllen == 0 {
packet.msg_control = core::ptr::null_mut();
}
msg_control
};
let sockfd = socket.as_raw_fd();
let msg = packet;
let flags = Default::default();
let result = libc!(sendmsg(sockfd, msg, flags));
#[cfg(any(target_os = "macos", target_os = "ios", test))]
{
msg.msg_control = msg_control;
}
#[cfg(debug_assertions)]
{
assert_eq!(
prev_msg_control_ptr, msg.msg_control,
"msg_control pointer was modified by the OS"
);
}
stats.send().on_operation_result(&result, |_len| 1);
let cf = match result {
Ok(_) => events.on_complete(1),
Err(err) => events.on_error(err),
};
if cf.is_break() {
return;
}
}
}
#[inline]
pub fn recv<'a, Sock: AsRawFd, P: IntoIterator<Item = &'a mut msghdr>, E: SocketEvents>(
socket: &Sock,
socket_type: SocketType,
packets: P,
events: &mut E,
stats: &stats::Sender,
) {
let mut flags = match socket_type {
SocketType::Blocking => Default::default(),
SocketType::NonBlocking => libc::MSG_DONTWAIT,
};
for packet in packets {
#[cfg(debug_assertions)]
let prev_msg_control_ptr = packet.msg_control;
let sockfd = socket.as_raw_fd();
let msg = packet;
let result = libc!(recvmsg(sockfd, msg, flags));
#[cfg(debug_assertions)]
{
assert_eq!(
prev_msg_control_ptr, msg.msg_control,
"msg_control pointer was modified by the OS"
);
}
stats.recv().on_operation_result(&result, |_len| 1);
let cf = match result {
Ok(payload_len) => {
unsafe {
msg.set_payload_len(payload_len.min(u16::MAX as _).max(0) as _);
}
events.on_complete(1)
}
Err(err) => events.on_error(err),
};
if cf.is_break() {
return;
}
flags = libc::MSG_DONTWAIT;
}
}