zerodds_cli_common/
lib.rs

1// SPDX-License-Identifier: Apache-2.0
2// Copyright 2026 ZeroDDS Contributors
3
4//! `zerodds-cli-common` — interne Helfer für die ZeroDDS-CLI-Tools.
5//!
6//! Crate `zerodds-cli-common`. Safety classification: **COMFORT**.
7//! Reine Tooling-Helfer (CLI-frontend), keine Runtime-Pfade.
8//!
9//! Sammelt die wenigen pieces of boilerplate die alle 7 Tools
10//! (`zerodds-record`, `-bench`, `-monitor`, `-spy`, `-snitch`,
11//! `-pcap`, `-mq`) brauchen: SIGINT/SIGTERM-Hook, GUID-Prefix-
12//! Generation, Duration-Parsing mit `s/m/h`-Suffixen.
13//!
14//! **Nicht für externe Konsumenten** — keine Stabilitäts-Garantie,
15//! keine Crates.io-Publikation.
16
17#![warn(missing_docs)]
18#![allow(clippy::module_name_repetitions)]
19
20use std::sync::Arc;
21use std::sync::atomic::{AtomicBool, Ordering};
22use std::time::{Duration, SystemTime, UNIX_EPOCH};
23
24use zerodds_dcps::runtime::UserReaderConfig;
25use zerodds_qos::{DeadlineQosPolicy, DurabilityKind, LivelinessQosPolicy, OwnershipKind};
26use zerodds_rtps::wire_types::GuidPrefix;
27
28/// Erzeugt einen prozess-stabilen `GuidPrefix` aus PID + nanos +
29/// Tool-Marker-Byte.
30///
31/// `marker` (z.B. `0xFE` für record, `0xFD` für bench) landet im
32/// vorletzten Byte und macht die Prefixe pro Tool trennbar.
33#[must_use]
34pub fn stable_prefix(marker: u8) -> GuidPrefix {
35    let mut bytes = [0u8; 12];
36    let pid = std::process::id();
37    bytes[0..4].copy_from_slice(&pid.to_le_bytes());
38    let nanos = SystemTime::now()
39        .duration_since(UNIX_EPOCH)
40        .unwrap_or_default()
41        .subsec_nanos();
42    bytes[4..8].copy_from_slice(&nanos.to_le_bytes());
43    bytes[8] = marker;
44    GuidPrefix::from_bytes(bytes)
45}
46
47/// Berechnet die Participant-GUID (16 Byte: 12 prefix + 4 EntityId
48/// `00 00 00 C1` für `ENTITYID_PARTICIPANT`).
49#[must_use]
50pub fn participant_guid(prefix: GuidPrefix) -> [u8; 16] {
51    let mut g = [0u8; 16];
52    g[..12].copy_from_slice(&prefix.0);
53    g[12..15].copy_from_slice(&[0, 0, 0]);
54    g[15] = 0xC1;
55    g
56}
57
58/// Unix-Zeit in Nanosekunden (i64; -1 bei System-Clock-Failure).
59#[must_use]
60pub fn unix_ns_now() -> i64 {
61    let dur = SystemTime::now()
62        .duration_since(UNIX_EPOCH)
63        .unwrap_or_default();
64    let total = dur
65        .as_secs()
66        .saturating_mul(1_000_000_000)
67        .saturating_add(u64::from(dur.subsec_nanos()));
68    i64::try_from(total).unwrap_or(i64::MAX)
69}
70
71/// Fehler beim Parsen einer Duration-Spec wie `30s`.
72#[derive(Debug, Clone, PartialEq, Eq)]
73pub struct DurationParseError {
74    /// Eingabe die nicht parse-bar war.
75    pub input: String,
76}
77
78impl std::fmt::Display for DurationParseError {
79    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
80        write!(f, "invalid duration spec: {}", self.input)
81    }
82}
83
84impl std::error::Error for DurationParseError {}
85
86/// Parst `5`, `5s`, `2m`, `1h` zu einer `Duration`.
87///
88/// # Errors
89/// [`DurationParseError`] bei nicht-numerischem Prefix oder unbekannter Einheit.
90pub fn parse_duration(s: &str) -> Result<Duration, DurationParseError> {
91    let bad = || DurationParseError {
92        input: s.to_string(),
93    };
94    let (num, unit) = s
95        .find(|c: char| c.is_alphabetic())
96        .map_or((s, "s"), |idx| (&s[..idx], &s[idx..]));
97    let n: u64 = num.parse().map_err(|_| bad())?;
98    let secs = match unit {
99        "s" | "" => n,
100        "m" => n.checked_mul(60).ok_or_else(bad)?,
101        "h" => n.checked_mul(3600).ok_or_else(bad)?,
102        _ => return Err(bad()),
103    };
104    Ok(Duration::from_secs(secs))
105}
106
107/// Installiert einen SIGINT/SIGTERM-Handler der bei Receive das
108/// `stop`-Flag auf `true` setzt. Auf Windows ist das eine no-op
109/// (User stoppt mit Task-Kill oder `--duration`).
110pub fn install_signal_handler(stop: Arc<AtomicBool>) {
111    install_inner(stop);
112}
113
114#[cfg(unix)]
115fn install_inner(stop: Arc<AtomicBool>) {
116    use std::sync::Mutex;
117    static HOOK: Mutex<Option<Arc<AtomicBool>>> = Mutex::new(None);
118    if let Ok(mut g) = HOOK.lock() {
119        *g = Some(stop);
120    }
121    extern "C" fn handler(_: i32) {
122        if let Ok(g) = HOOK.lock() {
123            if let Some(s) = g.as_ref() {
124                s.store(true, Ordering::Relaxed);
125            }
126        }
127    }
128    // SAFETY: libc::signal nimmt einen C-ABI-Funktionspointer; `handler`
129    // ist `extern "C"` und passt auf die libc-Signatur.
130    unsafe {
131        libc::signal(libc::SIGINT, handler as usize);
132        libc::signal(libc::SIGTERM, handler as usize);
133    }
134}
135
136#[cfg(not(unix))]
137fn install_inner(_stop: Arc<AtomicBool>) {}
138
139/// Default `UserReaderConfig` für untyped/`zerodds::RawBytes`-Topics.
140#[must_use]
141pub fn raw_reader_config(topic: &str) -> UserReaderConfig {
142    raw_reader_config_typed(topic, RAW_BYTES_TYPE_NAME)
143}
144
145/// The opaque "raw bytes" type name a monitoring reader announces when it has
146/// no concrete type to follow.
147pub const RAW_BYTES_TYPE_NAME: &str = "zerodds::RawBytes";
148
149/// Like [`raw_reader_config`] but announces a caller-supplied `type_name`.
150///
151/// This is the building block for **type-following** monitoring tools
152/// (`zerodds-spy`, `zerodds-record`): DDS matches reader↔writer by type-name
153/// equality on BOTH ends, so a generic monitor that wants to see a typed topic
154/// (e.g. `cuas::Track`) must announce that writer's *actual* `type_name`, not a
155/// generic `RawBytes`. The payload is still consumed opaquely (the reader never
156/// decodes it), so one config shape works for every followed type.
157///
158/// `type_identifier` stays `None` → the match falls back to pure `type_name`
159/// comparison (DDS 1.4 §2.2.3 default path), which is exactly what we want: no
160/// TypeObject is required to attach.
161#[must_use]
162pub fn raw_reader_config_typed(topic: &str, type_name: &str) -> UserReaderConfig {
163    UserReaderConfig {
164        topic_name: topic.to_string(),
165        type_name: type_name.to_string(),
166        reliable: true,
167        durability: DurabilityKind::Volatile,
168        deadline: DeadlineQosPolicy::default(),
169        liveliness: LivelinessQosPolicy::default(),
170        ownership: OwnershipKind::Shared,
171        partition: Vec::new(),
172        user_data: Vec::new(),
173        topic_data: Vec::new(),
174        group_data: Vec::new(),
175        type_identifier: zerodds_types::TypeIdentifier::None,
176        type_consistency: zerodds_types::qos::TypeConsistencyEnforcement::default(),
177        data_representation_offer: None,
178    }
179}
180
181/// Tracks which `(topic, type_name)` publications a monitoring tool has already
182/// attached to, so it can attach a type-following raw reader to each *newly*
183/// discovered writer type (including late joiners) exactly once.
184///
185/// The dedup core ([`TypeFollower::newly_seen`]) is pure — feed it the
186/// discovered pairs and it returns only the ones not seen before — so it is
187/// unit-testable without a live runtime. [`TypeFollower::poll`] is the
188/// convenience that reads `DcpsRuntime::discovered_publication_topics()`.
189#[derive(Debug, Default)]
190pub struct TypeFollower {
191    /// Topic allow-list; empty ⇒ follow every topic.
192    topics: std::collections::HashSet<String>,
193    /// `(topic, type_name)` pairs already reported, so each attaches once.
194    seen: std::collections::HashSet<(String, String)>,
195}
196
197impl TypeFollower {
198    /// Follow only the given topics. An empty iterator follows *all* topics.
199    #[must_use]
200    pub fn new<I, S>(topics: I) -> Self
201    where
202        I: IntoIterator<Item = S>,
203        S: Into<String>,
204    {
205        Self {
206            topics: topics.into_iter().map(Into::into).collect(),
207            seen: std::collections::HashSet::new(),
208        }
209    }
210
211    /// True if this follower accepts samples for `topic`.
212    #[must_use]
213    pub fn wants_topic(&self, topic: &str) -> bool {
214        self.topics.is_empty() || self.topics.contains(topic)
215    }
216
217    /// Pure dedup core: given the currently discovered `(topic, type_name)`
218    /// pairs, return those matching the topic filter that have not been seen
219    /// before, marking them seen. Order-preserving; duplicates within the input
220    /// collapse to one.
221    pub fn newly_seen<I, A, B>(&mut self, discovered: I) -> Vec<(String, String)>
222    where
223        I: IntoIterator<Item = (A, B)>,
224        A: Into<String>,
225        B: Into<String>,
226    {
227        let mut fresh = Vec::new();
228        for (t, ty) in discovered {
229            let pair = (t.into(), ty.into());
230            if !self.wants_topic(&pair.0) {
231                continue;
232            }
233            if self.seen.insert(pair.clone()) {
234                fresh.push(pair);
235            }
236        }
237        fresh
238    }
239
240    /// Poll a runtime's discovered publications and return the newly-appeared
241    /// `(topic, type_name)` pairs to attach a follower reader to.
242    pub fn poll(&mut self, runtime: &zerodds_dcps::runtime::DcpsRuntime) -> Vec<(String, String)> {
243        self.newly_seen(runtime.discovered_publication_topics())
244    }
245}
246
247#[cfg(test)]
248#[allow(clippy::unwrap_used, clippy::expect_used, clippy::panic)]
249mod tests {
250    use super::*;
251
252    #[test]
253    fn raw_reader_config_defaults_to_rawbytes_type() {
254        let c = raw_reader_config("Track");
255        assert_eq!(c.topic_name, "Track");
256        assert_eq!(c.type_name, RAW_BYTES_TYPE_NAME);
257    }
258
259    #[test]
260    fn raw_reader_config_typed_carries_the_supplied_type_name() {
261        let c = raw_reader_config_typed("Track", "cuas::Track");
262        assert_eq!(c.topic_name, "Track");
263        assert_eq!(c.type_name, "cuas::Track");
264        // Same opaque shape as the RawBytes config (reliable, volatile, no TID).
265        assert!(c.reliable);
266        assert!(matches!(
267            c.type_identifier,
268            zerodds_types::TypeIdentifier::None
269        ));
270    }
271
272    #[test]
273    fn type_follower_reports_each_pair_once() {
274        let mut tf = TypeFollower::new(Vec::<String>::new()); // all topics
275        let fresh = tf.newly_seen([("Track", "cuas::Track"), ("Track", "cuas::Track")]);
276        assert_eq!(
277            fresh,
278            vec![("Track".to_string(), "cuas::Track".to_string())]
279        );
280        // Re-polling the same discovery yields nothing new.
281        assert!(tf.newly_seen([("Track", "cuas::Track")]).is_empty());
282    }
283
284    #[test]
285    fn type_follower_picks_up_late_joiner_types() {
286        let mut tf = TypeFollower::new(Vec::<String>::new());
287        assert_eq!(tf.newly_seen([("Track", "cuas::Track")]).len(), 1);
288        // A second writer with a DIFFERENT type on the same topic is new.
289        let fresh = tf.newly_seen([("Track", "cuas::Track"), ("Track", "cuas::TrackV2")]);
290        assert_eq!(
291            fresh,
292            vec![("Track".to_string(), "cuas::TrackV2".to_string())]
293        );
294    }
295
296    #[test]
297    fn type_follower_filters_by_topic_allowlist() {
298        let mut tf = TypeFollower::new(["Track"]);
299        assert!(tf.wants_topic("Track"));
300        assert!(!tf.wants_topic("Other"));
301        let fresh = tf.newly_seen([("Track", "cuas::Track"), ("Other", "x::Y")]);
302        assert_eq!(
303            fresh,
304            vec![("Track".to_string(), "cuas::Track".to_string())]
305        );
306    }
307
308    #[test]
309    fn parse_duration_seconds() {
310        assert_eq!(parse_duration("5").unwrap(), Duration::from_secs(5));
311        assert_eq!(parse_duration("5s").unwrap(), Duration::from_secs(5));
312    }
313
314    #[test]
315    fn parse_duration_minutes() {
316        assert_eq!(parse_duration("3m").unwrap(), Duration::from_secs(180));
317    }
318
319    #[test]
320    fn parse_duration_hours() {
321        assert_eq!(parse_duration("2h").unwrap(), Duration::from_secs(7200));
322    }
323
324    #[test]
325    fn parse_duration_rejects_garbage() {
326        assert!(parse_duration("3x").is_err());
327        assert!(parse_duration("abc").is_err());
328    }
329
330    #[test]
331    fn stable_prefix_carries_marker() {
332        let p = stable_prefix(0xAB);
333        assert_eq!(p.0[8], 0xAB);
334    }
335
336    #[test]
337    fn participant_guid_has_participant_eid() {
338        let prefix = stable_prefix(0x42);
339        let g = participant_guid(prefix);
340        assert_eq!(&g[..12], &prefix.0[..]);
341        assert_eq!(&g[12..], &[0, 0, 0, 0xC1]);
342    }
343
344    #[test]
345    fn unix_ns_now_is_positive() {
346        assert!(unix_ns_now() > 0);
347    }
348}
zerodds_cli_common/lib.rs

zerodds_cli_common/
lib.rs
