rmqtt_utils/
lib.rs

1//! Utilities module providing essential types and functions for common system operations
2//!
3//! ## Core Features:
4//! - **Byte Size Handling**: Human-readable byte size parsing/formatting with [`Bytesize`]
5//! - **Duration Conversion**: String-to-Duration parsing supporting multiple time units
6//! - **Timestamp Utilities**: Precise timestamp handling with millisecond resolution
7//! - **Network Addressing**: Cluster node address parsing ([`NodeAddr`]) and socket address handling
8//! - **Counter Implementation**: Thread-safe counter with merge modes ([`Counter`])
9//! - **Rate Counter**: Lock-free per-second rate tracking ([`RateCounter`])
10//!
11//! ## Key Components:
12//! - `Bytesize`: Handles 2G512M-style conversions with serialization support
13//! - Time functions: `timestamp_secs()`, `format_timestamp_now()`, and datetime parsing
14//! - `NodeAddr`: Cluster node representation (ID@Address) with parser
15//! - Network address utilities with proper error handling
16//! - Custom serde helpers for duration and address types
17//!
18//! ## Usage Examples:
19//! ```rust
20//! use rmqtt_utils::{Bytesize, NodeAddr, to_bytesize, to_duration, format_timestamp_now};
21//!
22//! // Byte size parsing
23//! let size = Bytesize::try_from("2G512M").unwrap();
24//! assert_eq!(size.as_usize(), 2_684_354_560);
25//!
26//! // Duration conversion
27//! let duration = to_duration("1h30m15s");
28//! assert_eq!(duration.as_secs(), 5415);
29//!
30//! // Node address parsing
31//! let node: NodeAddr = "1@mqtt-node:1883".parse().unwrap();
32//! assert_eq!(node.id, 1);
33//!
34//! // Timestamp formatting
35//! let now = format_timestamp_now();
36//! assert!(now.contains("2026")); // Current year
37//! ```
38//!
39//! ## Safety Guarantees:
40//! - Zero `unsafe` code usage (enforced by `#![deny(unsafe_code)]`)
41//! - Comprehensive error handling for parsing operations
42//! - Platform-agnostic network address handling
43//! - Chrono-based timestamp calculations with proper timezone handling
44//!
45//! Overall usage example:
46//!
47//! ```
48//! use rmqtt_utils::{
49//!     Bytesize, NodeAddr,
50//!     to_bytesize, to_duration,
51//!     timestamp_secs, format_timestamp_now
52//! };
53//!
54//! // Parse byte size from string
55//! let size = Bytesize::try_from("2G512M");
56//!
57//! // Convert duration string
58//! let duration = to_duration("1h30m15s");
59//!
60//! // Parse node address
61//! let node: NodeAddr = "123@127.0.0.1:1883".parse().unwrap();
62//!
63//! // Get formatted timestamp
64//! let now = format_timestamp_now();
65//! ```
66
67#![deny(unsafe_code)]
68
69use std::fmt;
70use std::net::SocketAddr;
71use std::ops::{Deref, DerefMut};
72use std::str::FromStr;
73use std::time::Duration;
74
75use anyhow::{anyhow, Error};
76use bytestring::ByteString;
77use chrono::LocalResult;
78use serde::{
79    de::{self, Deserializer},
80    ser::Serializer,
81    Deserialize, Serialize,
82};
83
84mod counter;
85mod rate_counter;
86
87pub use counter::{Counter, StatsMergeMode};
88pub use rate_counter::RateCounter;
89
90/// Cluster node identifier type (64-bit unsigned integer)
91pub type NodeId = u64;
92
93/// Network address storage using efficient ByteString
94pub type Addr = ByteString;
95
96/// Timestamp representation in seconds since Unix epoch
97pub type Timestamp = i64;
98
99/// Timestamp representation in milliseconds since Unix epoch
100pub type TimestampMillis = i64;
101
102const BYTESIZE_K: usize = 1024;
103const BYTESIZE_M: usize = 1048576;
104const BYTESIZE_G: usize = 1073741824;
105
106/// Human-readable byte size representation with parsing/serialization support
107///
108/// # Example:
109/// ```
110/// use rmqtt_utils::Bytesize;
111///
112/// // Create from string
113/// let size = Bytesize::try_from("2G512M").unwrap();
114/// assert_eq!(size.as_usize(), 2_684_354_560);
115///
116/// // Create from integer
117/// let size = Bytesize::from(1024);
118/// assert_eq!(size.string(), "1K");
119/// ```
120#[derive(Clone, Copy, Default)]
121pub struct Bytesize(pub usize);
122
123impl Bytesize {
124    /// Convert to u32 (may truncate on 32-bit platforms)
125    ///
126    /// # Example:
127    /// ```
128    /// let size = rmqtt_utils::Bytesize(5000);
129    /// assert_eq!(size.as_u32(), 5000);
130    /// ```
131    #[inline]
132    pub fn as_u32(&self) -> u32 {
133        self.0 as u32
134    }
135
136    /// Convert to u64
137    ///
138    /// # Example:
139    /// ```
140    /// let size = rmqtt_utils::Bytesize(usize::MAX);
141    /// assert_eq!(size.as_u64(), usize::MAX as u64);
142    /// ```
143    #[inline]
144    pub fn as_u64(&self) -> u64 {
145        self.0 as u64
146    }
147
148    /// Get underlying usize value
149    ///
150    /// # Example:
151    /// ```
152    /// let size = rmqtt_utils::Bytesize(1024);
153    /// assert_eq!(size.as_usize(), 1024);
154    /// ```
155    #[inline]
156    pub fn as_usize(&self) -> usize {
157        self.0
158    }
159
160    /// Format bytesize to human-readable string
161    ///
162    /// # Example:
163    /// ```
164    /// let size = rmqtt_utils::Bytesize(3145728);
165    /// assert_eq!(size.string(), "3M");
166    ///
167    /// let mixed = rmqtt_utils::Bytesize(2148532224);
168    /// assert_eq!(mixed.string(), "2G1M");
169    /// ```
170    #[inline]
171    pub fn string(&self) -> String {
172        let mut v = self.0;
173        let mut res = String::new();
174
175        let g = v / BYTESIZE_G;
176        if g > 0 {
177            res.push_str(&format!("{g}G"));
178            v %= BYTESIZE_G;
179        }
180
181        let m = v / BYTESIZE_M;
182        if m > 0 {
183            res.push_str(&format!("{m}M"));
184            v %= BYTESIZE_M;
185        }
186
187        let k = v / BYTESIZE_K;
188        if k > 0 {
189            res.push_str(&format!("{k}K"));
190            v %= BYTESIZE_K;
191        }
192
193        if v > 0 {
194            res.push_str(&format!("{v}B"));
195        }
196
197        res
198    }
199}
200
201impl Deref for Bytesize {
202    type Target = usize;
203    fn deref(&self) -> &Self::Target {
204        &self.0
205    }
206}
207
208impl DerefMut for Bytesize {
209    fn deref_mut(&mut self) -> &mut Self::Target {
210        &mut self.0
211    }
212}
213
214impl From<usize> for Bytesize {
215    fn from(v: usize) -> Self {
216        Bytesize(v)
217    }
218}
219
220impl TryFrom<&str> for Bytesize {
221    type Error = ParseSizeError;
222    fn try_from(v: &str) -> Result<Self, Self::Error> {
223        let value = to_bytesize(v)?;
224        Ok(Bytesize(value))
225    }
226}
227
228impl FromStr for Bytesize {
229    type Err = ParseSizeError;
230    fn from_str(s: &str) -> Result<Self, Self::Err> {
231        Ok(Bytesize(to_bytesize(s)?))
232    }
233}
234
235impl fmt::Debug for Bytesize {
236    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
237        write!(f, "{}", self.string())?;
238        Ok(())
239    }
240}
241
242impl fmt::Display for Bytesize {
243    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
244        write!(f, "{}", self.string())
245    }
246}
247
248impl Serialize for Bytesize {
249    #[inline]
250    fn serialize<S>(&self, serializer: S) -> std::result::Result<S::Ok, S::Error>
251    where
252        S: Serializer,
253    {
254        serializer.serialize_str(&self.to_string())
255    }
256}
257
258impl<'de> Deserialize<'de> for Bytesize {
259    #[inline]
260    fn deserialize<D>(deserializer: D) -> std::result::Result<Self, D::Error>
261    where
262        D: Deserializer<'de>,
263    {
264        let v = to_bytesize(&String::deserialize(deserializer)?).map_err(de::Error::custom)?;
265        Ok(Bytesize(v))
266    }
267}
268
269/// Parse human-readable byte size string to usize
270///
271/// # Example:
272/// ```
273/// let bytes = rmqtt_utils::to_bytesize("2G512K");
274/// assert_eq!(bytes, Ok(2148007936));
275///
276/// let complex = rmqtt_utils::to_bytesize("1G500M256K1024B");
277/// assert_eq!(complex, Ok(1598292992));
278/// ```
279#[inline]
280pub fn to_bytesize(text: &str) -> Result<usize, ParseSizeError> {
281    let text = text.to_uppercase().replace("GB", "G").replace("MB", "M").replace("KB", "K");
282    text.split_inclusive(['G', 'M', 'K', 'B'])
283        .map(|x| {
284            let mut chars = x.chars();
285            let u = chars.nth_back(0).ok_or(ParseSizeError::InvalidFormat)?;
286            let num_str = chars.as_str();
287            let v =
288                num_str.parse::<usize>().map_err(|_| ParseSizeError::InvalidNumber(num_str.to_string()))?;
289            match u {
290                'B' => Ok(v),
291                'K' => Ok(v * BYTESIZE_K),
292                'M' => Ok(v * BYTESIZE_M),
293                'G' => Ok(v * BYTESIZE_G),
294                _ => Err(ParseSizeError::InvalidUnit(u)),
295            }
296        })
297        .sum()
298}
299
300/// Errors that can occur when parsing a byte size string.
301#[derive(Debug, Eq, PartialEq, Clone)]
302pub enum ParseSizeError {
303    InvalidFormat,
304    InvalidNumber(String),
305    InvalidUnit(char),
306}
307
308impl std::fmt::Display for ParseSizeError {
309    fn fmt(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result {
310        match self {
311            Self::InvalidFormat => write!(f, "invalid size format"),
312            Self::InvalidNumber(s) => write!(f, "invalid number: '{s}'"),
313            Self::InvalidUnit(c) => write!(f, "invalid unit: '{c}'"),
314        }
315    }
316}
317
318impl std::error::Error for ParseSizeError {}
319
320/// Deserialize Duration from human-readable string format
321#[inline]
322pub fn deserialize_duration<'de, D>(deserializer: D) -> std::result::Result<Duration, D::Error>
323where
324    D: Deserializer<'de>,
325{
326    let v = String::deserialize(deserializer)?;
327    Ok(to_duration(&v))
328}
329
330/// Deserialize optional Duration from string
331#[inline]
332pub fn deserialize_duration_option<'de, D>(deserializer: D) -> std::result::Result<Option<Duration>, D::Error>
333where
334    D: Deserializer<'de>,
335{
336    let v = String::deserialize(deserializer)?;
337    if v.is_empty() {
338        Ok(None)
339    } else {
340        Ok(Some(to_duration(&v)))
341    }
342}
343
344/// Convert human-readable duration string to Duration
345///
346/// # Supported units:
347/// - ms: milliseconds
348/// - s: seconds
349/// - m: minutes
350/// - h: hours
351/// - d: days
352/// - w: weeks
353/// - f: fortnight (2 weeks)
354///
355/// # Example:
356/// ```
357/// let duration = rmqtt_utils::to_duration("1h30m15s");
358/// assert_eq!(duration.as_secs(), 5415);
359///
360/// let complex = rmqtt_utils::to_duration("2w3d12h");
361/// assert_eq!(complex.as_secs(), 1512000);
362/// ```
363#[inline]
364pub fn to_duration(text: &str) -> Duration {
365    let text = text.to_lowercase().replace("ms", "Y");
366    let ms: u64 = text
367        .split_inclusive(['s', 'm', 'h', 'd', 'w', 'f', 'Y'])
368        .map(|x| {
369            let mut chars = x.chars();
370            let u = match chars.nth_back(0) {
371                None => return 0,
372                Some(u) => u,
373            };
374            let v = match chars.as_str().parse::<u64>() {
375                Err(_e) => return 0,
376                Ok(v) => v,
377            };
378            match u {
379                'Y' => v,
380                's' => v * 1000,
381                'm' => v * 60000,
382                'h' => v * 3600000,
383                'd' => v * 86400000,
384                'w' => v * 604800000,
385                'f' => v * 1209600000,
386                _ => 0,
387            }
388        })
389        .sum();
390    Duration::from_millis(ms)
391}
392
393/// Deserialize SocketAddr with error handling
394#[inline]
395pub fn deserialize_addr<'de, D>(deserializer: D) -> std::result::Result<SocketAddr, D::Error>
396where
397    D: Deserializer<'de>,
398{
399    let addr = String::deserialize(deserializer)?
400        .parse::<std::net::SocketAddr>()
401        .map_err(serde::de::Error::custom)?;
402    Ok(addr)
403}
404
405/// Deserialize optional SocketAddr with port handling
406#[inline]
407pub fn deserialize_addr_option<'de, D>(
408    deserializer: D,
409) -> std::result::Result<Option<std::net::SocketAddr>, D::Error>
410where
411    D: Deserializer<'de>,
412{
413    let addr = String::deserialize(deserializer).map(|mut addr| {
414        if !addr.contains(':') {
415            addr += ":0";
416        }
417        addr
418    })?;
419    let addr = addr.parse::<std::net::SocketAddr>().map_err(serde::de::Error::custom)?;
420    Ok(Some(addr))
421}
422
423/// Deserialize optional datetime from string
424#[inline]
425pub fn deserialize_datetime_option<'de, D>(deserializer: D) -> std::result::Result<Option<Duration>, D::Error>
426where
427    D: Deserializer<'de>,
428{
429    let t_str = String::deserialize(deserializer)?;
430    if t_str.is_empty() {
431        Ok(None)
432    } else {
433        let t = if let Ok(d) = timestamp_parse_from_str(&t_str, "%Y-%m-%d %H:%M:%S") {
434            Duration::from_secs(d as u64)
435        } else {
436            let d = t_str.parse::<u64>().map_err(serde::de::Error::custom)?;
437            Duration::from_secs(d)
438        };
439        Ok(Some(t))
440    }
441}
442
443/// Serialize optional datetime to string
444#[inline]
445pub fn serialize_datetime_option<S>(t: &Option<Duration>, s: S) -> std::result::Result<S::Ok, S::Error>
446where
447    S: Serializer,
448{
449    if let Some(t) = t {
450        t.as_secs().to_string().serialize(s)
451    } else {
452        "".serialize(s)
453    }
454}
455
456/// Internal datetime parsing helper
457#[inline]
458fn timestamp_parse_from_str(ts: &str, fmt: &str) -> anyhow::Result<i64> {
459    let ndt = chrono::NaiveDateTime::parse_from_str(ts, fmt)?;
460    let ndt = ndt.and_local_timezone(*chrono::Local::now().offset());
461    match ndt {
462        LocalResult::None => Err(anyhow::Error::msg("Impossible")),
463        LocalResult::Single(d) => Ok(d.timestamp()),
464        LocalResult::Ambiguous(d, _tz) => Ok(d.timestamp()),
465    }
466}
467
468/// Get current timestamp as Duration
469///
470/// # Example:
471/// ```
472/// let ts = rmqtt_utils::timestamp();
473/// assert!(ts.as_secs() > 0);
474/// ```
475#[inline]
476pub fn timestamp() -> Duration {
477    use std::time::{SystemTime, UNIX_EPOCH};
478    SystemTime::now().duration_since(UNIX_EPOCH).unwrap_or_else(|_| {
479        let now = chrono::Local::now();
480        Duration::new(now.timestamp() as u64, now.timestamp_subsec_nanos())
481    })
482}
483
484/// Get current timestamp in seconds
485///
486/// # Example:
487/// ```
488/// let ts = rmqtt_utils::timestamp_secs();
489/// assert!(ts > 0);
490/// ```
491#[inline]
492pub fn timestamp_secs() -> Timestamp {
493    use std::time::{SystemTime, UNIX_EPOCH};
494    SystemTime::now()
495        .duration_since(UNIX_EPOCH)
496        .map(|t| t.as_secs() as i64)
497        .unwrap_or_else(|_| chrono::Local::now().timestamp())
498}
499
500/// Get current timestamp in milliseconds
501///
502/// # Example:
503/// ```
504/// let ts = rmqtt_utils::timestamp_millis();
505/// assert!(ts > 0);
506/// ```
507#[inline]
508pub fn timestamp_millis() -> TimestampMillis {
509    use std::time::{SystemTime, UNIX_EPOCH};
510    SystemTime::now()
511        .duration_since(UNIX_EPOCH)
512        .map(|t| t.as_millis() as i64)
513        .unwrap_or_else(|_| chrono::Local::now().timestamp_millis())
514}
515
516/// Format timestamp (seconds) to human-readable string
517#[inline]
518pub fn format_timestamp(t: Timestamp) -> String {
519    if t <= 0 {
520        "".into()
521    } else {
522        use chrono::TimeZone;
523        if let chrono::LocalResult::Single(t) = chrono::Local.timestamp_opt(t, 0) {
524            t.format("%Y-%m-%d %H:%M:%S").to_string()
525        } else {
526            "".into()
527        }
528    }
529}
530
531/// Format current timestamp to string
532///
533/// # Example:
534/// ```
535/// let now = rmqtt_utils::format_timestamp_now();
536/// assert!(!now.is_empty());
537/// ```
538#[inline]
539pub fn format_timestamp_now() -> String {
540    format_timestamp(timestamp_secs())
541}
542
543/// Format millisecond timestamp to string
544#[inline]
545pub fn format_timestamp_millis(t: TimestampMillis) -> String {
546    if t <= 0 {
547        "".into()
548    } else {
549        use chrono::TimeZone;
550        if let chrono::LocalResult::Single(t) = chrono::Local.timestamp_millis_opt(t) {
551            t.format("%Y-%m-%d %H:%M:%S%.3f").to_string()
552        } else {
553            "".into()
554        }
555    }
556}
557
558/// Format current millisecond timestamp to string
559///
560/// # Example:
561/// ```
562/// let now = rmqtt_utils::format_timestamp_millis_now();
563/// assert!(!now.is_empty());
564/// ```
565#[inline]
566pub fn format_timestamp_millis_now() -> String {
567    format_timestamp_millis(timestamp_millis())
568}
569
570/// Cluster node address representation (ID@Address)
571///
572/// # Example:
573/// ```
574/// use rmqtt_utils::NodeAddr;
575///
576/// // Parse from string
577/// let node: NodeAddr = "123@mqtt.example.com:1883".parse().unwrap();
578/// assert_eq!(node.id, 123);
579/// assert_eq!(node.addr, "mqtt.example.com:1883");
580///
581/// // Direct construction
582/// let node = NodeAddr {
583///     id: 456,
584///     addr: rmqtt_utils::Addr::from("localhost:8883")
585/// };
586/// ```
587#[derive(Clone, Serialize)]
588pub struct NodeAddr {
589    /// Unique node identifier
590    pub id: NodeId,
591
592    /// Network address in host:port format
593    pub addr: Addr,
594}
595
596impl std::fmt::Debug for NodeAddr {
597    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
598        write!(f, "{}@{:?}", self.id, self.addr)
599    }
600}
601
602impl FromStr for NodeAddr {
603    type Err = Error;
604    fn from_str(s: &str) -> Result<Self, Self::Err> {
605        let parts: Vec<&str> = s.split('@').collect();
606        if parts.len() < 2 {
607            return Err(anyhow!(format!("NodeAddr format error, {}", s)));
608        }
609        let id = NodeId::from_str(parts[0])?;
610        let addr = Addr::from(parts[1]);
611        Ok(NodeAddr { id, addr })
612    }
613}
614
615impl<'de> de::Deserialize<'de> for NodeAddr {
616    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
617    where
618        D: de::Deserializer<'de>,
619    {
620        NodeAddr::from_str(&String::deserialize(deserializer)?).map_err(de::Error::custom)
621    }
622}
623
624/// Expand all environment variable placeholders in the form `${ENV:VAR_NAME}`
625/// within a string.
626///
627/// Each occurrence of `${ENV:VAR_NAME}` will be replaced with the value of the
628/// corresponding environment variable `VAR_NAME`.  
629/// If an environment variable is not set, it will be replaced with an empty string
630/// and a warning will be logged.
631///
632/// # Example
633///
634/// ```
635/// use std::env;
636/// env::set_var("MQTT_USER", "user");
637/// env::set_var("MQTT_PASS", "pass");
638///
639/// let p = rmqtt_utils::expand_env_vars("${env:MQTT_PASS}");
640/// assert_eq!(p, "pass");
641///
642/// let s = rmqtt_utils::expand_env_vars("mqtt://${ENV:MQTT_USER}:${ENV:MQTT_PASS}@localhost");
643/// assert_eq!(s, "mqtt://user:pass@localhost");
644/// ```
645#[inline]
646pub fn expand_env_vars(value: &str) -> String {
647    static ENV_PATTERN: once_cell::sync::Lazy<regex::Regex> = once_cell::sync::Lazy::new(|| {
648        regex::Regex::new(r"(?i)\$\{ENV:([A-Z0-9_]+)\}").expect("Invalid regex pattern")
649    });
650
651    ENV_PATTERN
652        .replace_all(value, |caps: &regex::Captures| {
653            let env_name = &caps[1];
654            std::env::var(env_name).unwrap_or_else(|_| {
655                log::warn!("environment variable `{env_name}` not set");
656                String::new()
657            })
658        })
659        .into_owned()
660}
661
662/// Deserializes a string with `${ENV:VAR}` placeholders expanded from environment variables.
663///
664/// Returns the expanded string. Unset environment variables log a warning
665/// and are replaced with an empty string.
666#[inline]
667pub fn deserialize_expand_env_vars<'de, D>(deserializer: D) -> Result<String, D::Error>
668where
669    D: Deserializer<'de>,
670{
671    let v = String::deserialize(deserializer)?;
672    Ok(expand_env_vars(&v))
673}
674
675/// Deserializes an optional string with `${ENV:VAR}` placeholders expanded.
676///
677/// Returns `None` if the resulting expanded string is empty.
678#[inline]
679pub fn deserialize_expand_env_vars_option<'de, D>(deserializer: D) -> Result<Option<String>, D::Error>
680where
681    D: Deserializer<'de>,
682{
683    String::deserialize(deserializer).map(|s| expand_env_vars(&s)).map(|s| {
684        if s.is_empty() {
685            None
686        } else {
687            Some(s)
688        }
689    })
690}
rmqtt_utils/lib.rs

rmqtt_utils/
lib.rs
