1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
//! Posix Message Queue functions
//!
//! # Example
//!
// no_run because a kernel module may be required.
//! ```no_run
//! # use std::ffi::CString;
//! # use nix::mqueue::*;
//! use nix::sys::stat::Mode;
//!
//! const MSG_SIZE: mq_attr_member_t = 32;
//! let mq_name= "/a_nix_test_queue";
//!
//! let oflag0 = MQ_OFlag::O_CREAT | MQ_OFlag::O_WRONLY;
//! let mode = Mode::S_IWUSR | Mode::S_IRUSR | Mode::S_IRGRP | Mode::S_IROTH;
//! let mqd0 = mq_open(mq_name, oflag0, mode, None).unwrap();
//! let msg_to_send = b"msg_1";
//! mq_send(&mqd0, msg_to_send, 1).unwrap();
//!
//! let oflag1 = MQ_OFlag::O_CREAT | MQ_OFlag::O_RDONLY;
//! let mqd1 = mq_open(mq_name, oflag1, mode, None).unwrap();
//! let mut buf = [0u8; 32];
//! let mut prio = 0u32;
//! let len = mq_receive(&mqd1, &mut buf, &mut prio).unwrap();
//! assert_eq!(prio, 1);
//! assert_eq!(msg_to_send, &buf[0..len]);
//!
//! mq_close(mqd1).unwrap();
//! mq_close(mqd0).unwrap();
//! ```
//! [Further reading and details on the C API](https://man7.org/linux/man-pages/man7/mq_overview.7.html)

use crate::errno::Errno;
use crate::NixPath;
use crate::Result;

use crate::sys::stat::Mode;
use libc::{self, mqd_t, size_t};
use std::mem;
#[cfg(any(
    target_os = "linux",
    target_os = "netbsd",
    target_os = "dragonfly"
))]
use std::os::unix::io::{
    AsFd, AsRawFd, BorrowedFd, FromRawFd, IntoRawFd, RawFd,
};

libc_bitflags! {
    /// Used with [`mq_open`].
    pub struct MQ_OFlag: libc::c_int {
        /// Open the message queue for receiving messages.
        O_RDONLY;
        /// Open the queue for sending messages.
        O_WRONLY;
        /// Open the queue for both receiving and sending messages
        O_RDWR;
        /// Create a message queue.
        O_CREAT;
        /// If set along with `O_CREAT`, `mq_open` will fail if the message
        /// queue name exists.
        O_EXCL;
        /// `mq_send` and `mq_receive` should fail with `EAGAIN` rather than
        /// wait for resources that are not currently available.
        O_NONBLOCK;
        /// Set the close-on-exec flag for the message queue descriptor.
        O_CLOEXEC;
    }
}

/// A message-queue attribute, optionally used with [`mq_setattr`] and
/// [`mq_getattr`] and optionally [`mq_open`],
#[repr(C)]
#[derive(Clone, Copy, Debug, Eq, Hash, PartialEq)]
pub struct MqAttr {
    mq_attr: libc::mq_attr,
}

/// Identifies an open POSIX Message Queue
// A safer wrapper around libc::mqd_t, which is a pointer on some platforms
// Deliberately is not Clone to prevent use-after-close scenarios
#[repr(transparent)]
#[derive(Debug)]
#[allow(missing_copy_implementations)]
pub struct MqdT(mqd_t);

// x32 compatibility
// See https://sourceware.org/bugzilla/show_bug.cgi?id=21279
/// Size of a message queue attribute member
#[cfg(all(target_arch = "x86_64", target_pointer_width = "32"))]
pub type mq_attr_member_t = i64;
/// Size of a message queue attribute member
#[cfg(not(all(target_arch = "x86_64", target_pointer_width = "32")))]
pub type mq_attr_member_t = libc::c_long;

impl MqAttr {
    /// Create a new message queue attribute
    ///
    /// # Arguments
    ///
    /// - `mq_flags`:   Either `0` or `O_NONBLOCK`.
    /// - `mq_maxmsg`:  Maximum number of messages on the queue.
    /// - `mq_msgsize`: Maximum message size in bytes.
    /// - `mq_curmsgs`: Number of messages currently in the queue.
    pub fn new(
        mq_flags: mq_attr_member_t,
        mq_maxmsg: mq_attr_member_t,
        mq_msgsize: mq_attr_member_t,
        mq_curmsgs: mq_attr_member_t,
    ) -> MqAttr {
        let mut attr = mem::MaybeUninit::<libc::mq_attr>::uninit();
        unsafe {
            let p = attr.as_mut_ptr();
            (*p).mq_flags = mq_flags;
            (*p).mq_maxmsg = mq_maxmsg;
            (*p).mq_msgsize = mq_msgsize;
            (*p).mq_curmsgs = mq_curmsgs;
            MqAttr {
                mq_attr: attr.assume_init(),
            }
        }
    }

