Struct PosixMq

pub struct PosixMq { /* private fields */ }

A descriptor for an open posix message queue.

Message queues can be sent to and / or received from depending on the options it was opened with.

The descriptor is closed when this struct is dropped.

See the documentation in the crate root for examples, portability notes and OS details.

Implementations

impl PosixMq

pub fn open<N: AsRef<[u8]> + ?Sized>(name: &N) -> Result<Self, Error>

Open an existing message queue in read-write mode.

See OpenOptions::open() for details and possible errors.

pub fn create<N: AsRef<[u8]> + ?Sized>(name: &N) -> Result<Self, Error>

Open a message queue in read-write mode, creating it if it doesn’t exists.

See OpenOptions::open() for details and possible errors.

Examples found in repository

examples/limits.rs (line 81)

fn main() {
    println!("{}:", std::env::consts::OS);

    let mq = posixmq::OpenOptions::readwrite()
        .create_new()
        .open("/default_capacities")
        .expect("Cannot create new posix message queue /default_capacities");
    posixmq::remove_queue("/default_capacities")
        .expect("Cannot delete posix message queue /default_capacities");

    let attrs = mq.attributes().expect("Cannot get attributes for queue");
    println!("default queue capacity:         {}", attrs.capacity);
    println!("default maximum message length: {}", attrs.max_msg_len);

    println!("max message priority:           {}", btry(0xff_ff_ff_ff, |priority| {
        mq.send(priority as u32, b"b")
            .map(|_| mq.recv(&mut vec![0; attrs.max_msg_len]).unwrap() )
            .is_ok()
    }));
    println!("allows empty messages:          {}", mq.send(0, b"").is_ok());
    drop(mq);

    println!("max queue capacity:             {}", btry(1_000_000, |capacity| {
        posixmq::OpenOptions::readwrite()
            .capacity(capacity)
            .max_msg_len(1)
            .create_new()
            .open("/max_capacity")
            .map(|_| posixmq::remove_queue("/max_capacity").unwrap() )
            .is_ok()
    }));

    println!("max message length:             {}", btry(1_000_000, |length| {
        posixmq::OpenOptions::readwrite()
            .max_msg_len(length)
            .capacity(1)
            .create_new()
            .open("/max_length")
            .map(|_| posixmq::remove_queue("/max_length").unwrap() )
            .is_ok()
    }));

    println!("max equal:                      {}", btry(1_000_000, |equal| {
        posixmq::OpenOptions::readwrite()
            .max_msg_len(equal)
            .capacity(equal)
            .create_new()
            .open("/max_equal")
            .map(|_| posixmq::remove_queue("/max_equal").unwrap() )
            .is_ok()
    }));

    println!("allows just \"/\":                {}", {
        let result = posixmq::PosixMq::create("/");
        let _ = posixmq::remove_queue("/");
        result.is_ok()
    });
    println!("enforces name rules:            {}", {
        let noslash = std::ffi::CStr::from_bytes_with_nul(b"noslash\0").unwrap();
        let result = posixmq::OpenOptions::readwrite().create().open_c(noslash);
        let _ = posixmq::remove_queue_c(noslash);
        result.is_err()
    });
}

pub fn send(&self, priority: u32, msg: &[u8]) -> Result<(), Error>

Add a message to the queue.

For maximum portability, avoid using priorities >= 32 or sending zero-length messages.

Errors

• Queue is full and opened in nonblocking mode (EAGAIN) => ErrorKind::WouldBlock
• Message is too big for the queue (EMSGSIZE) => ErrorKind::Other
• Message is zero-length and the OS doesn’t allow this (EMSGSIZE) => ErrorKind::Other
• Priority is too high (EINVAL) => ErrorKind::InvalidInput
• Queue is opened in read-only mode (EBADF) => ErrorKind::Other
• Possibly other => ErrorKind::Other

Examples found in repository

examples/send.rs (line 31)

fn main() {
    let args = args_os().skip(1).collect::<Vec<_>>();
    if args.len() % 2 == 0 {
        eprintln!("Usage: cargo run --example send /queue-name [priority message] ...");
        return;
    }

    // open the message queue
    let mq = posixmq::OpenOptions::writeonly()
        .create()
        .mode(0o666) // FIXME set umask to actually make it sendable for others
        .open(args[0].as_bytes())
        .expect("opening failed");

    if args.len() == 1 {
        // read from stdin
        for line in stdin().lock().lines() {
            let line = line.expect("input must be UTF-8, sorry");
            let line = line.trim_start();
            let num_end = line.find(' ').expect("no space in line");
            let priority = line[..num_end].parse::<u32>().expect("priority is not a number");
            let msg = line[num_end+1..].trim_start();
            mq.send(priority, msg.as_bytes()).expect("sending failed");
        }
    } else {
        // read from the command line
        for i in (1..args.len()).step_by(2) {
            let priority = args[i].to_str()
                .and_then(|s| s.parse::<u32>().ok() )
                .expect("priority is not a number");
            let msg = args[i+1].as_bytes();
            mq.send(priority, msg).expect("sending failed");
        }
    }
}

