Files
arrayvec
async_attributes
async_channel
async_executor
async_global_executor
async_io
async_lock
async_mutex
async_process
async_rustbus
async_std
collections
fs
future
io
net
option
os
path
pin
result
rt
stream
string
sync
task
unit
vec
async_task
atomic_waker
bitflags
blocking
cache_padded
cfg_if
concurrent_queue
crossbeam_utils
ctor
derive_utils
event_listener
fastrand
find_crate
futures
futures_channel
futures_core
futures_enum
futures_executor
futures_io
futures_lite
futures_macro
futures_sink
futures_task
futures_util
async_await
future
io
lock
sink
stream
task
kv_log_macro
lazy_static
libc
log
memchr
nix
num_cpus
once_cell
parking
pin_project_lite
pin_utils
polling
proc_macro2
proc_macro_hack
proc_macro_nested
quote
rustable
rustbus
rustbus_derive
serde
signal_hook
signal_hook_registry
slab
socket2
syn
toml
unicode_xid
value_bag
void
waker_fn
xml
  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
//! Posix Message Queue functions
//!
//! [Further reading and details on the C API](http://man7.org/linux/man-pages/man7/mq_overview.7.html)

use Result;
use errno::Errno;

use libc::{self, c_char, c_long, mqd_t, size_t};
use std::ffi::CString;
use sys::stat::Mode;
use std::mem;

libc_bitflags!{
    pub struct MQ_OFlag: libc::c_int {
        O_RDONLY;
        O_WRONLY;
        O_RDWR;
        O_CREAT;
        O_EXCL;
        O_NONBLOCK;
        O_CLOEXEC;
    }
}

libc_bitflags!{
    pub struct FdFlag: libc::c_int {
        FD_CLOEXEC;
    }
}

#[repr(C)]
#[derive(Clone, Copy, Debug, Eq, Hash, PartialEq)]
pub struct MqAttr {
    mq_attr: libc::mq_attr,
}

impl MqAttr {
    pub fn new(mq_flags: c_long,
               mq_maxmsg: c_long,
               mq_msgsize: c_long,
               mq_curmsgs: c_long)
               -> 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() }
        }
    }

    pub fn flags(&self) -> c_long {
        self.mq_attr.mq_flags
    }
}


/// Open a message queue
///
/// See also [`mq_open(2)`](http://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(name: &CString,
               oflag: MQ_OFlag,
               mode: Mode,
               attr: Option<&MqAttr>)
               -> Result<mqd_t> {
    let res = match attr {
        Some(mq_attr) => unsafe {
            libc::mq_open(name.as_ptr(),
                          oflag.bits(),
                          mode.bits() as libc::c_int,
                          &mq_attr.mq_attr as *const libc::mq_attr)
        },
        None => unsafe { libc::mq_open(name.as_ptr(), oflag.bits()) },
    };
    Errno::result(res)
}

/// Remove a message queue
///
/// See also [`mq_unlink(2)`](http://pubs.opengroup.org/onlinepubs/9699919799/functions/mq_unlink.html)
pub fn mq_unlink(name: &CString) -> Result<()> {
    let res = unsafe { libc::mq_unlink(name.as_ptr()) };
    Errno::result(res).map(drop)
}

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

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

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

/// Get message queue attributes
///
/// See also [`mq_getattr(2)`](http://pubs.opengroup.org/onlinepubs/9699919799/functions/mq_getattr.html)
pub fn mq_getattr(mqd: mqd_t) -> Result<MqAttr> {
    let mut attr = mem::MaybeUninit::<libc::mq_attr>::uninit();
    let res = unsafe { libc::mq_getattr(mqd, 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](http://pubs.opengroup.org/onlinepubs/9699919799/functions/mq_setattr.html)
pub fn mq_setattr(mqd: mqd_t, newattr: &MqAttr) -> Result<MqAttr> {
    let mut attr = mem::MaybeUninit::<libc::mq_attr>::uninit();
    let res = unsafe {
        libc::mq_setattr(mqd, &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
pub fn mq_set_nonblock(mqd: mqd_t) -> Result<MqAttr> {
    let oldattr = mq_getattr(mqd)?;
    let newattr = MqAttr::new(c_long::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: mqd_t) -> 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)
}