nautilus_common/msgbus/
mod.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//! A common in-memory `MessageBus` supporting multiple messaging patterns:
17//!
18//! - Point-to-Point
19//! - Pub/Sub
20//! - Request/Response
21
22pub mod core;
23pub mod database;
24pub mod handler;
25pub mod matching;
26pub mod message;
27pub mod stubs;
28pub mod switchboard;
29
30#[cfg(test)]
31mod tests;
32
33pub use core::{Endpoint, MStr, MessageBus, Pattern, Subscription, Topic};
34use std::{
35    self,
36    any::Any,
37    cell::{OnceCell, RefCell},
38    rc::Rc,
39};
40
41use handler::ShareableMessageHandler;
42use matching::is_matching_backtracking;
43use nautilus_core::UUID4;
44use nautilus_model::data::Data;
45use ustr::Ustr;
46
47use crate::messages::data::DataResponse;
48pub use crate::msgbus::message::BusMessage;
49
50// Thread-local storage for MessageBus instances. Each thread (including async runtimes)
51// gets its own MessageBus instance, eliminating the need for unsafe Send/Sync implementations
52// while maintaining the global singleton access pattern that the framework expects.
53thread_local! {
54    static MESSAGE_BUS: OnceCell<Rc<RefCell<MessageBus>>> = const { OnceCell::new() };
55}
56
57/// Sets the thread-local message bus.
58///
59/// # Panics
60///
61/// Panics if a message bus has already been set for this thread.
62pub fn set_message_bus(msgbus: Rc<RefCell<MessageBus>>) {
63    MESSAGE_BUS.with(|bus| {
64        if bus.set(msgbus).is_err() {
65            panic!("Failed to set MessageBus: already initialized for this thread");
66        }
67    });
68}
69
70/// Gets the thread-local message bus.
71///
72/// If no message bus has been set for this thread, a default one is created and initialized.
73/// This ensures each thread gets its own MessageBus instance, preventing data races while
74/// maintaining the singleton pattern that the codebase expects.
75pub fn get_message_bus() -> Rc<RefCell<MessageBus>> {
76    MESSAGE_BUS.with(|bus| {
77        bus.get_or_init(|| {
78            let msgbus = MessageBus::default();
79            Rc::new(RefCell::new(msgbus))
80        })
81        .clone()
82    })
83}
84
85/// Sends the `message` to the `endpoint`.
86pub fn send_any(endpoint: MStr<Endpoint>, message: &dyn Any) {
87    let handler = get_message_bus().borrow().get_endpoint(endpoint).cloned();
88    if let Some(handler) = handler {
89        handler.0.handle(message);
90    } else {
91        log::error!("send_any: no registered endpoint '{endpoint}'");
92    }
93}
94
95/// Sends the `message` to the `endpoint`.
96pub fn send<T: 'static>(endpoint: MStr<Endpoint>, message: T) {
97    let handler = get_message_bus().borrow().get_endpoint(endpoint).cloned();
98    if let Some(handler) = handler {
99        handler.0.handle(&message);
100    } else {
101        log::error!("send: no registered endpoint '{endpoint}'");
102    }
103}
104
105/// Sends the [`DataResponse`] to the registered correlation ID handler.
106pub fn send_response(correlation_id: &UUID4, message: &DataResponse) {
107    let handler = get_message_bus()
108        .borrow()
109        .get_response_handler(correlation_id)
110        .cloned();
111
112    if let Some(handler) = handler {
113        match message {
114            DataResponse::Data(resp) => handler.0.handle(resp),
115            DataResponse::Instrument(resp) => handler.0.handle(resp.as_ref()),
116            DataResponse::Instruments(resp) => handler.0.handle(resp),
117            DataResponse::Book(resp) => handler.0.handle(resp),
118            DataResponse::Quotes(resp) => handler.0.handle(resp),
119            DataResponse::Trades(resp) => handler.0.handle(resp),
120            DataResponse::Bars(resp) => handler.0.handle(resp),
121        }
122    } else {
123        log::error!("send_response: handler not found for correlation_id '{correlation_id}'");
124    }
125}
126
127/// Publish [`Data`] to a topic.
128pub fn publish_data(topic: &Ustr, message: Data) {
129    let matching_subs = get_message_bus()
130        .borrow_mut()
131        .matching_subscriptions(*topic);
132
133    for sub in matching_subs {
134        sub.handler.0.handle(&message);
135    }
136}
137
138pub fn register_response_handler(correlation_id: &UUID4, handler: ShareableMessageHandler) {
139    if let Err(e) = get_message_bus()
140        .borrow_mut()
141        .register_response_handler(correlation_id, handler)
142    {
143        log::error!("Failed to register request handler: {e}");
144    }
145}
146
147/// Publishes the `message` to the `topic`.
148pub fn publish(topic: MStr<Topic>, message: &dyn Any) {
149    let matching_subs = get_message_bus()
150        .borrow_mut()
151        .inner_matching_subscriptions(topic);
152
153    for sub in matching_subs {
154        sub.handler.0.handle(message);
155    }
156}
157
158/// Registers the `handler` for the `endpoint` address.
159pub fn register(endpoint: MStr<Endpoint>, handler: ShareableMessageHandler) {
160    log::debug!(
161        "Registering endpoint '{endpoint}' with handler ID {}",
162        handler.0.id(),
163    );
164
165    // Updates value if key already exists
166    get_message_bus()
167        .borrow_mut()
168        .endpoints
169        .insert(endpoint, handler);
170}
171
172/// Deregisters the handler for the `endpoint` address.
173pub fn deregister(endpoint: MStr<Endpoint>) {
174    log::debug!("Deregistering endpoint '{endpoint}'");
175
176    // Removes entry if it exists for endpoint
177    get_message_bus()
178        .borrow_mut()
179        .endpoints
180        .shift_remove(&endpoint);
181}
182
183/// Subscribes the `handler` to the `pattern` with an optional `priority`.
184///
185/// # Warnings
186///
187/// Assigning priority handling is an advanced feature which *shouldn't
188/// normally be needed by most users*. **Only assign a higher priority to the
189/// subscription if you are certain of what you're doing**. If an inappropriate
190/// priority is assigned then the handler may receive messages before core
191/// system components have been able to process necessary calculations and
192/// produce potential side effects for logically sound behavior.
193pub fn subscribe(pattern: MStr<Pattern>, handler: ShareableMessageHandler, priority: Option<u8>) {
194    let msgbus = get_message_bus();
195    let mut msgbus_ref_mut = msgbus.borrow_mut();
196    let sub = Subscription::new(pattern, handler, priority);
197
198    log::debug!(
199        "Subscribing {:?} for pattern '{}'",
200        sub.handler,
201        sub.pattern
202    );
203
204    // Prevent duplicate subscriptions for the exact pattern regardless of handler identity. This
205    // guards against callers accidentally registering multiple handlers for the same topic, which
206    // can lead to duplicated message delivery and unexpected side-effects.
207    if msgbus_ref_mut.subscriptions.contains(&sub) {
208        log::warn!("{sub:?} already exists");
209        return;
210    }
211
212    // Find existing patterns which match this topic
213    for (topic, subs) in &mut msgbus_ref_mut.topics {
214        if is_matching_backtracking(*topic, sub.pattern) {
215            // TODO: Consider binary_search and then insert
216            subs.push(sub.clone());
217            subs.sort();
218            log::debug!("Added subscription for '{topic}'");
219        }
220    }
221
222    msgbus_ref_mut.subscriptions.insert(sub);
223}
224
225pub fn subscribe_topic(topic: MStr<Topic>, handler: ShareableMessageHandler, priority: Option<u8>) {
226    subscribe(topic.into(), handler, priority);
227}
228
229pub fn subscribe_str<T: AsRef<str>>(
230    pattern: T,
231    handler: ShareableMessageHandler,
232    priority: Option<u8>,
233) {
234    subscribe(MStr::from(pattern.as_ref()), handler, priority);
235}
236
237/// Unsubscribes the `handler` from the `pattern`.
238pub fn unsubscribe(pattern: MStr<Pattern>, handler: ShareableMessageHandler) {
239    log::debug!("Unsubscribing {handler:?} from pattern '{pattern}'");
240
241    let sub = core::Subscription::new(pattern, handler, None);
242
243    get_message_bus()
244        .borrow_mut()
245        .topics
246        .values_mut()
247        .for_each(|subs| {
248            if let Ok(index) = subs.binary_search(&sub) {
249                subs.remove(index);
250            }
251        });
252
253    let removed = get_message_bus().borrow_mut().subscriptions.remove(&sub);
254
255    if removed {
256        log::debug!("Handler for pattern '{pattern}' was removed");
257    } else {
258        log::debug!("No matching handler for pattern '{pattern}' was found");
259    }
260}
261
262pub fn unsubscribe_topic(topic: MStr<Topic>, handler: ShareableMessageHandler) {
263    unsubscribe(topic.into(), handler);
264}
265
266pub fn unsubscribe_str<T: AsRef<str>>(pattern: T, handler: ShareableMessageHandler) {
267    unsubscribe(MStr::from(pattern.as_ref()), handler);
268}
269
270pub fn is_subscribed<T: AsRef<str>>(pattern: T, handler: ShareableMessageHandler) -> bool {
271    let pattern = MStr::from(pattern.as_ref());
272    let sub = Subscription::new(pattern, handler, None);
273    get_message_bus().borrow().subscriptions.contains(&sub)
274}
275
276pub fn subscriptions_count<T: AsRef<str>>(topic: T) -> usize {
277    get_message_bus().borrow().subscriptions_count(topic)
278}
nautilus_common/msgbus/mod.rs

nautilus_common/msgbus/
mod.rs
