nautilus_common/msgbus/
mstr.rs

1// -------------------------------------------------------------------------------------------------
2//  Copyright (C) 2015-2026 Nautech Systems Pty Ltd. All rights reserved.
3//  https://nautechsystems.io
4//
5//  Licensed under the GNU Lesser General Public License Version 3.0 (the "License");
6//  You may not use this file except in compliance with the License.
7//  You may obtain a copy of the License at https://www.gnu.org/licenses/lgpl-3.0.en.html
8//
9//  Unless required by applicable law or agreed to in writing, software
10//  distributed under the License is distributed on an "AS IS" BASIS,
11//  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12//  See the License for the specific language governing permissions and
13//  limitations under the License.
14// -------------------------------------------------------------------------------------------------
15
16//! Type-safe string wrappers for message bus patterns, topics, and endpoints.
17
18use std::{fmt::Display, ops::Deref};
19
20use nautilus_core::correctness::{FAILED, check_valid_string_utf8};
21use serde::{Deserialize, Serialize};
22use ustr::Ustr;
23
24/// Check that a string contains no wildcard characters.
25#[inline(always)]
26fn check_no_wildcards(value: &Ustr, key: &str) -> anyhow::Result<()> {
27    // Check bytes directly - faster than chars() for ASCII wildcards
28    if value.as_bytes().iter().any(|&b| b == b'*' || b == b'?') {
29        anyhow::bail!("{key} `value` contained invalid characters, was {value}");
30    }
31    Ok(())
32}
33
34/// Marker for subscription patterns. Allows wildcards (`*`, `?`).
35#[derive(Copy, Clone, Debug, PartialEq, Eq, Hash)]
36pub struct Pattern;
37
38/// Marker for publish topics. No wildcards allowed.
39#[derive(Copy, Clone, Debug, PartialEq, Eq, Hash)]
40pub struct Topic;
41
42/// Marker for direct message endpoints. No wildcards allowed.
43#[derive(Copy, Clone, Debug, PartialEq, Eq, Hash)]
44pub struct Endpoint;
45
46/// A message bus string type parameterized by marker type.
47///
48/// - `MStr<Pattern>` - for subscriptions, allows wildcards (`*`, `?`)
49/// - `MStr<Topic>` - for publishing, no wildcards
50/// - `MStr<Endpoint>` - for direct messages, no wildcards
51#[derive(Copy, Clone, Debug, PartialEq, Eq, Hash, Serialize, Deserialize)]
52#[serde(transparent)]
53pub struct MStr<T> {
54    value: Ustr,
55    #[serde(skip)]
56    _marker: std::marker::PhantomData<T>,
57}
58
59impl<T> Display for MStr<T> {
60    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
61        write!(f, "{}", self.value)
62    }
63}
64
65impl<T> Deref for MStr<T> {
66    type Target = Ustr;
67
68    fn deref(&self) -> &Self::Target {
69        &self.value
70    }
71}
72
73impl<T> AsRef<str> for MStr<T> {
74    fn as_ref(&self) -> &str {
75        self.value.as_str()
76    }
77}
78
79impl MStr<Pattern> {
80    /// Create a new pattern from a string.
81    pub fn pattern<T: AsRef<str>>(value: T) -> Self {
82        let value = Ustr::from(value.as_ref());
83
84        Self {
85            value,
86            _marker: std::marker::PhantomData,
87        }
88    }
89}
90
91impl From<&str> for MStr<Pattern> {
92    fn from(value: &str) -> Self {
93        Self::pattern(value)
94    }
95}
96
97impl From<String> for MStr<Pattern> {
98    fn from(value: String) -> Self {
99        value.as_str().into()
100    }
101}
102
103impl From<&String> for MStr<Pattern> {
104    fn from(value: &String) -> Self {
105        value.as_str().into()
106    }
107}
108
109impl From<MStr<Topic>> for MStr<Pattern> {
110    fn from(value: MStr<Topic>) -> Self {
111        Self {
112            value: value.value,
113            _marker: std::marker::PhantomData,
114        }
115    }
116}
117
118impl MStr<Topic> {
119    /// Create a new topic from a fully qualified string.
120    ///
121    /// # Errors
122    ///
123    /// Returns an error if the topic has white space or invalid characters.
124    pub fn topic<T: AsRef<str>>(value: T) -> anyhow::Result<Self> {
125        let topic = Ustr::from(value.as_ref());
126        check_valid_string_utf8(value, stringify!(value))?;
127        check_no_wildcards(&topic, stringify!(Topic))?;
128
129        Ok(Self {
130            value: topic,
131            _marker: std::marker::PhantomData,
132        })
133    }
134
135    /// Create a topic from an already-interned Ustr.
136    ///
137    /// # Errors
138    ///
139    /// Returns an error if the topic is empty, all whitespace, or contains wildcard characters.
140    pub fn topic_from_ustr(value: Ustr) -> anyhow::Result<Self> {
141        check_valid_string_utf8(value.as_str(), stringify!(value))?;
142        check_no_wildcards(&value, stringify!(Topic))?;
143
144        Ok(Self {
145            value,
146            _marker: std::marker::PhantomData,
147        })
148    }
149}
150
151impl From<&str> for MStr<Topic> {
152    fn from(value: &str) -> Self {
153        Self::topic(value).expect(FAILED)
154    }
155}
156
157impl From<String> for MStr<Topic> {
158    fn from(value: String) -> Self {
159        value.as_str().into()
160    }
161}
162
163impl From<&String> for MStr<Topic> {
164    fn from(value: &String) -> Self {
165        value.as_str().into()
166    }
167}
168
169impl From<Ustr> for MStr<Topic> {
170    fn from(value: Ustr) -> Self {
171        Self::topic_from_ustr(value).expect(FAILED)
172    }
173}
174
175impl From<&Ustr> for MStr<Topic> {
176    fn from(value: &Ustr) -> Self {
177        (*value).into()
178    }
179}
180
181impl MStr<Endpoint> {
182    /// Create a new endpoint from a fully qualified string.
183    ///
184    /// # Errors
185    ///
186    /// Returns an error if the endpoint has white space or invalid characters.
187    pub fn endpoint<T: AsRef<str>>(value: T) -> anyhow::Result<Self> {
188        let endpoint = Ustr::from(value.as_ref());
189        check_valid_string_utf8(value, stringify!(value))?;
190        check_no_wildcards(&endpoint, stringify!(Endpoint))?;
191
192        Ok(Self {
193            value: endpoint,
194            _marker: std::marker::PhantomData,
195        })
196    }
197
198    /// Create an endpoint from an already-interned Ustr.
199    ///
200    /// # Errors
201    ///
202    /// Returns an error if the endpoint is empty, all whitespace, or contains wildcard characters.
203    pub fn endpoint_from_ustr(value: Ustr) -> anyhow::Result<Self> {
204        check_valid_string_utf8(value.as_str(), stringify!(value))?;
205        check_no_wildcards(&value, stringify!(Endpoint))?;
206
207        Ok(Self {
208            value,
209            _marker: std::marker::PhantomData,
210        })
211    }
212}
213
214impl From<&str> for MStr<Endpoint> {
215    fn from(value: &str) -> Self {
216        Self::endpoint(value).expect(FAILED)
217    }
218}
219
220impl From<String> for MStr<Endpoint> {
221    fn from(value: String) -> Self {
222        value.as_str().into()
223    }
224}
225
226impl From<&String> for MStr<Endpoint> {
227    fn from(value: &String) -> Self {
228        value.as_str().into()
229    }
230}
231
232impl From<Ustr> for MStr<Endpoint> {
233    fn from(value: Ustr) -> Self {
234        Self::endpoint_from_ustr(value).expect(FAILED)
235    }
236}
237
238#[cfg(test)]
239mod tests {
240    use proptest::prelude::*;
241    use rstest::rstest;
242
243    use super::*;
244
245    #[rstest]
246    #[case("data.quotes.BINANCE.BTCUSDT")]
247    #[case("events.order.filled")]
248    #[case("a")]
249    #[case("a.b.c.d.e.f")]
250    fn test_topic_valid(#[case] input: &str) {
251        let topic = MStr::<Topic>::topic(input).unwrap();
252        assert_eq!(topic.as_ref(), input);
253    }
254
255    #[rstest]
256    #[case("data.*.BINANCE")]
257    #[case("events.order.*")]
258    #[case("*")]
259    #[case("data.quotes.?")]
260    #[case("a?b")]
261    fn test_topic_rejects_wildcards(#[case] input: &str) {
262        assert!(MStr::<Topic>::topic(input).is_err());
263    }
264
265    #[rstest]
266    #[case("DataEngine.execute")]
267    #[case("RiskEngine.process")]
268    fn test_endpoint_valid(#[case] input: &str) {
269        let endpoint = MStr::<Endpoint>::endpoint(input).unwrap();
270        assert_eq!(endpoint.as_ref(), input);
271    }
272
273    #[rstest]
274    #[case("DataEngine.*")]
275    #[case("*.execute")]
276    #[case("Risk?Engine")]
277    fn test_endpoint_rejects_wildcards(#[case] input: &str) {
278        assert!(MStr::<Endpoint>::endpoint(input).is_err());
279    }
280
281    #[rstest]
282    #[case("data.*")]
283    #[case("*.quotes.*")]
284    #[case("data.?.BINANCE")]
285    #[case("*")]
286    #[case("exact.match.no.wildcards")]
287    fn test_pattern_accepts_all(#[case] input: &str) {
288        let pattern = MStr::<Pattern>::pattern(input);
289        assert_eq!(pattern.as_ref(), input);
290    }
291
292    #[rstest]
293    fn test_topic_to_pattern_conversion() {
294        let topic: MStr<Topic> = "data.quotes.BINANCE.BTCUSDT".into();
295        let pattern: MStr<Pattern> = topic.into();
296        assert_eq!(pattern.as_ref(), "data.quotes.BINANCE.BTCUSDT");
297    }
298
299    #[rstest]
300    fn test_topic_from_ustr_valid() {
301        let ustr = Ustr::from("data.quotes.BINANCE");
302        let topic = MStr::<Topic>::topic_from_ustr(ustr).unwrap();
303        assert_eq!(topic.as_ref(), "data.quotes.BINANCE");
304    }
305
306    #[rstest]
307    #[case("")]
308    #[case("   ")]
309    #[case("\t\n")]
310    fn test_topic_from_ustr_rejects_empty_whitespace(#[case] input: &str) {
311        let ustr = Ustr::from(input);
312        assert!(MStr::<Topic>::topic_from_ustr(ustr).is_err());
313    }
314
315    #[rstest]
316    #[case("data.*")]
317    #[case("a?b")]
318    fn test_topic_from_ustr_rejects_wildcards(#[case] input: &str) {
319        let ustr = Ustr::from(input);
320        assert!(MStr::<Topic>::topic_from_ustr(ustr).is_err());
321    }
322
323    #[rstest]
324    fn test_endpoint_from_ustr_valid() {
325        let ustr = Ustr::from("DataEngine.execute");
326        let endpoint = MStr::<Endpoint>::endpoint_from_ustr(ustr).unwrap();
327        assert_eq!(endpoint.as_ref(), "DataEngine.execute");
328    }
329
330    #[rstest]
331    #[case("")]
332    #[case("   ")]
333    fn test_endpoint_from_ustr_rejects_empty_whitespace(#[case] input: &str) {
334        let ustr = Ustr::from(input);
335        assert!(MStr::<Endpoint>::endpoint_from_ustr(ustr).is_err());
336    }
337
338    #[rstest]
339    #[case("Engine.*")]
340    #[case("a?b")]
341    fn test_endpoint_from_ustr_rejects_wildcards(#[case] input: &str) {
342        let ustr = Ustr::from(input);
343        assert!(MStr::<Endpoint>::endpoint_from_ustr(ustr).is_err());
344    }
345
346    #[rstest]
347    fn test_from_impls_equivalent() {
348        let s = "test.topic";
349        let from_str: MStr<Topic> = s.into();
350        let from_string: MStr<Topic> = s.to_string().into();
351        let from_string_ref: MStr<Topic> = (&s.to_string()).into();
352        let from_ustr: MStr<Topic> = Ustr::from(s).into();
353
354        assert_eq!(from_str, from_string);
355        assert_eq!(from_string, from_string_ref);
356        assert_eq!(from_string_ref, from_ustr);
357    }
358
359    #[rstest]
360    fn test_deref_to_ustr() {
361        let topic: MStr<Topic> = "test.topic".into();
362        let ustr: &Ustr = &topic;
363        assert_eq!(ustr.as_str(), "test.topic");
364    }
365
366    fn valid_segment() -> impl Strategy<Value = String> {
367        "[a-zA-Z][a-zA-Z0-9_]{0,15}".prop_filter("non-empty", |s| !s.is_empty())
368    }
369
370    fn valid_topic_string() -> impl Strategy<Value = String> {
371        prop::collection::vec(valid_segment(), 1..=5).prop_map(|segs| segs.join("."))
372    }
373
374    fn string_with_wildcards() -> impl Strategy<Value = String> {
375        prop::collection::vec(
376            prop_oneof![
377                valid_segment(),
378                Just("*".to_string()),
379                Just("?".to_string()),
380            ],
381            1..=5,
382        )
383        .prop_map(|segs| segs.join("."))
384        .prop_filter("must contain wildcard", |s| {
385            s.contains('*') || s.contains('?')
386        })
387    }
388
389    proptest! {
390        #[rstest]
391        fn prop_topic_roundtrip(s in valid_topic_string()) {
392            let topic = MStr::<Topic>::topic(&s).unwrap();
393            prop_assert_eq!(topic.as_ref(), s.as_str());
394        }
395
396        #[rstest]
397        fn prop_endpoint_roundtrip(s in valid_topic_string()) {
398            let endpoint = MStr::<Endpoint>::endpoint(&s).unwrap();
399            prop_assert_eq!(endpoint.as_ref(), s.as_str());
400        }
401
402        #[rstest]
403        fn prop_pattern_accepts_wildcards(s in string_with_wildcards()) {
404            let pattern = MStr::<Pattern>::pattern(&s);
405            prop_assert_eq!(pattern.as_ref(), s.as_str());
406        }
407
408        #[rstest]
409        fn prop_topic_rejects_wildcards(s in string_with_wildcards()) {
410            prop_assert!(MStr::<Topic>::topic(&s).is_err());
411        }
412
413        #[rstest]
414        fn prop_endpoint_rejects_wildcards(s in string_with_wildcards()) {
415            prop_assert!(MStr::<Endpoint>::endpoint(&s).is_err());
416        }
417
418        #[rstest]
419        fn prop_topic_to_pattern_preserves_value(s in valid_topic_string()) {
420            let topic: MStr<Topic> = MStr::topic(&s).unwrap();
421            let pattern: MStr<Pattern> = topic.into();
422            prop_assert_eq!(pattern.as_ref(), s.as_str());
423        }
424
425        #[rstest]
426        fn prop_from_impls_consistent(s in valid_topic_string()) {
427            let from_str: MStr<Topic> = s.as_str().into();
428            let from_string: MStr<Topic> = s.clone().into();
429            let from_ustr: MStr<Topic> = Ustr::from(&s).into();
430
431            prop_assert_eq!(from_str, from_string);
432            prop_assert_eq!(from_string, from_ustr);
433        }
434    }
435}
nautilus_common/msgbus/mstr.rs

nautilus_common/msgbus/
mstr.rs
