Refine message bus topic and endpoint validations

Author: cjdsellers

crates/common/src/msgbus/core.rs

@@ -26,13 +26,24 @@ use ahash::{AHashMap, AHashSet};
 use handler::ShareableMessageHandler;
 use indexmap::IndexMap;
 use matching::is_matching_backtracking;
-use nautilus_core::UUID4;
+use nautilus_core::{
+    UUID4,
+    correctness::{FAILED, check_predicate_true, check_valid_string},
+};
 use nautilus_model::identifiers::TraderId;
 use switchboard::MessagingSwitchboard;
 use ustr::Ustr;
 
 use super::{handler, matching, set_message_bus, switchboard};
 
+#[inline(always)]
+fn check_fully_qualified_string(value: &Ustr, key: &str) -> anyhow::Result<()> {
+    check_predicate_true(
+        !value.chars().any(|c| c == '*' || c == '?'),
+        &format!("{key} `value` contained invalid characters, was {value}"),
+    )
+}
+
 /// A string pattern for a subscription. The pattern is used to match topics.
 ///
 /// A pattern is made of characters:
@@ -80,20 +91,29 @@ impl From<&Topic> for Pattern {
 pub struct Topic(pub Ustr);
 
 impl Topic {
-    pub fn as_pattern(&self) -> Pattern {
-        Pattern(self.0)
+    /// Creates a new [`Topic`] instance.
+    ///
+    /// # Errors
+    ///
+    /// Returns an error if:
+    /// - `value` string is empty, all whitespace, or contains non-ASCII characters.
+    /// - `value` string contains wildcard characters '*' or '?'.
+    pub fn new<T: AsRef<str>>(value: T) -> anyhow::Result<Self> {
+        let topic = Ustr::from(value.as_ref());
+        check_valid_string(value, stringify!(value))?;
+        check_fully_qualified_string(&topic, stringify!(Topic))?;
+
+        Ok(Self(topic))
     }
 
-    // TODO: Every initialization should validate
-    pub fn validate<T: AsRef<str>>(topic: T) -> bool {
-        let topic = Ustr::from(topic.as_ref());
-        !topic.chars().any(|c| c == '*' || c == '?')
+    pub fn as_pattern(&self) -> Pattern {
+        Pattern(self.0)
     }
 }
 
 impl<T: AsRef<str>> From<T> for Topic {
     fn from(value: T) -> Self {
-        Self(Ustr::from(value.as_ref()))
+        Self::new(value).expect(FAILED)
     }
 }
 
@@ -118,16 +138,25 @@ impl Display for Topic {
 pub struct Endpoint(pub Ustr);
 
 impl Endpoint {
-    // TODO: Every initialization should validate
-    pub fn validate<T: AsRef<str>>(endpoint: T) -> bool {
-        let endpoint = Ustr::from(endpoint.as_ref());
-        !endpoint.chars().any(|c| c == '*' || c == '?')
+    /// Creates a new [`Endpoint`] instance.
+    ///
+    /// # Errors
+    ///
+    /// Returns an error if:
+    /// - `value` string is empty, all whitespace, or contains non-ASCII characters.
+    /// - `value` string contains wildcard characters '*' or '?'.
+    pub fn new<T: AsRef<str>>(value: T) -> anyhow::Result<Self> {
+        let endpoint = Ustr::from(value.as_ref());
+        check_valid_string(value, stringify!(value))?;
+        check_fully_qualified_string(&endpoint, stringify!(Endpoint))?;
+
+        Ok(Self(endpoint))
     }
 }
 
 impl<T: AsRef<str>> From<T> for Endpoint {
     fn from(value: T) -> Self {
-        Self(Ustr::from(value.as_ref()))
+        Self::new(value).expect(FAILED)
     }
 }
 
@@ -305,12 +334,12 @@ impl MessageBus {
             .collect()
     }
 
-    /// Returns whether there are subscribers for the given `topic`.
+    /// Returns whether there are subscribers for the `topic`.
     pub fn has_subscribers<T: AsRef<str>>(&self, topic: T) -> bool {
         self.subscriptions_count(topic) > 0
     }
 
-    /// Returns the count of subscribers for the given `topic`.
+    /// Returns the count of subscribers for the `topic`.
     #[must_use]
     pub fn subscriptions_count<T: AsRef<str>>(&self, topic: T) -> usize {
         let topic = Topic::from(topic);
@@ -335,14 +364,14 @@ impl MessageBus {
             .collect()
     }
 
-    /// Returns whether there is a registered endpoint for the given `pattern`.
+    /// Returns whether there is a registered endpoint for the `pattern`.
     #[must_use]
     pub fn is_registered<T: AsRef<str>>(&self, endpoint: T) -> bool {
         self.endpoints
             .contains_key(&Endpoint::from(endpoint.as_ref()))
     }
 
-    /// Returns whether the given `handler` is subscribed to the given `pattern`.
+    /// Returns whether the `handler` is subscribed to the `pattern`.
     #[must_use]
     pub fn is_subscribed<T: AsRef<str>>(
         &self,
@@ -358,25 +387,25 @@ impl MessageBus {
     ///
     /// # Errors
     ///
-    /// This function never returns an error (TBD).
+    /// This function never returns an error (TBD once backing database added).
     pub const fn close(&self) -> anyhow::Result<()> {
         // TODO: Integrate the backing database
         Ok(())
     }
 
-    /// Returns the handler for the given `endpoint`.
+    /// Returns the handler for the `endpoint`.
     #[must_use]
     pub fn get_endpoint(&self, endpoint: &Endpoint) -> Option<&ShareableMessageHandler> {
         self.endpoints.get(endpoint)
     }
 
-    /// Returns the handler for the given `correlation_id`.
+    /// Returns the handler for the `correlation_id`.
     #[must_use]
     pub fn get_response_handler(&self, correlation_id: &UUID4) -> Option<&ShareableMessageHandler> {
         self.correlation_index.get(correlation_id)
     }
 
-    /// Finds the subscriptions with pattern matching the given `topic`.
+    /// Finds the subscriptions with pattern matching the `topic`.
     pub(crate) fn find_topic_matches(&self, topic: &Topic) -> Vec<Subscription> {
         self.subscriptions
             .iter()
@@ -390,7 +419,7 @@ impl MessageBus {
             .collect()
     }
 
-    /// Finds the subscriptions which match the given `topic` and caches the
+    /// Finds the subscriptions which match the `topic` and caches the
     /// results in the `patterns` map.
     #[must_use]
     pub fn matching_subscriptions<T: AsRef<str>>(&mut self, topic: T) -> Vec<Subscription> {
@@ -411,7 +440,7 @@ impl MessageBus {
     ///
     /// # Errors
     ///
-    /// Returns an error if a handler is already registered for the given correlation ID.
+    /// Returns an error if `handler` is already registered for the `correlation_id`.
     pub fn register_response_handler(
         &mut self,
         correlation_id: &UUID4,

crates/common/src/msgbus/mod.rs

@@ -87,6 +87,7 @@ pub fn get_message_bus() -> Rc<RefCell<MessageBus>> {
 
 /// Sends the `message` to the `endpoint`.
 pub fn send(endpoint: &Endpoint, message: &dyn Any) {
+    // TODO: This should return a Result (in case endpoint doesn't exist)
     let handler = get_message_bus().borrow().get_endpoint(endpoint).cloned();
     if let Some(handler) = handler {
         handler.0.handle(message);
@@ -178,7 +179,7 @@ pub fn deregister(endpoint: &Endpoint) {
         .shift_remove(endpoint);
 }
 
-/// Subscribes the given `handler` to the `pattern` with an optional `priority`.
+/// Subscribes the `handler` to the `pattern` with an optional `priority`.
 ///
 /// # Warnings
 ///

crates/common/src/msgbus/tests.rs

@@ -208,7 +208,6 @@ fn test_matching_subscriptions() {
 }
 
 #[rstest]
-#[case("*", "*", true)]
 #[case("a", "*", true)]
 #[case("a", "a", true)]
 #[case("a", "b", false)]
@@ -310,9 +309,9 @@ fn test_subscription_pattern_matching() {
     let handler2 = get_stub_shareable_handler(Some(Ustr::from("2")));
     let handler3 = get_stub_shareable_handler(Some(Ustr::from("3")));
 
-    msgbus::subscribe(Pattern::from("data.quotes.*"), handler1, None);
-    msgbus::subscribe(Pattern::from("data.trades.*"), handler2, None);
-    msgbus::subscribe(Pattern::from("data.*.BINANCE.*"), handler3, None);
+    msgbus::subscribe_str("data.quotes.*", handler1, None);
+    msgbus::subscribe_str("data.trades.*", handler2, None);
+    msgbus::subscribe_str("data.*.BINANCE.*", handler3, None);
     assert_eq!(msgbus.borrow().subscriptions().len(), 3);
 
     let topic = "data.quotes.BINANCE.ETHUSDT";

crates/common/src/python/msgbus.rs

@@ -15,20 +15,20 @@
 
 use std::rc::Rc;
 
-use pyo3::{PyObject, pymethods};
-use ustr::Ustr;
+use nautilus_core::python::to_pyvalue_err;
+use pyo3::{PyObject, PyResult, pymethods};
 
 use super::handler::PythonMessageHandler;
 use crate::msgbus::{
-    BusMessage, Endpoint, MessageBus, Pattern, deregister, get_message_bus,
-    handler::ShareableMessageHandler, register, subscribe, unsubscribe,
+    BusMessage, Endpoint, MessageBus, Pattern, Topic, deregister, handler::ShareableMessageHandler,
+    publish, register, send, subscribe, unsubscribe,
 };
 
 #[pymethods]
 impl BusMessage {
     #[getter]
     #[pyo3(name = "topic")]
-    fn py_close(&mut self) -> String {
+    fn py_topic(&mut self) -> String {
         self.topic.to_string()
     }
 
@@ -49,37 +49,45 @@ impl BusMessage {
 
 #[pymethods]
 impl MessageBus {
-    /// Sends a message to an endpoint.
+    /// Sends the `message` to the `endpoint`.
+    ///
+    /// # Errors
+    ///
+    /// Returns an error if `endpoint` is invalid.
     #[pyo3(name = "send")]
-    pub fn py_send(&self, endpoint: &str, message: PyObject) {
-        let endpoint = Endpoint::from(endpoint);
-
-        if let Some(handler) = self.get_endpoint(&endpoint) {
-            handler.0.handle(&message);
-        }
+    pub fn py_send(&self, endpoint: &str, message: PyObject) -> PyResult<()> {
+        let endpoint = Endpoint::new(endpoint).map_err(to_pyvalue_err)?;
+        send(&endpoint, &message);
+        Ok(())
     }
 
-    /// Publish a message to a topic.
+    /// Publishes the `message` to the `topic`.
+    ///
+    /// # Errors
+    ///
+    /// Returns an error if `topic` is invalid.
     #[pyo3(name = "publish")]
     #[staticmethod]
-    pub fn py_publish(topic: &str, message: PyObject) {
-        let topic = Ustr::from(topic);
-        let matching_subs = get_message_bus().borrow_mut().matching_subscriptions(topic);
-
-        for sub in matching_subs {
-            sub.handler.0.handle(&message);
-        }
+    pub fn py_publish(topic: &str, message: PyObject) -> PyResult<()> {
+        let topic = Topic::new(topic).map_err(to_pyvalue_err)?;
+        publish(&topic, &message);
+        Ok(())
     }
 
     /// Registers the given `handler` for the `endpoint` address.
+    ///
+    /// Updates endpoint handler if already exists.
+    ///
+    /// # Errors
+    ///
+    /// Returns an error if `endpoint` is invalid.
     #[pyo3(name = "register")]
     #[staticmethod]
-    pub fn py_register(endpoint: &str, handler: PythonMessageHandler) {
-        let endpoint = Endpoint::from(endpoint);
-
-        // Updates value if key already exists
+    pub fn py_register(endpoint: &str, handler: PythonMessageHandler) -> PyResult<()> {
+        let endpoint = Endpoint::new(endpoint).map_err(to_pyvalue_err)?;
         let handler = ShareableMessageHandler(Rc::new(handler));
         register(endpoint, handler);
+        Ok(())
     }
 
     /// Subscribes the given `handler` to the `topic`.
@@ -90,6 +98,8 @@ impl MessageBus {
     ///
     /// Safety: Priority should be between 0 and 255
     ///
+    /// Updates topic handler if already exists.
+    ///
     /// # Warnings
     ///
     /// Assigning priority handling is an advanced feature which *shouldn't
@@ -102,7 +112,6 @@ impl MessageBus {
     #[pyo3(signature = (topic, handler, priority=None))]
     #[staticmethod]
     pub fn py_subscribe(topic: &str, handler: PythonMessageHandler, priority: Option<u8>) {
-        // Updates value if key already exists
         let pattern = Pattern::from(topic);
         let handler = ShareableMessageHandler(Rc::new(handler));
         subscribe(pattern, handler, priority);
@@ -133,10 +142,15 @@ impl MessageBus {
     }
 
     /// Deregisters the given `handler` for the `endpoint` address.
+    ///
+    /// # Errors
+    ///
+    /// Returns an error if `endpoint` is invalid.
     #[pyo3(name = "deregister")]
     #[staticmethod]
-    pub fn py_deregister(endpoint: &str) {
-        let endpoint = Endpoint::from(endpoint);
+    pub fn py_deregister(endpoint: &str) -> PyResult<()> {
+        let endpoint = Endpoint::new(endpoint).map_err(to_pyvalue_err)?;
         deregister(&endpoint);
+        Ok(())
     }
 }