examples/sort.rs (line 35)

fn main() {
    let args = args_os()
        .skip(1)
        .map(|osstring| osstring.into_vec() )
        .collect::<Vec<Vec<u8>>>();
    if args.is_empty() {
        return; // nothing to do
    }

    let mut recv_buf = vec![0; args.iter().map(Vec::len).max().unwrap()];

    let name = CStr::from_bytes_with_nul(b"/sort\0").unwrap();

    // create queue with the necessary permissions and open it
    let mq = posixmq::OpenOptions::readwrite()
        .nonblocking() // use WouldBlock to detect that the queue is empty
        .create_new()
        .mode(0o000) // only affects future attempts at opening it
        .capacity(args.len())
        .max_msg_len(recv_buf.len())
        .open_c(name)
        .expect("opening queue failed");

    // write arguments to the queue
    for arg in args {
        mq.send(arg.len() as u32, &arg).expect("sending failed");
    }

    // read until the queue is empty
    let stdout = stdout();
    let mut stdout = stdout.lock();
    loop {
        match mq.recv(&mut recv_buf) {
            Err(ref e) if e.kind() == ErrorKind::WouldBlock => break,
            Err(e) => panic!("receiving failed: {}", e),
            Ok((_priority, len)) => {
                stdout.write_all(&recv_buf[..len])
                    .and_then(|()| stdout.write_all(b"\n") )
                    .expect("writing to stdout failed");
            }
        }
    }

    // clean up
    posixmq::remove_queue_c(name).expect("deleting queue failed")
}

examples/limits.rs (line 43)

fn main() {
    println!("{}:", std::env::consts::OS);

    let mq = posixmq::OpenOptions::readwrite()
        .create_new()
        .open("/default_capacities")
        .expect("Cannot create new posix message queue /default_capacities");
    posixmq::remove_queue("/default_capacities")
        .expect("Cannot delete posix message queue /default_capacities");

    let attrs = mq.attributes().expect("Cannot get attributes for queue");
    println!("default queue capacity:         {}", attrs.capacity);
    println!("default maximum message length: {}", attrs.max_msg_len);

    println!("max message priority:           {}", btry(0xff_ff_ff_ff, |priority| {
        mq.send(priority as u32, b"b")
            .map(|_| mq.recv(&mut vec![0; attrs.max_msg_len]).unwrap() )
            .is_ok()
    }));
    println!("allows empty messages:          {}", mq.send(0, b"").is_ok());
    drop(mq);

    println!("max queue capacity:             {}", btry(1_000_000, |capacity| {
        posixmq::OpenOptions::readwrite()
            .capacity(capacity)
            .max_msg_len(1)
            .create_new()
            .open("/max_capacity")
            .map(|_| posixmq::remove_queue("/max_capacity").unwrap() )
            .is_ok()
    }));

    println!("max message length:             {}", btry(1_000_000, |length| {
        posixmq::OpenOptions::readwrite()
            .max_msg_len(length)
            .capacity(1)
            .create_new()
            .open("/max_length")
            .map(|_| posixmq::remove_queue("/max_length").unwrap() )
            .is_ok()
    }));

    println!("max equal:                      {}", btry(1_000_000, |equal| {
        posixmq::OpenOptions::readwrite()
            .max_msg_len(equal)
            .capacity(equal)
            .create_new()
            .open("/max_equal")
            .map(|_| posixmq::remove_queue("/max_equal").unwrap() )
            .is_ok()
    }));

    println!("allows just \"/\":                {}", {
        let result = posixmq::PosixMq::create("/");
        let _ = posixmq::remove_queue("/");
        result.is_ok()
    });
    println!("enforces name rules:            {}", {
        let noslash = std::ffi::CStr::from_bytes_with_nul(b"noslash\0").unwrap();
        let result = posixmq::OpenOptions::readwrite().create().open_c(noslash);
        let _ = posixmq::remove_queue_c(noslash);
        result.is_err()
    });
}

examples/merge.rs (line 66)

