ruststream_pulsar/
topic.rs

1//! [`PulsarTopic`]: validated first-class topic addressing.
2//!
3//! A Pulsar topic name carries four independent meanings (persistence, tenant, namespace,
4//! topic); the client treats it as a plain string and defers errors to the broker. The
5//! newtype validates on construction instead, so a malformed name fails before any I/O.
6
7use crate::error::PulsarError;
8
9/// A validated Pulsar topic name.
10///
11/// # Examples
12///
13/// ```
14/// use ruststream_pulsar::PulsarTopic;
15///
16/// let topic = PulsarTopic::persistent("acme", "orders", "created");
17/// assert_eq!(topic.as_str(), "persistent://acme/orders/created");
18///
19/// let parsed = PulsarTopic::parse("persistent://acme/orders/created")?;
20/// assert_eq!(parsed, topic);
21/// # Ok::<(), ruststream_pulsar::PulsarError>(())
22/// ```
23#[derive(Debug, Clone, PartialEq, Eq, Hash)]
24#[must_use]
25pub struct PulsarTopic {
26    full: String,
27}
28
29fn valid_part(part: &str) -> bool {
30    !part.is_empty()
31        && part
32            .chars()
33            .all(|c| c.is_ascii_alphanumeric() || matches!(c, '-' | '_' | '.'))
34}
35
36impl PulsarTopic {
37    fn of(scheme: &str, tenant: &str, namespace: &str, topic: &str) -> Result<Self, PulsarError> {
38        for (label, part) in [
39            ("tenant", tenant),
40            ("namespace", namespace),
41            ("topic", topic),
42        ] {
43            if !valid_part(part) {
44                return Err(PulsarError::Invalid(format!(
45                    "{label} '{part}' must be non-empty and contain only alphanumerics, '-', '_', '.'"
46                )));
47            }
48        }
49        Ok(Self {
50            full: format!("{scheme}://{tenant}/{namespace}/{topic}"),
51        })
52    }
53
54    /// A persistent topic: `persistent://tenant/namespace/topic`.
55    ///
56    /// # Panics
57    ///
58    /// Panics when a component is empty or carries characters Pulsar rejects; use
59    /// [`parse`](Self::parse) for fallible construction from untrusted input.
60    pub fn persistent(tenant: &str, namespace: &str, topic: &str) -> Self {
61        Self::of("persistent", tenant, namespace, topic).expect("invalid pulsar topic component")
62    }
63
64    /// A non-persistent topic: `non-persistent://tenant/namespace/topic`.
65    ///
66    /// # Panics
67    ///
68    /// Panics when a component is empty or carries characters Pulsar rejects; use
69    /// [`parse`](Self::parse) for fallible construction from untrusted input.
70    pub fn non_persistent(tenant: &str, namespace: &str, topic: &str) -> Self {
71        Self::of("non-persistent", tenant, namespace, topic)
72            .expect("invalid pulsar topic component")
73    }
74
75    /// Parses a fully qualified name (`persistent://t/ns/topic`), a `tenant/namespace/topic`
76    /// triple (defaulting to persistent), or a bare topic name (defaulting to
77    /// `persistent://public/default/`).
78    ///
79    /// # Errors
80    ///
81    /// Returns [`PulsarError::Invalid`] when the shape or a component is invalid.
82    pub fn parse(name: &str) -> Result<Self, PulsarError> {
83        let (scheme, rest) = name.strip_prefix("non-persistent://").map_or_else(
84            || {
85                name.strip_prefix("persistent://")
86                    .map_or(("persistent", name), |rest| ("persistent", rest))
87            },
88            |rest| ("non-persistent", rest),
89        );
90        let parts: Vec<&str> = rest.split('/').collect();
91        match parts.as_slice() {
92            [topic] => Self::of(scheme, "public", "default", topic),
93            [tenant, namespace, topic] => Self::of(scheme, tenant, namespace, topic),
94            _ => Err(PulsarError::Invalid(format!(
95                "topic '{name}' must be 'topic', 'tenant/namespace/topic', or fully qualified"
96            ))),
97        }
98    }
99
100    /// The fully qualified name sent to the broker.
101    #[must_use]
102    pub fn as_str(&self) -> &str {
103        &self.full
104    }
105}
106
107impl std::fmt::Display for PulsarTopic {
108    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
109        f.write_str(&self.full)
110    }
111}
112
113#[cfg(test)]
114mod tests {
115    use super::*;
116
117    #[test]
118    fn constructors_qualify_fully() {
119        assert_eq!(
120            PulsarTopic::persistent("acme", "orders", "created").as_str(),
121            "persistent://acme/orders/created"
122        );
123        assert_eq!(
124            PulsarTopic::non_persistent("acme", "telemetry", "ticks").as_str(),
125            "non-persistent://acme/telemetry/ticks"
126        );
127    }
128
129    #[test]
130    fn parse_defaults_bare_names_to_public_default() {
131        assert_eq!(
132            PulsarTopic::parse("orders").expect("parses").as_str(),
133            "persistent://public/default/orders"
134        );
135    }
136
137    #[test]
138    fn parse_rejects_malformed_shapes() {
139        assert!(PulsarTopic::parse("a/b").is_err());
140        assert!(PulsarTopic::parse("persistent://a//c").is_err());
141        assert!(PulsarTopic::parse("bad name").is_err());
142    }
143}
ruststream_pulsar/topic.rs

ruststream_pulsar/
topic.rs
