mqrs: Implement send and recv of sysvmq [wip]

TBD: CHANGELOG.md and adapt to recent changes is sysvmq

src/main.rs

@@ -1,5 +1,7 @@
 use anyhow::Result;
-use clap::{ArgAction, Parser};
+use clap::Parser;
+use log::Level;
+use std::env;
 
 mod posix;
 mod sysv;
@@ -30,6 +32,8 @@ enum SysvCommand {
     Info(sysv::Info),
     List(sysv::List),
     Unlink(sysv::Unlink),
+    Send(sysv::Send),
+    Recv(sysv::Recv),
 }
 
 #[derive(Debug, Parser)]
@@ -39,9 +43,9 @@ enum SysvCommand {
     subcommand_required = true
 )]
 struct Opts {
-    /// Produce verbose output, multiple -v options increase the verbosity (max. 3)
-    #[clap(short, long, global = true, action = ArgAction::Count)]
-    verbose: u32,
+    /// Set a log level
+    #[arg(short, long, value_name = "LEVEL", default_value_t = Level::Info)]
+    pub log_level: Level,
     /// Backend to be used
     #[clap(subcommand)]
     backend: Backend,
@@ -50,15 +54,11 @@ struct Opts {
 fn main() -> Result<()> {
     let opts: Opts = Opts::parse();
 
-    env_logger::Builder::from_env(env_logger::Env::default().default_filter_or(
-        match opts.verbose {
-            0 => "warn",
-            1 => "info",
-            2 => "debug",
-            _ => "trace",
-        },
-    ))
-    .init();
+    if env::var("RUST_LOG").is_err() {
+        env::set_var("RUST_LOG", format!("{}", opts.log_level));
+    }
+
+    env_logger::init();
 
     match opts.backend {
         Backend::Posix(p) => match p {
@@ -74,6 +74,8 @@ fn main() -> Result<()> {
             SysvCommand::Info(i) => i.run()?,
             SysvCommand::List(l) => l.run()?,
             SysvCommand::Unlink(u) => u.run()?,
+            SysvCommand::Send(s) => s.run()?,
+            SysvCommand::Recv(r) => r.run()?,
         },
     }
 

src/sysv/mod.rs

@@ -1,9 +1,13 @@
 mod create;
 mod info;
 mod list;
+mod recv;
+mod send;
 mod unlink;
 
 pub use create::Create;
 pub use info::Info;
 pub use list::List;
+pub use recv::Recv;
+pub use send::Send;
 pub use unlink::Unlink;

src/sysv/recv.rs

@@ -0,0 +1,42 @@
+use anyhow::Result;
+// use chrono::DateTime;
+use clap::Parser;
+// use humantime::Duration;
+use log::info;
+use sysvmq::SysvMq;
+
+/// Send a message to a message queue
+#[derive(Debug, Parser)]
+pub struct Recv {
+    // /// Set a different priority, priority >= 0
+    // #[clap(short, long, default_value = "0")]
+    // priority: u32,
+    // /// Do not block
+    // #[clap(short, long)]
+    // non_blocking: bool,
+    // /// Timeout, example "5h 23min 42ms"
+    // #[clap(short = 'o', long, conflicts_with = "deadline")]
+    // timeout: Option<String>,
+    // /// Deadline until messages are sent (format: "%Y-%m-%d %H:%M:%S")
+    // #[clap(short, long, conflicts_with = "timeout")]
+    // deadline: Option<String>,
+    // /// Name of the queue
+    // #[clap(value_name = "QUEUE")]
+    // queue: String,
+    /// Key of of the queue to write to
+    #[clap(value_name = "Key")]
+    key: i32,
+    // /// Message to be sent to the queue
+    // #[clap(value_name = "MESSAGE")]
+    // msg: String,
+}
+
+impl Recv {
+    pub fn run(&self) -> Result<()> {
+        let mq = SysvMq::<String>::new();
+
+        mq.open(self.key)?.recv();
+
+        Ok(())
+    }
+}

src/sysv/send.rs

@@ -0,0 +1,42 @@
+use anyhow::Result;
+// use chrono::DateTime;
+use clap::Parser;
+// use humantime::Duration;
+use log::info;
+use sysvmq::SysvMq;
+
+/// Send a message to a message queue
+#[derive(Debug, Parser)]
+pub struct Send {
+    // /// Set a different priority, priority >= 0
+    // #[clap(short, long, default_value = "0")]
+    // priority: u32,
+    // /// Do not block
+    // #[clap(short, long)]
+    // non_blocking: bool,
+    // /// Timeout, example "5h 23min 42ms"
+    // #[clap(short = 'o', long, conflicts_with = "deadline")]
+    // timeout: Option<String>,
+    // /// Deadline until messages are sent (format: "%Y-%m-%d %H:%M:%S")
+    // #[clap(short, long, conflicts_with = "timeout")]
+    // deadline: Option<String>,
+    // /// Name of the queue
+    // #[clap(value_name = "QUEUE")]
+    // queue: String,
+    /// Key of of the queue to write to
+    #[clap(value_name = "Key")]
+    key: i32,
+    /// Message to be sent to the queue
+    #[clap(value_name = "MESSAGE")]
+    msg: String,
+}
+
+impl Send {
+    pub fn run(&self) -> Result<()> {
+        let mq = SysvMq::<String>::new();
+
+        mq.open(self.key)?.send(self.msg.as_bytes());
+
+        Ok(())
+    }
+}

sysvmq/CHANGELOG.md

@@ -8,8 +8,19 @@ adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
 ## [Unreleased]
 
+### Added
+
+- A basic test to test the happy path of a queue.
+- The function `SysvMq::send()` to send messages to a queue.
+- The function `SysvMq::recv()` to receive messages from a queue.
+- The function `SysvMq::destroy()` to destroy a queue.
+
 ### Changed
 
+- Remove `PhantomData` from the `SysvMq` struct.
+- The function `SysvMq::create()` is replaced by `new()`.
+- Remove generic type for `SysvMq`.
+- Rename `unlink_id()` to `destroy()`.
 - Update to the latest verscion of `nix`.
 - Fix several clippy findings in preperation to enable several lint
   groups.

sysvmq/src/lib.rs

@@ -1,8 +1,27 @@
 use libc::{
-    msgctl, msgget, msqid_ds, IPC_CREAT, IPC_INFO, IPC_NOWAIT, IPC_RMID, MSG_INFO,g MSG_STAT,
+    c_void,
+    msgctl,
+    msgget,
+    msgrcv,
+    msgsnd,
+    // msginfo,
+    msqid_ds,
+    IPC_CREAT,
+    // IPC_EXCL,
+    IPC_INFO,
+    IPC_NOWAIT,
+    // IPC_PRIVATE,
+    IPC_RMID,
+    // IPC_SET,
+    // IPC_STAT,
+    // MSG_COPY,
+    // MSG_EXCEPT,
+    MSG_INFO,
+    // MSG_NOERROR,
+    MSG_STAT,
 };
 use nix::errno::{errno, Errno};
-use std::{marker::PhantomData, mem::MaybeUninit, ptr};
+use std::{mem::MaybeUninit, ptr};
 use thiserror::Error;
 
 #[derive(Debug, Error)]
@@ -11,14 +30,57 @@ pub enum SysvMqError {
     ErrnoError(&'static str),
 }
 
+// /// IPC bit flags
+// #[repr(i32)]
+// enum Flags {
+//     /// Create key if key does not exist.
+//     CreateKey = IPC_CREAT,
+//     // /// Fail if key exists.
+//     // Exclusive = IPC_EXCL,
+//     /// Return error on wait.
+//     NoWait = IPC_NOWAIT,
+//     // /// No error if message is too big.
+//     // NoError = MSG_NOERROR,
+//     // /// Receive any message except of specified type.
+//     // Except = MSG_EXCEPT,
+//     // /// Copy (not remove) all queue messages.
+//     // Copy = MSG_COPY,
+//     // /// Private key (Special key value).
+//     // Private = IPC_PRIVATE,
+// }
+
+// /// Commands for `msgctl()`
+// #[repr(i32)]
+// enum ControlCommands {
+//     /// Remove identifier (Control command for `msgctl`, `semctl`, and `shmctl`).
+//     Remove = IPC_RMID,
+//     // /// Set `ipc_perm` options (Control command for `msgctl`, `semctl`, and `shmctl`).
+//     // SetPerm = IPC_SET,
+//     // /// Get `ipc_perm` options (Control command for `msgctl`, `semctl`, and `shmctl`).
+//     // GetPerm = IPC_STAT,
+//     /// See ipcs (Control command for `msgctl`, `semctl`, and `shmctl`).
+//     IpcInfo = IPC_INFO,
+//     /// IPCS control command.
+//     Stat = MSG_STAT,
+//     /// IPCS control command.
+//     MsgInfo = MSG_INFO,
+// }
+
+fn msgctl_wrapper(msqid: i32, cmd: i32, buf: *mut msqid_ds) -> Result<i32, SysvMqError> {
+    match unsafe { msgctl(msqid, cmd, buf) } {
+        -1 => Err(SysvMqError::ErrnoError(Errno::from_i32(errno()).desc())),
+        result => Ok(result),
+    }
+}
+
 #[allow(clippy::doc_markdown)]
-/// Unlink (delete) an existing SysV IPC message queue.
+/// Destroy an SysV IPC message queue by id.
 ///
 /// # Errors
 ///
 /// Return an `SysvMqError` when no queue with the given key can be
 /// found.
-pub fn unlink_id(id: i32) -> Result<(), SysvMqError> {
+pub fn destroy(id: i32) -> Result<(), SysvMqError> {
     let res = unsafe { msgctl(id, IPC_RMID, ptr::null::<msqid_ds>().cast_mut()) };
 
     match res {
@@ -83,28 +145,29 @@ pub fn msg_info(id: i32) {
     println!("info: {msg_info:?}");
 }
 
-pub struct SysvMq<T> {
-    pub id: i32,
-    pub key: i32,
+pub struct SysvMq {
+    id: i32,
+    key: i32,
     message_mask: i32,
     mode: i32,
-    types: PhantomData<T>,
 }
 
-impl<T> SysvMq<T> {
+impl SysvMq {
     /// Create a new message queye with the given key.
     ///
     /// # Errors
     ///
     /// Return an `SysvMqError` when no queue with the given key can be
     /// created.
-    pub fn create(&mut self, key: i32) -> Result<&Self, SysvMqError> {
-        self.key = key;
-        self.id = unsafe { msgget(self.key, IPC_CREAT | self.mode) };
+    pub fn new(key: i32) -> Result<Self, SysvMqError> {
+        let mut mq = Self::default();
 
-        match self.id {
+        mq.key = key;
+        mq.id = unsafe { msgget(mq.key, IPC_CREAT | mq.mode) };
+
+        match mq.id {
             -1 => Err(SysvMqError::ErrnoError(Errno::from_i32(errno()).desc())),
-            _ => Ok(self),
+            _ => Ok(mq),
         }
     }
 
@@ -124,30 +187,62 @@ impl<T> SysvMq<T> {
         }
     }
 
-    pub fn mode(&mut self, mode: i32) -> &Self {
+    /// Set the mode
+    pub fn mode(&mut self, mode: i32) {
         self.mode = mode;
-        self
     }
 
-    pub fn nonblocking(&mut self) -> &Self {
+    pub fn nonblocking(&mut self) {
         self.message_mask |= IPC_NOWAIT;
-        self
     }
 
-    #[must_use]
-    pub const fn new() -> Self {
+    pub fn send(&mut self, msg: &[u8]) -> Result<(), SysvMqError> {
+        match unsafe {
+            msgsnd(
+                self.id,
+                msg.as_ptr().cast::<c_void>(),
+                msg.len(),
+                self.message_mask,
+            )
+        } {
+            -1 => Err(SysvMqError::ErrnoError(Errno::from_i32(errno()).desc())),
+            _ => Ok(()),
+        }
+    }
+
+    pub fn recv(&mut self, msg: &mut [u8]) -> Result<(), SysvMqError> {
+        match unsafe {
+            msgrcv(
+                self.id,
+                msg.as_mut_ptr().cast::<c_void>(),
+                msg.len(),
+                0,
+                self.message_mask,
+            )
+        } {
+            -1 => Err(SysvMqError::ErrnoError(Errno::from_i32(errno()).desc())),
+            _ => Ok(()),
+        }
+    }
+
+    /// Destroy an existing SysV IPC message queue.
+    ///
+    /// # Errors
+    ///
+    /// Return an `SysvMqError` when no queue with the given key can be
+    /// found.
+    pub fn destroy(&mut self) -> Result<(), SysvMqError> {
+        destroy(self.id)
+    }
+}
+
+impl Default for SysvMq {
+    fn default() -> Self {
         Self {
             id: -1,
             key: 0,
             message_mask: 0,
             mode: 0o644,
-            types: PhantomData,
         }
     }
 }
-
-impl<T> Default for SysvMq<T> {
-    fn default() -> Self {
-        Self::new()
-    }
-}

sysvmq/tests/basic_tests.rs

@@ -0,0 +1,16 @@
+use sysvmq::SysvMq;
+
+#[test]
+fn new_send_recv_delete() {
+    let mut sysvmq = SysvMq::new(0).expect("could not create SysV message queue with key 0");
+    let msg = b"this is a test";
+    let mut buf = [0u8; 14];
+    sysvmq
+        .send(msg)
+        .expect("could not send message to SysV message queue");
+    sysvmq.recv(&mut buf);
+    sysvmq
+        .destroy()
+        .expect("could not destroy SysV message queue");
+    assert_eq!(msg, &buf);
+}