fn main() {
    use std::env::args;
    use std::io::ErrorKind;

    use mio::{Poll, Events, Interest, Token};

    let mut queues = args().skip(1).collect::<Vec<_>>();
    let dst = queues.pop().expect("arguments required");
    
    // open source queues
    let mut src = Vec::new();
    for name in queues {
        match posixmq::OpenOptions::readonly().nonblocking().open(&name) {
            Ok(mq) => src.push((mq, name)),
            Err(e) => panic!("Cannot open {:?} for receiving: {}", name, e),
        }
    }

    // open destination queue
    let mut dst = match posixmq::OpenOptions::writeonly().nonblocking().create().open(&dst) {
        Ok(mq) => (mq, dst),
        Err(e) => panic!("Cannot open or create {:?} for sending: {}", dst, e),
    };

    let mut poll = Poll::new().expect("Cannot create selector");
    poll.registry()
        .register(&mut dst.0, Token(0), Interest::WRITABLE)
        .expect("registering destination failed");
    for (i, &mut(ref mut src, _)) in src.iter_mut().enumerate() {
        poll.registry()
            .register(src, Token(i+1), Interest::READABLE)
            .expect("registering a source failed");
    }

    let mut unsent = Vec::<(u32, Box<[u8]>, &str)>::new();
    let mut buf = [0; 8192];
    let mut events = Events::with_capacity(1024);
    loop {
        poll.poll(&mut events, None).expect("Cannot poll selector");

        // receive all available messages from queues that are ready
        for event in events.iter() {
            if event.token() == Token(0) {
                // dst; will try to send below even without this event
                continue;
            }

            let &(ref mq, ref name) = &src[event.token().0-1];
            loop {
                match mq.recv(&mut buf) {
                    Err(ref e) if e.kind() == ErrorKind::WouldBlock => break,
                    Err(e) => panic!("Error receiving from {}: {}", name, e),
                    Ok((priority, len)) => unsent.push((priority, Box::from(&buf[..len]), name)),
                }
            }
        }

        // send as many messages as possible
        while let Some(&(priority, ref msg, ref src)) = unsent.last() {
            if let Err(e) = dst.0.send(priority, msg) {
                if e.kind() == ErrorKind::WouldBlock {
                    break;
                }
                panic!("Error sending to {}: {}", dst.1, e);
            }
            println!("message of priority {} with {} bytes from {} sent to {}", 
                priority, msg.len(), src, dst.1
            );
            let _ = unsent.pop();
        }
    }
}

pub fn recv(&self, msgbuf: &mut [u8]) -> Result<(u32, usize), Error>

Take the message with the highest priority from the queue.

The buffer must be at least as big as the maximum message length.

Errors

• Queue is empty and opened in nonblocking mode (EAGAIN) => ErrorKind::WouldBlock
• The receive buffer is smaller than the queue’s maximum message size (EMSGSIZE) => ErrorKind::Other
• Queue is opened in write-only mode (EBADF) => ErrorKind::Other
• Possibly other => ErrorKind::Other

Examples found in repository

examples/sort.rs (line 42)

fn main() {
    let args = args_os()
        .skip(1)
        .map(|osstring| osstring.into_vec() )
        .collect::<Vec<Vec<u8>>>();
    if args.is_empty() {
        return; // nothing to do
    }

    let mut recv_buf = vec![0; args.iter().map(Vec::len).max().unwrap()];

    let name = CStr::from_bytes_with_nul(b"/sort\0").unwrap();

    // create queue with the necessary permissions and open it
    let mq = posixmq::OpenOptions::readwrite()
        .nonblocking() // use WouldBlock to detect that the queue is empty
        .create_new()
        .mode(0o000) // only affects future attempts at opening it
        .capacity(args.len())
        .max_msg_len(recv_buf.len())
        .open_c(name)
        .expect("opening queue failed");

    // write arguments to the queue
    for arg in args {
        mq.send(arg.len() as u32, &arg).expect("sending failed");
    }

    // read until the queue is empty
    let stdout = stdout();
    let mut stdout = stdout.lock();
    loop {
        match mq.recv(&mut recv_buf) {
            Err(ref e) if e.kind() == ErrorKind::WouldBlock => break,
            Err(e) => panic!("receiving failed: {}", e),
            Ok((_priority, len)) => {
                stdout.write_all(&recv_buf[..len])
                    .and_then(|()| stdout.write_all(b"\n") )
                    .expect("writing to stdout failed");
            }
        }
    }

    // clean up
    posixmq::remove_queue_c(name).expect("deleting queue failed")
}

examples/limits.rs (line 44)

