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//! In-memory message bus for intra-process communication.
17//!
18//! # Messaging patterns
19//!
20//! - **Point-to-point**: Send messages to named endpoints via `send_*` functions.
21//! - **Pub/sub**: Publish messages to topics via `publish_*`, subscribers receive
22//!   all messages matching their pattern.
23//! - **Request/response**: Register correlation IDs for response sequence tracking.
24//!
25//! # Architecture
26//!
27//! The bus uses thread-local storage for single-threaded async runtimes. Each
28//! thread gets its own `MessageBus` instance, avoiding synchronization overhead.
29//!
30//! Two routing mechanisms serve different needs:
31//!
32//! - **Typed routing** (`publish_quote`, `subscribe_quotes`): Zero-cost dispatch
33//!   for known types. Handlers receive `&T` directly with no runtime type checking.
34//! - **Any-based routing** (`publish_any`, `subscribe_any`): Flexible dispatch for
35//!   custom types and Python interop. Handlers receive `&dyn Any`.
36//!
37//! See [`core`] module documentation for design decisions and performance details.
38
39pub mod config;
40pub mod core;
41pub mod matching;
42pub mod message;
43pub mod mstr;
44pub mod stubs;
45pub mod switchboard;
46pub mod typed_endpoints;
47pub mod typed_handler;
48pub mod typed_router;
49
50pub(crate) mod external;
51
52mod api;
53mod backing;
54
55use std::{
56    any::Any,
57    cell::{Cell, RefCell},
58    rc::Rc,
59};
60
61use nautilus_core::UUID4;
62#[cfg(feature = "defi")]
63use nautilus_model::defi::{Block, Pool, PoolFeeCollect, PoolFlash, PoolLiquidityUpdate, PoolSwap};
64use nautilus_model::{
65    data::{
66        Bar, FundingRateUpdate, GreeksData, IndexPriceUpdate, MarkPriceUpdate, OrderBookDeltas,
67        OrderBookDepth, QuoteTick, TradeTick,
68        option_chain::{OptionChainSlice, OptionGreeks},
69    },
70    events::{AccountState, OrderEventAny, PortfolioSnapshot, PositionEvent},
71    instruments::InstrumentAny,
72    orderbook::OrderBook,
73};
74use smallvec::SmallVec;
75
76#[cfg(feature = "live")]
77pub use self::backing::{
78    MessageBusExternalIngress, MessageBusExternalReceiver, external_io_from_backing,
79};
80pub use self::{
81    api::*,
82    backing::{
83        MessageBusBacking, MessageBusBackingFactory, MessageBusExternalEgress,
84        external_egress_from_backing,
85    },
86    config::MessageBusConfig,
87    core::{MessageBus, Subscription},
88    external::{current_external_source, republish_pending_external_response},
89    message::{BusMessage, BusMessageSource, BusPayloadCategory, BusPayloadType},
90    mstr::{Endpoint, MStr, Pattern, Topic},
91    switchboard::MessagingSwitchboard,
92    typed_endpoints::{EndpointMap, IntoEndpointMap},
93    typed_handler::{
94        CallbackHandler, Handler, IntoHandler, ShareableMessageHandler, TypedHandler,
95        TypedIntoHandler,
96    },
97    typed_router::{TopicRouter, TypedSubscription},
98};
99use crate::timer::TimeEvent;
100
101/// Inline capacity for handler buffers before heap allocation.
102pub(super) const HANDLER_BUFFER_CAP: usize = 64;
103
104// MessageBus is designed for single-threaded use within each async runtime.
105// Thread-local storage ensures each thread gets its own instance, eliminating
106// the need for unsafe Send/Sync implementations.
107//
108// Handler buffers provide zero-allocation publish on hot paths.
109// Each buffer stores up to 64 handlers inline before spilling to heap.
110// Publish functions use move-out/move-back to avoid holding RefCell borrows
111// during handler calls (enabling re-entrant publishes).
112thread_local! {
113    pub(super) static MESSAGE_BUS: RefCell<Option<Rc<RefCell<MessageBus>>>> = const { RefCell::new(None) };
114    pub(super) static HAS_EXTERNAL_EGRESS: Cell<bool> = const { Cell::new(false) };
115    pub(super) static SUPPRESS_EXTERNAL_DEPTH: Cell<u32> = const { Cell::new(0) };
116
117    pub(super) static ANY_HANDLERS: RefCell<SmallVec<[ShareableMessageHandler; HANDLER_BUFFER_CAP]>> =
118        RefCell::new(SmallVec::new());
119
120    pub(super) static DELTAS_HANDLERS: RefCell<SmallVec<[TypedHandler<OrderBookDeltas>; HANDLER_BUFFER_CAP]>> =
121        RefCell::new(SmallVec::new());
122    pub(super) static DEPTH_HANDLERS: RefCell<SmallVec<[TypedHandler<OrderBookDepth>; HANDLER_BUFFER_CAP]>> =
123        RefCell::new(SmallVec::new());
124    pub(super) static BOOK_HANDLERS: RefCell<SmallVec<[TypedHandler<OrderBook>; HANDLER_BUFFER_CAP]>> =
125        RefCell::new(SmallVec::new());
126    pub(super) static QUOTE_HANDLERS: RefCell<SmallVec<[TypedHandler<QuoteTick>; HANDLER_BUFFER_CAP]>> =
127        RefCell::new(SmallVec::new());
128    pub(super) static TRADE_HANDLERS: RefCell<SmallVec<[TypedHandler<TradeTick>; HANDLER_BUFFER_CAP]>> =
129        RefCell::new(SmallVec::new());
130    pub(super) static BAR_HANDLERS: RefCell<SmallVec<[TypedHandler<Bar>; HANDLER_BUFFER_CAP]>> =
131        RefCell::new(SmallVec::new());
132    pub(super) static MARK_PRICE_HANDLERS: RefCell<SmallVec<[TypedHandler<MarkPriceUpdate>; HANDLER_BUFFER_CAP]>> =
133        RefCell::new(SmallVec::new());
134    pub(super) static INDEX_PRICE_HANDLERS: RefCell<SmallVec<[TypedHandler<IndexPriceUpdate>; HANDLER_BUFFER_CAP]>> =
135        RefCell::new(SmallVec::new());
136    pub(super) static FUNDING_RATE_HANDLERS: RefCell<SmallVec<[TypedHandler<FundingRateUpdate>; HANDLER_BUFFER_CAP]>> =
137        RefCell::new(SmallVec::new());
138    pub(super) static GREEKS_HANDLERS: RefCell<SmallVec<[TypedHandler<GreeksData>; HANDLER_BUFFER_CAP]>> =
139        RefCell::new(SmallVec::new());
140    pub(super) static OPTION_GREEKS_HANDLERS: RefCell<SmallVec<[TypedHandler<OptionGreeks>; HANDLER_BUFFER_CAP]>> =
141        RefCell::new(SmallVec::new());
142    pub(super) static OPTION_CHAIN_HANDLERS: RefCell<SmallVec<[TypedHandler<OptionChainSlice>; HANDLER_BUFFER_CAP]>> =
143        RefCell::new(SmallVec::new());
144    pub(super) static ACCOUNT_STATE_HANDLERS: RefCell<SmallVec<[TypedHandler<AccountState>; HANDLER_BUFFER_CAP]>> =
145        RefCell::new(SmallVec::new());
146    pub(super) static PORTFOLIO_SNAPSHOT_HANDLERS: RefCell<SmallVec<[TypedHandler<PortfolioSnapshot>; HANDLER_BUFFER_CAP]>> =
147        RefCell::new(SmallVec::new());
148    pub(super) static ORDER_EVENT_HANDLERS: RefCell<SmallVec<[TypedHandler<OrderEventAny>; HANDLER_BUFFER_CAP]>> =
149        RefCell::new(SmallVec::new());
150    pub(super) static POSITION_EVENT_HANDLERS: RefCell<SmallVec<[TypedHandler<PositionEvent>; HANDLER_BUFFER_CAP]>> =
151        RefCell::new(SmallVec::new());
152    pub(super) static INSTRUMENT_HANDLERS: RefCell<SmallVec<[TypedHandler<InstrumentAny>; HANDLER_BUFFER_CAP]>> =
153        RefCell::new(SmallVec::new());
154
155    #[cfg(feature = "defi")]
156    pub(super) static DEFI_BLOCK_HANDLERS: RefCell<SmallVec<[TypedHandler<Block>; HANDLER_BUFFER_CAP]>> =
157        RefCell::new(SmallVec::new());
158    #[cfg(feature = "defi")]
159    pub(super) static DEFI_POOL_HANDLERS: RefCell<SmallVec<[TypedHandler<Pool>; HANDLER_BUFFER_CAP]>> =
160        RefCell::new(SmallVec::new());
161    #[cfg(feature = "defi")]
162    pub(super) static DEFI_SWAP_HANDLERS: RefCell<SmallVec<[TypedHandler<PoolSwap>; HANDLER_BUFFER_CAP]>> =
163        RefCell::new(SmallVec::new());
164    #[cfg(feature = "defi")]
165    pub(super) static DEFI_LIQUIDITY_HANDLERS: RefCell<SmallVec<[TypedHandler<PoolLiquidityUpdate>; HANDLER_BUFFER_CAP]>> =
166        RefCell::new(SmallVec::new());
167    #[cfg(feature = "defi")]
168    pub(super) static DEFI_COLLECT_HANDLERS: RefCell<SmallVec<[TypedHandler<PoolFeeCollect>; HANDLER_BUFFER_CAP]>> =
169        RefCell::new(SmallVec::new());
170    #[cfg(feature = "defi")]
171    pub(super) static DEFI_FLASH_HANDLERS: RefCell<SmallVec<[TypedHandler<PoolFlash>; HANDLER_BUFFER_CAP]>> =
172        RefCell::new(SmallVec::new());
173}
174
175/// Guard that prevents republished external messages from being forwarded again.
176#[derive(Debug)]
177pub struct SuppressExternalGuard;
178
179impl SuppressExternalGuard {
180    #[must_use]
181    pub fn new() -> Self {
182        SUPPRESS_EXTERNAL_DEPTH.with(|depth| depth.set(depth.get().saturating_add(1)));
183        Self
184    }
185}
186
187impl Default for SuppressExternalGuard {
188    fn default() -> Self {
189        Self::new()
190    }
191}
192
193impl Drop for SuppressExternalGuard {
194    fn drop(&mut self) {
195        SUPPRESS_EXTERNAL_DEPTH.with(|depth| depth.set(depth.get().saturating_sub(1)));
196    }
197}
198
199/// Sets the thread-local message bus, replacing any existing one.
200pub fn set_message_bus(msgbus: Rc<RefCell<MessageBus>>) {
201    HAS_EXTERNAL_EGRESS.with(|flag| flag.set(msgbus.borrow().has_external_egress()));
202    MESSAGE_BUS.with(|bus| {
203        *bus.borrow_mut() = Some(msgbus);
204    });
205}
206
207/// Gets the thread-local message bus.
208///
209/// If no message bus has been set for this thread, a default one is created and initialized.
210pub fn get_message_bus() -> Rc<RefCell<MessageBus>> {
211    MESSAGE_BUS.with(|bus| {
212        let mut slot = bus.borrow_mut();
213        let rc = slot.get_or_insert_with(|| Rc::new(RefCell::new(MessageBus::default())));
214        rc.clone()
215    })
216}
217
218/// Returns the thread-local message bus if one has been set for this thread.
219pub fn try_get_message_bus() -> Option<Rc<RefCell<MessageBus>>> {
220    MESSAGE_BUS.with(|bus| bus.borrow().clone())
221}
222
223/// Observes dispatched bus traffic for the durable event store.
224///
225/// The bus invokes the registered tap (when present) before each publish, send, or
226/// correlation response fanout, so subscribers cannot observe a message that has not
227/// yet been handed to the tap. The tap callback runs on the engine thread and must be
228/// cheap; it must not re-enter the bus (the bus is single-threaded and the call site
229/// holds no live borrow of the bus, so any re-entrant publish would deadlock through
230/// downstream `RefCell::borrow_mut` calls inside the registered tap).
231pub trait BusTap: 'static {
232    /// Invoked before a publish fanout dispatches to subscribers on `topic`.
233    fn on_publish(&self, topic: MStr<Topic>, message: &dyn Any);
234
235    /// Invoked before a send dispatch reaches the endpoint handler.
236    fn on_send(&self, endpoint: MStr<Endpoint>, message: &dyn Any);
237
238    /// Invoked before a correlation response dispatch reaches the response handler.
239    fn on_response(&self, _correlation_id: &UUID4, _message: &dyn Any) {}
240}
241
242thread_local! {
243    pub(super) static BUS_TAP: RefCell<Option<Rc<dyn BusTap>>> = const { RefCell::new(None) };
244}
245
246/// Registers `tap` as the thread-local bus tap, replacing any previously installed tap.
247///
248/// The tap fires before each publish, send, and correlation response fanout. Callers
249/// are responsible for clearing the tap on shutdown via [`clear_bus_tap`] so a stale
250/// adapter does not outlive the writer it captures into.
251pub fn set_bus_tap(tap: Rc<dyn BusTap>) {
252    BUS_TAP.with(|slot| {
253        *slot.borrow_mut() = Some(tap);
254    });
255}
256
257/// Clears the registered bus tap on this thread.
258///
259/// A no-op when no tap is installed.
260pub fn clear_bus_tap() {
261    BUS_TAP.with(|slot| {
262        *slot.borrow_mut() = None;
263    });
264}
265
266#[inline]
267pub(super) fn dispatch_tap_publish(topic: MStr<Topic>, message: &dyn Any) {
268    // Clone the Rc so the cell borrow is released before the tap runs. The tap is
269    // single-threaded with the bus; a re-entrant `set_bus_tap` during dispatch would
270    // otherwise panic on RefCell.
271    let tap = BUS_TAP.with(|slot| slot.borrow().clone());
272    if let Some(tap) = tap {
273        tap.on_publish(topic, message);
274    }
275}
276
277#[inline]
278pub(super) fn dispatch_tap_send(endpoint: MStr<Endpoint>, message: &dyn Any) {
279    let tap = BUS_TAP.with(|slot| slot.borrow().clone());
280    if let Some(tap) = tap {
281        tap.on_send(endpoint, message);
282    }
283}
284
285#[inline]
286pub(super) fn dispatch_tap_response(correlation_id: &UUID4, message: &dyn Any) {
287    let tap = BUS_TAP.with(|slot| slot.borrow().clone());
288    if let Some(tap) = tap {
289        tap.on_response(correlation_id, message);
290    }
291}
292
293#[inline]
294pub(crate) fn dispatch_tap_time_event(event: &TimeEvent) {
295    dispatch_tap_publish(MessagingSwitchboard::time_event_topic(), event);
296}
nautilus_common/msgbus/mod.rs

nautilus_common/msgbus/
mod.rs
