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

nautilus_common/msgbus/
mod.rs