fn main() {
    println!("{}:", std::env::consts::OS);

    let mq = posixmq::OpenOptions::readwrite()
        .create_new()
        .open("/default_capacities")
        .expect("Cannot create new posix message queue /default_capacities");
    posixmq::remove_queue("/default_capacities")
        .expect("Cannot delete posix message queue /default_capacities");

    let attrs = mq.attributes().expect("Cannot get attributes for queue");
    println!("default queue capacity:         {}", attrs.capacity);
    println!("default maximum message length: {}", attrs.max_msg_len);

    println!("max message priority:           {}", btry(0xff_ff_ff_ff, |priority| {
        mq.send(priority as u32, b"b")
            .map(|_| mq.recv(&mut vec![0; attrs.max_msg_len]).unwrap() )
            .is_ok()
    }));
    println!("allows empty messages:          {}", mq.send(0, b"").is_ok());
    drop(mq);

    println!("max queue capacity:             {}", btry(1_000_000, |capacity| {
        posixmq::OpenOptions::readwrite()
            .capacity(capacity)
            .max_msg_len(1)
            .create_new()
            .open("/max_capacity")
            .map(|_| posixmq::remove_queue("/max_capacity").unwrap() )
            .is_ok()
    }));

    println!("max message length:             {}", btry(1_000_000, |length| {
        posixmq::OpenOptions::readwrite()
            .max_msg_len(length)
            .capacity(1)
            .create_new()
            .open("/max_length")
            .map(|_| posixmq::remove_queue("/max_length").unwrap() )
            .is_ok()
    }));

    println!("max equal:                      {}", btry(1_000_000, |equal| {
        posixmq::OpenOptions::readwrite()
            .max_msg_len(equal)
            .capacity(equal)
            .create_new()
            .open("/max_equal")
            .map(|_| posixmq::remove_queue("/max_equal").unwrap() )
            .is_ok()
    }));

    println!("allows just \"/\":                {}", {
        let result = posixmq::PosixMq::create("/");
        let _ = posixmq::remove_queue("/");
        result.is_ok()
    });
    println!("enforces name rules:            {}", {
        let noslash = std::ffi::CStr::from_bytes_with_nul(b"noslash\0").unwrap();
        let result = posixmq::OpenOptions::readwrite().create().open_c(noslash);
        let _ = posixmq::remove_queue_c(noslash);
        result.is_err()
    });
}

examples/merge.rs (line 56)

fn main() {
    use std::env::args;
    use std::io::ErrorKind;

    use mio::{Poll, Events, Interest, Token};

    let mut queues = args().skip(1).collect::<Vec<_>>();
    let dst = queues.pop().expect("arguments required");
    
    // open source queues
    let mut src = Vec::new();
    for name in queues {
        match posixmq::OpenOptions::readonly().nonblocking().open(&name) {
            Ok(mq) => src.push((mq, name)),
            Err(e) => panic!("Cannot open {:?} for receiving: {}", name, e),
        }
    }

    // open destination queue
    let mut dst = match posixmq::OpenOptions::writeonly().nonblocking().create().open(&dst) {
        Ok(mq) => (mq, dst),
        Err(e) => panic!("Cannot open or create {:?} for sending: {}", dst, e),
    };

    let mut poll = Poll::new().expect("Cannot create selector");
    poll.registry()
        .register(&mut dst.0, Token(0), Interest::WRITABLE)
        .expect("registering destination failed");
    for (i, &mut(ref mut src, _)) in src.iter_mut().enumerate() {
        poll.registry()
            .register(src, Token(i+1), Interest::READABLE)
            .expect("registering a source failed");
    }

    let mut unsent = Vec::<(u32, Box<[u8]>, &str)>::new();
    let mut buf = [0; 8192];
    let mut events = Events::with_capacity(1024);
    loop {
        poll.poll(&mut events, None).expect("Cannot poll selector");

        // receive all available messages from queues that are ready
        for event in events.iter() {
            if event.token() == Token(0) {
                // dst; will try to send below even without this event
                continue;
            }

            let &(ref mq, ref name) = &src[event.token().0-1];
            loop {
                match mq.recv(&mut buf) {
                    Err(ref e) if e.kind() == ErrorKind::WouldBlock => break,
                    Err(e) => panic!("Error receiving from {}: {}", name, e),
                    Ok((priority, len)) => unsent.push((priority, Box::from(&buf[..len]), name)),
                }
            }
        }

        // send as many messages as possible
        while let Some(&(priority, ref msg, ref src)) = unsent.last() {
            if let Err(e) = dst.0.send(priority, msg) {
                if e.kind() == ErrorKind::WouldBlock {
                    break;
                }
                panic!("Error sending to {}: {}", dst.1, e);
            }
            println!("message of priority {} with {} bytes from {} sent to {}", 
                priority, msg.len(), src, dst.1
            );
            let _ = unsent.pop();
        }
    }
}

