From 1100ba1b132d1ded5387a4232b7c5e4bcbaeb791 Mon Sep 17 00:00:00 2001
From: finga <finga@onders.org>
Date: Sun, 20 Jun 2021 16:49:40 +0200
Subject: [PATCH] Support timeout and deadline parameters for send

Add support for timeout and deadline parameters when messages are sent
to a message queue. To keep code duplication small, the `utils.rs`
file contains a helper function to parse the `Duration` from the
`String` received from the timeout parameter. Add conflict rules for
the timeout and deadline parameters.
---
 src/main.rs  |  1 +
 src/recv.rs  | 17 +++++------------
 src/send.rs  | 38 ++++++++++++++++++++++++++++++++++----
 src/utils.rs | 16 ++++++++++++++++
 4 files changed, 56 insertions(+), 16 deletions(-)
 create mode 100644 src/utils.rs

diff --git a/src/main.rs b/src/main.rs
index e6bc64a..d85fa83 100644
--- a/src/main.rs
+++ b/src/main.rs
@@ -7,6 +7,7 @@ mod info;
 mod recv;
 mod send;
 mod unlink;
+mod utils;
 
 use cli::{Command, Opts};
 
diff --git a/src/recv.rs b/src/recv.rs
index 5452d85..356829d 100644
--- a/src/recv.rs
+++ b/src/recv.rs
@@ -1,8 +1,9 @@
+use crate::utils;
 use anyhow::Result;
 use chrono::{DateTime, Local};
 use clap::Clap;
 use posixmq::PosixMq;
-use std::{str, time::Duration};
+use std::str;
 
 /// Receive and print a message from a message queue
 #[derive(Clap, Debug)]
@@ -17,10 +18,10 @@ pub struct Recv {
     #[clap(short, long)]
     pub timestamp: bool,
     /// Timeout in "<timeout>[s]" (default) or "<timeout>ms"
-    #[clap(short = 'o', long)]
+    #[clap(short = 'o', long, conflicts_with = "deadline")]
     pub timeout: Option<String>,
     /// Deadline until messages are received (format: "%Y-%m-%d %H:%M:%S")
-    #[clap(short, long)]
+    #[clap(short, long, conflicts_with = "timeout")]
     pub deadline: Option<String>,
     #[clap(value_name = "QUEUE")]
     pub queue: String,
@@ -43,15 +44,7 @@ impl Recv {
         let mut buf = vec![0; mq.attributes()?.max_msg_len];
 
         if let Some(timeout) = &self.timeout {
-            let timeout = if timeout.ends_with("ms") {
-                Duration::from_millis(timeout[0..timeout.len() - 2].parse::<u64>()?)
-            } else if timeout.ends_with("s") {
-                Duration::from_secs(timeout[0..timeout.len() - 1].parse::<u64>()?)
-            } else {
-                Duration::from_secs(timeout.parse::<u64>()?)
-            };
-
-            let (prio, len) = mq.recv_timeout(&mut buf, timeout)?;
+            let (prio, len) = mq.recv_timeout(&mut buf, utils::parse_duration(timeout)?)?;
 
             print_message(verbose, prio, len, self.timestamp, str::from_utf8(&buf)?);
         } else if let Some(deadline) = &self.deadline {
diff --git a/src/send.rs b/src/send.rs
index f1f846d..3c310b8 100644
--- a/src/send.rs
+++ b/src/send.rs
@@ -1,4 +1,6 @@
+use crate::utils;
 use anyhow::Result;
+use chrono::DateTime;
 use clap::Clap;
 
 /// Send a message to a message queue
@@ -10,6 +12,12 @@ pub struct Send {
     /// Do not block
     #[clap(short, long)]
     pub non_blocking: bool,
+    /// Timeout in "<timeout>[s]" (default) or "<timeout>ms"
+    #[clap(short = 'o', long, conflicts_with = "deadline")]
+    pub timeout: Option<String>,
+    /// Deadline until messages are received (format: "%Y-%m-%d %H:%M:%S")
+    #[clap(short, long, conflicts_with = "timeout")]
+    pub deadline: Option<String>,
     /// Name of the queue
     #[clap(value_name = "QUEUE")]
     pub queue: String,
@@ -26,11 +34,33 @@ impl Send {
             mq.nonblocking();
         }
 
-        mq.open(&self.queue)?
-            .send(self.priority, &self.msg.as_bytes())?;
+        if let Some(timeout) = &self.timeout {
+            mq.open(&self.queue)?.send_timeout(
+                self.priority,
+                &self.msg.as_bytes(),
+                utils::parse_duration(timeout)?,
+            )?;
 
-        if verbose {
-            println!("Sent message: \"{}\" to queue: {}", &self.msg, &self.queue);
+            if verbose {
+                println!("Sent message: \"{}\" to queue: {}", &self.msg, &self.queue);
+            }
+        } else if let Some(deadline) = &self.deadline {
+            mq.open(&self.queue)?.send_deadline(
+                self.priority,
+                &self.msg.as_bytes(),
+                DateTime::parse_from_str(deadline, "%Y-%m-%d %H:%M:%S")?.into(),
+            )?;
+
+            if verbose {
+                println!("Sent message: \"{}\" to queue: {}", &self.msg, &self.queue);
+            }
+        } else {
+            mq.open(&self.queue)?
+                .send(self.priority, &self.msg.as_bytes())?;
+
+            if verbose {
+                println!("Sent message: \"{}\" to queue: {}", &self.msg, &self.queue);
+            }
         }
 
         Ok(())
diff --git a/src/utils.rs b/src/utils.rs
new file mode 100644
index 0000000..f810f1e
--- /dev/null
+++ b/src/utils.rs
@@ -0,0 +1,16 @@
+use anyhow::Result;
+use std::{str, time::Duration};
+
+pub fn parse_duration(timeout: &str) -> Result<Duration> {
+    if timeout.ends_with("ms") {
+        Ok(Duration::from_millis(
+            timeout[0..timeout.len() - 2].parse::<u64>()?,
+        ))
+    } else if timeout.ends_with("s") {
+        Ok(Duration::from_secs(
+            timeout[0..timeout.len() - 1].parse::<u64>()?,
+        ))
+    } else {
+        Ok(Duration::from_secs(timeout.parse::<u64>()?))
+    }
+}