nautilus_common/python/
msgbus.rs

1// -------------------------------------------------------------------------------------------------
2//  Copyright (C) 2015-2025 2Posei Systems Pty Ltd. All rights reserved.
3//  https://poseitrader.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
16use std::rc::Rc;
17
18use nautilus_core::python::to_pyvalue_err;
19use pyo3::{PyObject, PyResult, pyfunction, pymethods};
20
21use super::handler::PythonMessageHandler;
22use crate::msgbus::{
23    BusMessage, MStr, Topic, core::Endpoint, deregister, get_message_bus,
24    handler::ShareableMessageHandler, publish, register, send_any, subscribe, unsubscribe,
25};
26
27#[pymethods]
28impl BusMessage {
29    #[getter]
30    #[pyo3(name = "topic")]
31    fn py_topic(&mut self) -> String {
32        self.topic.to_string()
33    }
34
35    #[getter]
36    #[pyo3(name = "payload")]
37    fn py_payload(&mut self) -> &[u8] {
38        self.payload.as_ref()
39    }
40
41    fn __repr__(&self) -> String {
42        format!("{}('{}')", stringify!(BusMessage), self)
43    }
44
45    fn __str__(&self) -> String {
46        self.to_string()
47    }
48}
49
50/// Sends the `message` to the `endpoint`.
51///
52/// # Errors
53///
54/// Returns an error if `endpoint` is invalid.
55#[pyfunction]
56#[pyo3(name = "msgbus_send")]
57pub fn py_msgbus_send(endpoint: &str, message: PyObject) -> PyResult<()> {
58    let endpoint = MStr::<Endpoint>::endpoint(endpoint).map_err(to_pyvalue_err)?;
59    send_any(endpoint, &message);
60    Ok(())
61}
62
63/// Returns whether there are subscribers for the given `pattern`.
64#[pyfunction]
65#[pyo3(name = "msgbus_is_subscribed")]
66pub fn py_msgbus_is_subscribed(topic: &str, handler: PythonMessageHandler) -> bool {
67    let handler = ShareableMessageHandler(Rc::new(handler));
68    get_message_bus().borrow().is_subscribed(topic, handler)
69}
70
71/// Returns whether there are subscribers for the given `pattern`.
72#[pyfunction]
73#[pyo3(name = "msgbus_is_registered")]
74pub fn py_msgbus_is_registered(endpoint: &str) -> bool {
75    get_message_bus().borrow().is_registered(endpoint)
76}
77
78/// Publishes the `message` to the `topic`.
79///
80/// # Errors
81///
82/// Returns an error if `topic` is invalid.
83#[pyfunction]
84#[pyo3(name = "msgbus_publish")]
85pub fn py_msgbus_publish(topic: &str, message: PyObject) -> PyResult<()> {
86    let topic = MStr::<Topic>::topic(topic).map_err(to_pyvalue_err)?;
87    publish(topic, &message);
88    Ok(())
89}
90
91/// Registers the given `handler` for the `endpoint` address.
92///
93/// Updates endpoint handler if already exists.
94///
95/// # Errors
96///
97/// Returns an error if `endpoint` is invalid.
98#[pyfunction]
99#[pyo3(name = "msgbus_register")]
100pub fn py_msgbus_register(endpoint: &str, handler: PythonMessageHandler) -> PyResult<()> {
101    let endpoint = MStr::<Endpoint>::endpoint(endpoint).map_err(to_pyvalue_err)?;
102    let handler = ShareableMessageHandler(Rc::new(handler));
103    register(endpoint, handler);
104    Ok(())
105}
106
107/// Subscribes the given `handler` to the `topic`.
108///
109/// The priority for the subscription determines the ordering of
110/// handlers receiving messages being processed, higher priority
111/// handlers will receive messages before lower priority handlers.
112///
113/// Safety: Priority should be between 0 and 255
114///
115/// Updates topic handler if already exists.
116///
117/// # Warnings
118///
119/// Assigning priority handling is an advanced feature which *shouldn't
120/// normally be needed by most users*. **Only assign a higher priority to the
121/// subscription if you are certain of what you're doing**. If an inappropriate
122/// priority is assigned then the handler may receive messages before core
123/// system components have been able to process necessary calculations and
124/// produce potential side effects for logically sound behavior.
125#[pyfunction]
126#[pyo3(name = "msgbus_subscribe")]
127#[pyo3(signature = (topic, handler, priority=None))]
128pub fn py_msgbus_subscribe(topic: &str, handler: PythonMessageHandler, priority: Option<u8>) {
129    let pattern = topic.into();
130    let handler = ShareableMessageHandler(Rc::new(handler));
131    subscribe(pattern, handler, priority);
132}
133
134/// Unsubscribes the given `handler` from the `topic`.
135#[pyfunction]
136#[pyo3(name = "msgbus_unsubscribe")]
137pub fn py_msgbus_unsubscribe(topic: &str, handler: PythonMessageHandler) {
138    let pattern = topic.into();
139    let handler = ShareableMessageHandler(Rc::new(handler));
140    unsubscribe(pattern, handler);
141}
142
143/// Deregisters the given `handler` for the `endpoint` address.
144///
145/// # Errors
146///
147/// Returns an error if `endpoint` is invalid.
148#[pyfunction]
149#[pyo3(name = "msgbus_deregister")]
150pub fn py_msgbus_deregister(endpoint: &str) -> PyResult<()> {
151    let endpoint = MStr::<Endpoint>::endpoint(endpoint).map_err(to_pyvalue_err)?;
152    deregister(endpoint);
153    Ok(())
154}
nautilus_common/python/msgbus.rs

nautilus_common/python/
msgbus.rs