pub fn iter<'a>(&'a self) -> Iter<'a>

Returns an Iterator which calls recv() repeatedly with an appropriately sized buffer.

If the message queue is opened in non-blocking mode the iterator can be used to drain the queue. Otherwise it will block and never end.

pub fn send_timeout( &self, priority: u32, msg: &[u8], timeout: Duration, ) -> Result<(), Error>

Add a message to the queue or cancel if it’s still full after a given duration.

Returns immediately if opened in nonblocking mode, and the timeout has no effect.

For maximum portability, avoid using priorities >= 32 or sending zero-length messages.

Errors

• Timeout expired (ETIMEDOUT) => ErrorKind::TimedOut
• Message is too big for the queue (EMSGSIZE) => ErrorKind::Other
• OS doesn’t allow empty messages (EMSGSIZE) => ErrorKind::Other
• Priority is too high (EINVAL) => ErrorKind::InvalidInput
• Queue is full and opened in nonblocking mode (EAGAIN) => ErrorKind::WouldBlock
• Queue is opened in write-only mode (EBADF) => ErrorKind::Other
• Timeout is too long / not representable => ErrorKind::InvalidInput
• Possibly other => ErrorKind::Other

pub fn send_deadline( &self, priority: u32, msg: &[u8], deadline: SystemTime, ) -> Result<(), Error>

Add a message to the queue or cancel if the queue is still full at a certain point in time.

Returns immediately if opened in nonblocking mode, and the timeout has no effect.
The deadline is a SystemTime because the queues are intended for inter-process commonication, and Instant might be process-specific.

For maximum portability, avoid using priorities >= 32 or sending zero-length messages.

Errors

• Deadline reached (ETIMEDOUT) => ErrorKind::TimedOut
• Message is too big for the queue (EMSGSIZE) => ErrorKind::Other
• OS doesn’t allow empty messages (EMSGSIZE) => ErrorKind::Other
• Priority is too high (EINVAL) => ErrorKind::InvalidInput
• Queue is full and opened in nonblocking mode (EAGAIN) => ErrorKind::WouldBlock
• Queue is opened in write-only mode (EBADF) => ErrorKind::Other
• Possibly other => ErrorKind::Other

pub fn recv_timeout( &self, msgbuf: &mut [u8], timeout: Duration, ) -> Result<(u32, usize), Error>

Take the message with the highest priority from the queue or cancel if the queue still empty after a given duration.

Returns immediately if opened in nonblocking mode, and the timeout has no effect.

Errors

• Timeout expired (ETIMEDOUT) => ErrorKind::TimedOut
• The receive buffer is smaller than the queue’s maximum message size (EMSGSIZE) => ErrorKind::Other
• Queue is empty and opened in nonblocking mode (EAGAIN) => ErrorKind::WouldBlock
• Queue is opened in read-only mode (EBADF) => ErrorKind::Other
• Timeout is too long / not representable => ErrorKind::InvalidInput
• Possibly other => ErrorKind::Other

Examples found in repository

examples/receive.rs (line 35)

fn main() {
    let mut args = args_os().skip(1);
    let name = args.next().expect("argument required");
    let timeout = args.next().map(|arg| {
        let arg = arg.into_string().expect("invalid timeout");
        if arg == "nonblocking" || arg == "drain" || arg == "available" || arg == "empty" {
            None
        } else if arg.ends_with("ms") {
            Some(Duration::from_millis(arg[..arg.len()-2].parse().expect("invalid duration")))
        } else if arg.ends_with("s") {
            Some(Duration::from_secs(arg[..arg.len()-1].parse().expect("invalid duration")))
        } else {
            Some(Duration::from_secs(arg.parse().expect("invalid duration")))
        }
    });

    let mut opts = posixmq::OpenOptions::readonly();
    if timeout == Some(None) {
        opts = *opts.nonblocking();
    }
    let mq = opts.open(name.as_bytes()).expect("opening failed");

    if let Some(Some(timeout)) = timeout {
        let mut buf = vec![0; mq.attributes().unwrap_or_default().max_msg_len];
        while let Ok((priority, len)) = mq.recv_timeout(&mut buf, timeout) {
            print!("{:3}\t", priority);
            stdout().write_all(&buf[..len]).expect("writing to stdout failed");
            println!();
        }
    } else {
        for (priority, msg) in mq {
            print!("{:3}\t", priority);
            stdout().write_all(&msg).expect("writing to stdout failed");
            println!();
        }
    }
}