    /// The current flags, either `0` or `O_NONBLOCK`.
    pub const fn flags(&self) -> mq_attr_member_t {
        self.mq_attr.mq_flags
    }

    /// The max number of messages that can be held by the queue
    pub const fn maxmsg(&self) -> mq_attr_member_t {
        self.mq_attr.mq_maxmsg
    }

    /// The maximum size of each message (in bytes)
    pub const fn msgsize(&self) -> mq_attr_member_t {
        self.mq_attr.mq_msgsize
    }

    /// The number of messages currently held in the queue
    pub const fn curmsgs(&self) -> mq_attr_member_t {
        self.mq_attr.mq_curmsgs
    }
}

/// Open a message queue
///
/// See also [`mq_open(2)`](https://pubs.opengroup.org/onlinepubs/9699919799/functions/mq_open.html)
// The mode.bits() cast is only lossless on some OSes
#[allow(clippy::cast_lossless)]
pub fn mq_open<P>(
    name: &P,
    oflag: MQ_OFlag,
    mode: Mode,
    attr: Option<&MqAttr>,
) -> Result<MqdT>
where
    P: ?Sized + NixPath,
{
    let res = name.with_nix_path(|cstr| match attr {
        Some(mq_attr) => unsafe {
            libc::mq_open(
                cstr.as_ptr(),
                oflag.bits(),
                mode.bits() as libc::c_int,
                &mq_attr.mq_attr as *const libc::mq_attr,
            )
        },
        None => unsafe { libc::mq_open(cstr.as_ptr(), oflag.bits()) },
    })?;

    Errno::result(res).map(MqdT)
}

/// Remove a message queue
///
/// See also [`mq_unlink(2)`](https://pubs.opengroup.org/onlinepubs/9699919799/functions/mq_unlink.html)
pub fn mq_unlink<P>(name: &P) -> Result<()>
where
    P: ?Sized + NixPath,
{
    let res =
        name.with_nix_path(|cstr| unsafe { libc::mq_unlink(cstr.as_ptr()) })?;
    Errno::result(res).map(drop)
}

/// Close a message queue
///
/// See also [`mq_close(2)`](https://pubs.opengroup.org/onlinepubs/9699919799/functions/mq_close.html)
pub fn mq_close(mqdes: MqdT) -> Result<()> {
    let res = unsafe { libc::mq_close(mqdes.0) };
    Errno::result(res).map(drop)
}

/// Receive a message from a message queue
///
/// See also [`mq_receive(2)`](https://pubs.opengroup.org/onlinepubs/9699919799/functions/mq_receive.html)
pub fn mq_receive(
    mqdes: &MqdT,
    message: &mut [u8],
    msg_prio: &mut u32,
) -> Result<usize> {
    let len = message.len() as size_t;
    let res = unsafe {
        libc::mq_receive(
            mqdes.0,
            message.as_mut_ptr().cast(),
            len,
            msg_prio as *mut u32,
        )
    };
    Errno::result(res).map(|r| r as usize)
}

feature! {
    #![feature = "time"]
    use crate::sys::time::TimeSpec;
    /// Receive a message from a message queue with a timeout
    ///
    /// See also ['mq_timedreceive(2)'](https://pubs.opengroup.org/onlinepubs/9699919799/functions/mq_receive.html)
    pub fn mq_timedreceive(
        mqdes: &MqdT,
        message: &mut [u8],
        msg_prio: &mut u32,
        abstime: &TimeSpec,
    ) -> Result<usize> {
        let len = message.len() as size_t;
        let res = unsafe {
            libc::mq_timedreceive(
                mqdes.0,
                message.as_mut_ptr().cast(),
                len,
                msg_prio as *mut u32,
                abstime.as_ref(),
            )
        };
        Errno::result(res).map(|r| r as usize)
    }
}

