distributed/bus/
topology.rs

1use super::router::MessageRouter;
2use super::TransportError;
3
4/// Default broker namespace applied when a transport is constructed without an
5/// explicit one.
6pub const DEFAULT_BUS_NAMESPACE: &str = "default";
7/// Maximum byte length accepted for a consumer group or namespace name.
8pub const MAX_TOPOLOGY_NAME_LEN: usize = 128;
9
10/// Consumer-group and namespace topology for a broker-backed bus.
11///
12/// Third-party transports can reuse this to apply the same portable naming
13/// rules (`group`/`namespace` validation) the built-in transports use, instead
14/// of reimplementing them. Construct via [`BusTopologyConfig::default`] and
15/// refine with [`group`](Self::group) / [`namespace`](Self::namespace), then
16/// [`validate_for`](Self::validate_for) before touching broker topology.
17#[derive(Clone, Debug, PartialEq, Eq)]
18pub struct BusTopologyConfig {
19    group: Option<String>,
20    namespace: String,
21}
22
23impl Default for BusTopologyConfig {
24    fn default() -> Self {
25        Self {
26            group: None,
27            namespace: DEFAULT_BUS_NAMESPACE.to_string(),
28        }
29    }
30}
31
32impl BusTopologyConfig {
33    /// The default namespace used when none is set.
34    pub fn default_namespace() -> &'static str {
35        DEFAULT_BUS_NAMESPACE
36    }
37
38    /// Set the durable consumer group.
39    pub fn group(mut self, group: impl Into<String>) -> Self {
40        self.group = Some(group.into());
41        self
42    }
43
44    /// Set the broker namespace/prefix.
45    pub fn namespace(mut self, namespace: impl Into<String>) -> Self {
46        self.namespace = namespace.into();
47        self
48    }
49
50    /// The configured namespace without re-validating it.
51    pub fn namespace_unchecked(&self) -> &str {
52        &self.namespace
53    }
54
55    /// Resolve the effective consumer group, falling back to the router's
56    /// identity, validating the result.
57    pub fn resolve_consumer_group<R: MessageRouter>(
58        &self,
59        router: &R,
60        transport: &str,
61    ) -> Result<String, TransportError> {
62        resolve_consumer_group(self.group.as_deref(), router, transport)
63    }
64
65    /// Validate and return the namespace for the given transport.
66    pub fn namespace_for(&self, transport: &str) -> Result<String, TransportError> {
67        validate_namespace(&self.namespace, transport)
68    }
69
70    /// Validate both group and namespace, returning a checked config.
71    pub fn validate_for(self, transport: &str) -> Result<Self, TransportError> {
72        let group = self
73            .group
74            .map(|group| validate_consumer_group(&group, transport))
75            .transpose()?;
76        let namespace = validate_namespace(&self.namespace, transport)?;
77        Ok(Self { group, namespace })
78    }
79}
80
81#[derive(Clone, Copy, Debug, PartialEq, Eq)]
82enum TopologyNameKind {
83    ConsumerGroup,
84    Namespace,
85}
86
87impl TopologyNameKind {
88    fn label(self) -> &'static str {
89        match self {
90            Self::ConsumerGroup => "consumer group",
91            Self::Namespace => "namespace",
92        }
93    }
94
95    fn allows_dot(self) -> bool {
96        matches!(self, Self::Namespace)
97    }
98}
99
100/// Validate a consumer-group name against the portable topology rules.
101pub fn validate_consumer_group(value: &str, transport: &str) -> Result<String, TransportError> {
102    validate_topology_name(value, TopologyNameKind::ConsumerGroup, transport)
103}
104
105/// Validate a namespace name against the portable topology rules (dots allowed).
106pub fn validate_namespace(value: &str, transport: &str) -> Result<String, TransportError> {
107    validate_topology_name(value, TopologyNameKind::Namespace, transport)
108}
109
110/// Resolve a consumer group from an explicit value or the router identity,
111/// validating the result.
112pub fn resolve_consumer_group<R: MessageRouter>(
113    explicit: Option<&str>,
114    router: &R,
115    transport: &str,
116) -> Result<String, TransportError> {
117    let Some(group) = explicit.or_else(|| router.consumer_group()) else {
118        return Err(TransportError::permanent(format!(
119            "{transport} bus requires a consumer group; call `Service::named(..)` \
120             for service consumers or `bus.group(..)` for direct consumers"
121        )));
122    };
123
124    validate_consumer_group(group, transport)
125}
126
127fn validate_topology_name(
128    value: &str,
129    kind: TopologyNameKind,
130    transport: &str,
131) -> Result<String, TransportError> {
132    let label = kind.label();
133    let trimmed = value.trim();
134    if trimmed.is_empty() {
135        return Err(topology_error(transport, label, "cannot be empty"));
136    }
137    if trimmed.len() != value.len() {
138        return Err(topology_error(
139            transport,
140            label,
141            "cannot contain leading or trailing whitespace",
142        ));
143    }
144    if value.len() > MAX_TOPOLOGY_NAME_LEN {
145        return Err(topology_error(
146            transport,
147            label,
148            format!("cannot exceed {MAX_TOPOLOGY_NAME_LEN} bytes"),
149        ));
150    }
151
152    for ch in value.chars() {
153        if ch.is_control() {
154            return Err(topology_error(
155                transport,
156                label,
157                format!("cannot contain control character {}", display_char(ch)),
158            ));
159        }
160        if ch.is_whitespace() {
161            return Err(topology_error(
162                transport,
163                label,
164                format!("cannot contain whitespace character {}", display_char(ch)),
165            ));
166        }
167        if matches!(ch, '*' | '>') {
168            return Err(topology_error(
169                transport,
170                label,
171                format!("cannot contain NATS wildcard {}", display_char(ch)),
172            ));
173        }
174        if matches!(ch, '/' | '\\') {
175            return Err(topology_error(
176                transport,
177                label,
178                format!("cannot contain path separator {}", display_char(ch)),
179            ));
180        }
181        if ch == '.' && !kind.allows_dot() {
182            return Err(topology_error(
183                transport,
184                label,
185                "cannot contain `.`; use `-` or `_` for portable group names",
186            ));
187        }
188        if !(ch.is_ascii_alphanumeric() || matches!(ch, '-' | '_') || ch == '.') {
189            return Err(topology_error(
190                transport,
191                label,
192                format!("cannot contain character {}", display_char(ch)),
193            ));
194        }
195    }
196
197    Ok(value.to_string())
198}
199
200fn topology_error(transport: &str, label: &str, reason: impl std::fmt::Display) -> TransportError {
201    TransportError::permanent(format!("{transport} bus {label} {reason}"))
202}
203
204fn display_char(ch: char) -> String {
205    if ch.is_control() {
206        format!("U+{:04X}", u32::from(ch))
207    } else {
208        format!("`{ch}`")
209    }
210}
211
212#[cfg(test)]
213mod tests {
214    use super::*;
215    use crate::bus::{Message, MessageKind, SubscriptionPlan};
216
217    struct TestRouter {
218        group: Option<&'static str>,
219    }
220
221    impl MessageRouter for TestRouter {
222        fn consumer_group(&self) -> Option<&str> {
223            self.group
224        }
225
226        fn handles(&self, _kind: MessageKind, _name: &str) -> bool {
227            false
228        }
229
230        fn subscription_plan(&self) -> SubscriptionPlan {
231            SubscriptionPlan::default()
232        }
233
234        async fn dispatch(&self, _message: &Message) -> Result<(), TransportError> {
235            Ok(())
236        }
237    }
238
239    #[test]
240    fn resolve_consumer_group_prefers_explicit_bus_group() {
241        let router = TestRouter {
242            group: Some("service"),
243        };
244        let group = resolve_consumer_group(Some("override"), &router, "test").unwrap();
245        assert_eq!(group, "override");
246    }
247
248    #[test]
249    fn resolve_consumer_group_uses_router_identity() {
250        let router = TestRouter {
251            group: Some("service"),
252        };
253        let group = resolve_consumer_group(None, &router, "test").unwrap();
254        assert_eq!(group, "service");
255    }
256
257    #[test]
258    fn resolve_consumer_group_rejects_missing_identity() {
259        let router = TestRouter { group: None };
260        let err = resolve_consumer_group(None, &router, "test").unwrap_err();
261        assert!(err.is_permanent());
262        assert!(err.message().contains("Service::named"));
263    }
264
265    #[test]
266    fn resolve_consumer_group_rejects_whitespace_identity() {
267        let router = TestRouter {
268            group: Some("  service"),
269        };
270        let err = resolve_consumer_group(None, &router, "test").unwrap_err();
271        assert!(err.is_permanent());
272        assert!(err.message().contains("whitespace"));
273    }
274
275    #[test]
276    fn validate_consumer_group_rejects_wildcards() {
277        let err = validate_consumer_group("orders>*", "nats").unwrap_err();
278        assert!(err.message().contains("NATS wildcard"));
279    }
280
281    #[test]
282    fn validate_consumer_group_rejects_path_separators() {
283        let err = validate_consumer_group("tenant/orders", "rabbitmq").unwrap_err();
284        assert!(err.message().contains("path separator"));
285    }
286
287    #[test]
288    fn validate_consumer_group_rejects_overlong_names() {
289        let value = "a".repeat(MAX_TOPOLOGY_NAME_LEN + 1);
290        let err = validate_consumer_group(&value, "kafka").unwrap_err();
291        assert!(err.message().contains("cannot exceed"));
292    }
293
294    #[test]
295    fn validate_namespace_accepts_dotted_names() {
296        let namespace = validate_namespace("todos-prod.v1", "kafka").unwrap();
297        assert_eq!(namespace, "todos-prod.v1");
298    }
299
300    #[test]
301    fn validate_namespace_rejects_control_characters() {
302        let err = validate_namespace("todos\nprod", "nats").unwrap_err();
303        assert!(err.message().contains("control character"));
304    }
305}
distributed/bus/topology.rs

distributed/bus/
topology.rs