pub fn recv_deadline( &self, msgbuf: &mut [u8], deadline: SystemTime, ) -> Result<(u32, usize), Error>

Take the message with the highest priority from the queue or cancel if the queue is still empty at a point in time.

Returns immediately if opened in nonblocking mode, and the timeout has no effect.
The deadline is a SystemTime because the queues are intended for inter-process commonication, and Instant might be process-specific.

Errors

• Deadline reached (ETIMEDOUT) => ErrorKind::TimedOut
• The receive buffer is smaller than the queue’s maximum message size (EMSGSIZE) => ErrorKind::Other
• Queue is empty and opened in nonblocking mode (EAGAIN) => ErrorKind::WouldBlock
• Queue is opened in read-only mode (EBADF) => ErrorKind::Other
• Possibly other => ErrorKind::Other

pub fn attributes(&self) -> Result<Attributes, Error>

Get information about the state of the message queue.

Errors

Retrieving these attributes should only fail if the underlying descriptor has been closed or is not a message queue.

On operating systems where the descriptor is a pointer, such as on FreeBSD and Illumos, such bugs will enable undefined behavior and this call will dereference freed or uninitialized memory.
(That doesn’t make this function unsafe though - PosixMq::from_raw_mqd() and mq_close() are.)

While a send() or recv() ran in place of this call would also have failed immediately and therefore not blocked, The descriptor might have become used for another queue when a later send() or recv() is performed. The descriptor might then be in blocking mode.

Examples

let mq = posixmq::OpenOptions::readwrite()
    .create_new()
    .max_msg_len(100)
    .capacity(3)
    .open("/with_custom_capacity")
    .expect("create queue");
let attrs = mq.attributes().expect("get attributes for queue");
assert_eq!(attrs.max_msg_len, 100);
assert_eq!(attrs.capacity, 3);
assert_eq!(attrs.current_messages, 0);
assert!(!attrs.nonblocking);

Ignore the error:

(Will only happen with buggy code (incorrect usage of from_raw_fd() or similar)).

let attrs = bad.attributes().unwrap_or_default();
assert_eq!(attrs.max_msg_len, 0);
assert_eq!(attrs.capacity, 0);
assert_eq!(attrs.current_messages, 0);
assert!(!attrs.nonblocking);

Examples found in repository

examples/receive.rs (line 34)

fn main() {
    let mut args = args_os().skip(1);
    let name = args.next().expect("argument required");
    let timeout = args.next().map(|arg| {
        let arg = arg.into_string().expect("invalid timeout");
        if arg == "nonblocking" || arg == "drain" || arg == "available" || arg == "empty" {
            None
        } else if arg.ends_with("ms") {
            Some(Duration::from_millis(arg[..arg.len()-2].parse().expect("invalid duration")))
        } else if arg.ends_with("s") {
            Some(Duration::from_secs(arg[..arg.len()-1].parse().expect("invalid duration")))
        } else {
            Some(Duration::from_secs(arg.parse().expect("invalid duration")))
        }
    });

    let mut opts = posixmq::OpenOptions::readonly();
    if timeout == Some(None) {
        opts = *opts.nonblocking();
    }
    let mq = opts.open(name.as_bytes()).expect("opening failed");

    if let Some(Some(timeout)) = timeout {
        let mut buf = vec![0; mq.attributes().unwrap_or_default().max_msg_len];
        while let Ok((priority, len)) = mq.recv_timeout(&mut buf, timeout) {
            print!("{:3}\t", priority);
            stdout().write_all(&buf[..len]).expect("writing to stdout failed");
            println!();
        }
    } else {
        for (priority, msg) in mq {
            print!("{:3}\t", priority);
            stdout().write_all(&msg).expect("writing to stdout failed");
            println!();
        }
    }
}

examples/limits.rs (line 38)