/// Send a message to a message queue
///
/// See also [`mq_send(2)`](https://pubs.opengroup.org/onlinepubs/9699919799/functions/mq_send.html)
pub fn mq_send(mqdes: &MqdT, message: &[u8], msq_prio: u32) -> Result<()> {
    let res = unsafe {
        libc::mq_send(mqdes.0, message.as_ptr().cast(), message.len(), msq_prio)
    };
    Errno::result(res).map(drop)
}

/// Get message queue attributes
///
/// See also [`mq_getattr(2)`](https://pubs.opengroup.org/onlinepubs/9699919799/functions/mq_getattr.html)
pub fn mq_getattr(mqd: &MqdT) -> Result<MqAttr> {
    let mut attr = mem::MaybeUninit::<libc::mq_attr>::uninit();
    let res = unsafe { libc::mq_getattr(mqd.0, attr.as_mut_ptr()) };
    Errno::result(res).map(|_| unsafe {
        MqAttr {
            mq_attr: attr.assume_init(),
        }
    })
}

/// Set the attributes of the message queue. Only `O_NONBLOCK` can be set,
/// everything else will be ignored. Returns the old attributes.
///
/// It is recommend to use the `mq_set_nonblock()` and `mq_remove_nonblock()`
/// convenience functions as they are easier to use.
///
/// [Further reading](https://pubs.opengroup.org/onlinepubs/9699919799/functions/mq_setattr.html)
pub fn mq_setattr(mqd: &MqdT, newattr: &MqAttr) -> Result<MqAttr> {
    let mut attr = mem::MaybeUninit::<libc::mq_attr>::uninit();
    let res = unsafe {
        libc::mq_setattr(
            mqd.0,
            &newattr.mq_attr as *const libc::mq_attr,
            attr.as_mut_ptr(),
        )
    };
    Errno::result(res).map(|_| unsafe {
        MqAttr {
            mq_attr: attr.assume_init(),
        }
    })
}

/// Convenience function.
/// Sets the `O_NONBLOCK` attribute for a given message queue descriptor
/// Returns the old attributes
#[allow(clippy::useless_conversion)] // Not useless on all OSes
pub fn mq_set_nonblock(mqd: &MqdT) -> Result<MqAttr> {
    let oldattr = mq_getattr(mqd)?;
    let newattr = MqAttr::new(
        mq_attr_member_t::from(MQ_OFlag::O_NONBLOCK.bits()),
        oldattr.mq_attr.mq_maxmsg,
        oldattr.mq_attr.mq_msgsize,
        oldattr.mq_attr.mq_curmsgs,
    );
    mq_setattr(mqd, &newattr)
}

/// Convenience function.
/// Removes `O_NONBLOCK` attribute for a given message queue descriptor
/// Returns the old attributes
pub fn mq_remove_nonblock(mqd: &MqdT) -> Result<MqAttr> {
    let oldattr = mq_getattr(mqd)?;
    let newattr = MqAttr::new(
        0,
        oldattr.mq_attr.mq_maxmsg,
        oldattr.mq_attr.mq_msgsize,
        oldattr.mq_attr.mq_curmsgs,
    );
    mq_setattr(mqd, &newattr)
}

#[cfg(any(target_os = "linux", target_os = "netbsd", target_os = "dragonfly"))]
impl AsFd for MqdT {
    /// Borrow the underlying message queue descriptor.
    fn as_fd(&self) -> BorrowedFd {
        // SAFETY: [MqdT] will only contain a valid fd by construction.
        unsafe { BorrowedFd::borrow_raw(self.0) }
    }
}

#[cfg(any(target_os = "linux", target_os = "netbsd", target_os = "dragonfly"))]
impl AsRawFd for MqdT {
    /// Return the underlying message queue descriptor.
    ///
    /// Returned descriptor is a "shallow copy" of the descriptor, so it refers
    ///  to the same underlying kernel object as `self`.
    fn as_raw_fd(&self) -> RawFd {
        self.0
    }
}

#[cfg(any(target_os = "linux", target_os = "netbsd", target_os = "dragonfly"))]
impl FromRawFd for MqdT {
    /// Construct an [MqdT] from [RawFd].
    ///
    /// # Safety
    /// The `fd` given must be a valid and open file descriptor for a message
    ///  queue.
    unsafe fn from_raw_fd(fd: RawFd) -> MqdT {
        MqdT(fd)
    }
}

#[cfg(any(target_os = "linux", target_os = "netbsd", target_os = "dragonfly"))]
impl IntoRawFd for MqdT {
    /// Consume this [MqdT] and return a [RawFd].
    fn into_raw_fd(self) -> RawFd {
        self.0
    }
}