fn main() {
    println!("{}:", std::env::consts::OS);

    let mq = posixmq::OpenOptions::readwrite()
        .create_new()
        .open("/default_capacities")
        .expect("Cannot create new posix message queue /default_capacities");
    posixmq::remove_queue("/default_capacities")
        .expect("Cannot delete posix message queue /default_capacities");

    let attrs = mq.attributes().expect("Cannot get attributes for queue");
    println!("default queue capacity:         {}", attrs.capacity);
    println!("default maximum message length: {}", attrs.max_msg_len);

    println!("max message priority:           {}", btry(0xff_ff_ff_ff, |priority| {
        mq.send(priority as u32, b"b")
            .map(|_| mq.recv(&mut vec![0; attrs.max_msg_len]).unwrap() )
            .is_ok()
    }));
    println!("allows empty messages:          {}", mq.send(0, b"").is_ok());
    drop(mq);

    println!("max queue capacity:             {}", btry(1_000_000, |capacity| {
        posixmq::OpenOptions::readwrite()
            .capacity(capacity)
            .max_msg_len(1)
            .create_new()
            .open("/max_capacity")
            .map(|_| posixmq::remove_queue("/max_capacity").unwrap() )
            .is_ok()
    }));

    println!("max message length:             {}", btry(1_000_000, |length| {
        posixmq::OpenOptions::readwrite()
            .max_msg_len(length)
            .capacity(1)
            .create_new()
            .open("/max_length")
            .map(|_| posixmq::remove_queue("/max_length").unwrap() )
            .is_ok()
    }));

    println!("max equal:                      {}", btry(1_000_000, |equal| {
        posixmq::OpenOptions::readwrite()
            .max_msg_len(equal)
            .capacity(equal)
            .create_new()
            .open("/max_equal")
            .map(|_| posixmq::remove_queue("/max_equal").unwrap() )
            .is_ok()
    }));

    println!("allows just \"/\":                {}", {
        let result = posixmq::PosixMq::create("/");
        let _ = posixmq::remove_queue("/");
        result.is_ok()
    });
    println!("enforces name rules:            {}", {
        let noslash = std::ffi::CStr::from_bytes_with_nul(b"noslash\0").unwrap();
        let result = posixmq::OpenOptions::readwrite().create().open_c(noslash);
        let _ = posixmq::remove_queue_c(noslash);
        result.is_err()
    });
}

pub fn is_nonblocking(&self) -> Result<bool, Error>

Check whether this descriptor is in nonblocking mode.

Errors

Should only fail as result of buggy code that either created this descriptor from something that is not a queue, or has already closed the underlying descriptor.
(This function will not silently succeed if the fd points to anything other than a queue (for example a socket), as this function is a wrapper around [attributes()][#method.attributes].)
To ignore failure, one can write .is_nonblocking().unwrap_or(false).

An error doesn’t guarantee that any further send() or recv() wont block.

While a send() or recv() ran in place of this call would also have failed immediately and therefore not blocked, the descriptor might have become used for another queue when a later send() or recv() is performed. The descriptor might then be in blocking mode.

pub fn set_nonblocking(&self, nonblocking: bool) -> Result<(), Error>

Enable or disable nonblocking mode for this descriptor.

This can also be set when opening the message queue, with OpenOptions::nonblocking().

Errors

Setting nonblocking mode should only fail due to incorrect usage of from_raw_fd() or as_raw_fd(), see the documentation on attributes() for details.

pub fn try_clone(&self) -> Result<Self, Error>

Create a new descriptor for the same message queue.

The new descriptor will have close-on-exec set.

This function is not available on FreeBSD, Illumos or Solaris.

pub fn is_cloexec(&self) -> Result<bool, Error>

Check whether this descriptor will be closed if the process execs into another program.

Posix message queues are closed on exec by default, but this can be changed with set_cloexec().

This function is not available on Illumos, Solaris or VxWorks.

Errors

Retrieving this flag should only fail if the descriptor is already closed.
In that case it will obviously not be open after execing, so treating errors as true should be safe.

Examples

let queue = posixmq::PosixMq::create("is_cloexec").expect("open queue");
assert!(queue.is_cloexec().unwrap_or(true));

pub fn set_cloexec(&self, cloexec: bool) -> Result<(), Error>

Change close-on-exec for this descriptor.

It is on by default, so this method should only be called when one wants the descriptor to remain open afte execing.

This function is not available on Illumos, Solaris or VxWorks.

Errors

This function should only fail if the underlying file descriptor has been closed (due to incorrect usage of from_raw_fd() or similar), and not reused for something else yet.

pub unsafe fn from_raw_mqd(mqd: mqd_t) -> Self

Create a PosixMq from an already opened message queue descriptor.

This function should only be used for ffi or if calling mq_open() directly for some reason.
Use from_raw_fd() instead if the surrounding code requires mqd_t to be a file descriptor.

Safety

On some operating systems mqd_t is a pointer, which means that the safety of most other methods depend on it being correct.

pub fn as_raw_mqd(&self) -> mqd_t

Get the raw message queue descriptor.

This function should only be used for passing to ffi code or to access portable features not exposed by this wrapper (such as calling mq_notify() or not automatically retrying on EINTR / ErrorKind::Interrupted when sending or receiving).

If you need a file descriptor, use as_raw_fd() instead for increased portability. (as_raw_fd() can sometimes retrieve an underlying file descriptor even if mqd_t is not an int.)

pub fn into_raw_mqd(self) -> mqd_t

Convert this wrapper into the raw message queue descriptor without closing it.

This function should only be used for ffi; If you need a file descriptor use into_raw_fd() instead.

Trait Implementations

impl AsRawFd for PosixMq

Get an underlying file descriptor for the message queue.

If you just need the raw mqd_t, use as_raw_mqd() instead for increased portability.

This impl is not available on Illumos, Solaris or VxWorks.

fn as_raw_fd(&self) -> RawFd

Extracts the raw file descriptor.

impl Debug for PosixMq

fn fmt(&self, fmtr: &mut Formatter<'_>) -> Result

Formats the value using the given formatter.

impl Drop for PosixMq

fn drop(&mut self)

Executes the destructor for this type.

impl Evented for PosixMq

Allow receiving event notifications through mio (version 0.6).

This impl requires the mio_06 feature to be enabled:

[dependencies]
posixmq = {version="1.0", features=["mio_06"]}

Remember to open the queue in non-blocking mode. (with OpenOptions.noblocking())

fn register( &self, poll: &Poll, token: Token, interest: Ready, opts: PollOpt, ) -> Result<(), Error>

Register self with the given Poll instance.

fn reregister( &self, poll: &Poll, token: Token, interest: Ready, opts: PollOpt, ) -> Result<(), Error>

Re-register self with the given Poll instance.

fn deregister(&self, poll: &Poll) -> Result<(), Error>

Deregister self from the given Poll instance

impl FromRawFd for PosixMq

Create a PosixMq wrapper from a raw file descriptor.

Note that the message queue will be closed when the returned PosixMq goes out of scope / is dropped.

This impl is not available on FreeBSD, Illumos or Solaris; If you got a mqd_t in a portable fashion (from FFI code or by calling mq_open() yourself for some reason), use from_raw_mqd() instead.

unsafe fn from_raw_fd(fd: RawFd) -> Self

Constructs a new instance of Self from the given raw file descriptor.

impl<'a> IntoIterator for &'a PosixMq

type Item = (u32, Vec<u8>)

The type of the elements being iterated over.

type IntoIter = Iter<'a>

Which kind of iterator are we turning this into?

fn into_iter(self) -> Iter<'a>

Creates an iterator from a value.

impl IntoIterator for PosixMq

type Item = (u32, Vec<u8>)

The type of the elements being iterated over.

type IntoIter = IntoIter

Which kind of iterator are we turning this into?

fn into_iter(self) -> IntoIter

Creates an iterator from a value.

impl IntoRawFd for PosixMq

Convert the PosixMq into a raw file descriptor without closing the message queue.

This impl is not available on FreeBSD, Illumos or Solaris. If you need to transfer ownership to FFI code accepting a mqd_t, use into_raw_mqd() instead.

fn into_raw_fd(self) -> RawFd

Consumes this object, returning the raw underlying file descriptor.

impl Source for &PosixMq

Allow receiving event notifications through mio (version 0.7).

This impl requires the mio_07 feature to be enabled:

[dependencies]
posixmq = {version="1.0", features=["mio_07"]}

Due to a long-lived bug in cargo this will currently enable the os_reactor feature of mio. This is not intended, and can change in the future.

You probably want to make the queue non-blocking: Either use OpenOptions.noblocking() when preparing to open the queue, or call set_nonblocking(true).

fn register( &mut self, registry: &Registry, token: Token, interest: Interest, ) -> Result<(), Error>

Register self with the given Registry instance.

fn reregister( &mut self, registry: &Registry, token: Token, interest: Interest, ) -> Result<(), Error>

Re-register self with the given Registry instance.

fn deregister(&mut self, registry: &Registry) -> Result<(), Error>

Deregister self from the given Registry instance.

impl Source for PosixMq

fn register( &mut self, registry: &Registry, token: Token, interest: Interest, ) -> Result<(), Error>

Register self with the given Registry instance.

fn reregister( &mut self, registry: &Registry, token: Token, interest: Interest, ) -> Result<(), Error>

Re-register self with the given Registry instance.

fn deregister(&mut self, registry: &Registry) -> Result<(), Error>

Deregister self from the given Registry instance.

impl Send for PosixMq