nautilus_common/msgbus/
core.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//! Core message bus implementation.
17//!
18//! # Design decisions
19//!
20//! ## Why two routing mechanisms?
21//!
22//! The message bus provides typed and Any-based routing to balance performance
23//! and flexibility:
24//!
25//! **Typed routing** optimizes for throughput on known data types:
26//! - `TopicRouter<T>` for pub/sub, `EndpointMap<T>` for point-to-point.
27//! - Handlers implement `Handler<T>`, receive `&T` directly.
28//! - No runtime type checking enables inlining and static dispatch.
29//! - Built-in routers: `QuoteTick`, `TradeTick`, `Bar`, `OrderBookDeltas`,
30//!   `OrderBookDepth`, `OrderEventAny`, `PositionEvent`, `AccountState`.
31//!
32//! **Any-based routing** provides flexibility for extensibility:
33//! - `subscriptions`/`topics` maps with `ShareableMessageHandler`.
34//! - Handlers implement `Handler<dyn Any>`, receive `&dyn Any`.
35//! - Supports arbitrary message types without modifying the bus.
36//! - Required for Python interop where types aren't known at compile time.
37//!
38//! ## Handler semantics
39//!
40//! **Typed handlers receive `&T` references:**
41//! - Same message delivered to N handlers without cloning.
42//! - Handler decides whether to clone (only if storing).
43//! - Zero-cost for `Copy` types (`QuoteTick`, `TradeTick`, `Bar`).
44//! - Efficient for large types (`OrderBookDeltas`).
45//!
46//! **Any-based handlers pay per-handler overhead:**
47//! - Each handler receives `&dyn Any`, must downcast to `&T`.
48//! - N handlers = N downcasts + N potential clones.
49//! - Runtime type checking on every dispatch.
50//!
51//! ## Performance trade-off
52//!
53//! Typed routing is faster (see `benches/msgbus_typed.rs`, AMD Ryzen 9 7950X):
54//!
55//! | Scenario                    | Typed vs Any |
56//! |-----------------------------|--------------|
57//! | Handler dispatch (noop)     | ~10x faster  |
58//! | Router with 5 subscribers   | ~3.5x faster |
59//! | Router with 10 subscribers  | ~2x faster   |
60//! | High volume (1M messages)   | ~7% faster   |
61//!
62//! Any-based routing pays for flexibility with runtime type checking. Use
63//! typed routing for hot-path data; Any-based for custom types and Python.
64//!
65//! ## Routing paths are separate
66//!
67//! Typed and Any-based routing use separate data structures:
68//! - `publish_quote` routes through `router_quotes`.
69//! - `publish_any` routes through `topics`.
70//!
71//! Publishers and subscribers must use matching APIs. Mixing them causes
72//! silent message loss.
73//!
74//! ## When to use each
75//!
76//! **Typed** (`publish_quote`, `subscribe_quotes`, etc.):
77//! - Market data (quotes, trades, bars, order book updates).
78//! - Order and position events.
79//! - High-frequency data with known types.
80//!
81//! **Any-based** (`publish_any`, `subscribe_any`):
82//! - Custom or user-defined data types.
83//! - Low-frequency messages.
84//! - Python callbacks.
85
86use std::{
87    any::{Any, TypeId},
88    cell::RefCell,
89    collections::HashMap,
90    fmt::Debug,
91    hash::{Hash, Hasher},
92    rc::Rc,
93};
94
95use ahash::{AHashMap, AHashSet};
96use indexmap::IndexMap;
97use nautilus_core::UUID4;
98use nautilus_model::{
99    data::{
100        Bar, Data, FundingRateUpdate, GreeksData, IndexPriceUpdate, MarkPriceUpdate,
101        OrderBookDeltas, OrderBookDepth, QuoteTick, TradeTick,
102        option_chain::{OptionChainSlice, OptionGreeks},
103    },
104    events::{AccountState, OrderEventAny, PortfolioSnapshot, PositionEvent},
105    identifiers::TraderId,
106    instruments::InstrumentAny,
107    orderbook::OrderBook,
108    orders::OrderAny,
109    position::Position,
110};
111use smallvec::SmallVec;
112use ustr::Ustr;
113
114use super::{
115    HAS_EXTERNAL_EGRESS, ShareableMessageHandler,
116    backing::MessageBusExternalEgress,
117    config::MessageBusConfig,
118    matching::is_matching_backtracking,
119    message::{BusPayloadCategory, BusPayloadType},
120    mstr::{Endpoint, MStr, Pattern, Topic},
121    set_message_bus,
122    switchboard::MessagingSwitchboard,
123    typed_endpoints::{EndpointMap, IntoEndpointMap},
124    typed_router::TopicRouter,
125};
126use crate::{
127    enums::SerializationEncoding,
128    messages::{
129        data::{DataCommand, DataResponse},
130        execution::{ExecutionReport, TradingCommand},
131    },
132};
133
134/// Represents a subscription to a particular topic.
135///
136/// This is an internal class intended to be used by the message bus to organize
137/// topics and their subscribers.
138#[derive(Clone, Debug)]
139pub struct Subscription {
140    /// The shareable message handler for the subscription.
141    pub handler: ShareableMessageHandler,
142    /// Store a copy of the handler ID for faster equality checks.
143    pub handler_id: Ustr,
144    /// The pattern for the subscription.
145    pub pattern: MStr<Pattern>,
146    /// The priority for the subscription determines the ordering of handlers receiving
147    /// messages being processed, higher priority handlers will receive messages before
148    /// lower priority handlers.
149    pub priority: u32,
150}
151
152impl Subscription {
153    /// Creates a new [`Subscription`] instance.
154    #[must_use]
155    pub fn new(
156        pattern: MStr<Pattern>,
157        handler: ShareableMessageHandler,
158        priority: Option<u32>,
159    ) -> Self {
160        Self {
161            handler_id: handler.0.id(),
162            pattern,
163            handler,
164            priority: priority.unwrap_or(0),
165        }
166    }
167
168    pub(crate) fn delivery_order(&self, other: &Self) -> std::cmp::Ordering {
169        other
170            .priority
171            .cmp(&self.priority)
172            .then_with(|| self.pattern.cmp(&other.pattern))
173            .then_with(|| self.handler_id.cmp(&other.handler_id))
174    }
175}
176
177impl PartialEq<Self> for Subscription {
178    fn eq(&self, other: &Self) -> bool {
179        self.pattern == other.pattern && self.handler_id == other.handler_id
180    }
181}
182
183impl Eq for Subscription {}
184
185impl PartialOrd for Subscription {
186    fn partial_cmp(&self, other: &Self) -> Option<std::cmp::Ordering> {
187        Some(self.cmp(other))
188    }
189}
190
191impl Ord for Subscription {
192    fn cmp(&self, other: &Self) -> std::cmp::Ordering {
193        self.pattern
194            .cmp(&other.pattern)
195            .then_with(|| self.handler_id.cmp(&other.handler_id))
196    }
197}
198
199impl Hash for Subscription {
200    fn hash<H: Hasher>(&self, state: &mut H) {
201        self.pattern.hash(state);
202        self.handler_id.hash(state);
203    }
204}
205
206/// A generic message bus to facilitate various messaging patterns.
207///
208/// The bus provides both a producer and consumer API for Pub/Sub, Req/Rep, as
209/// well as direct point-to-point messaging to registered endpoints.
210///
211/// Pub/Sub wildcard patterns for hierarchical topics are possible:
212///  - `*` asterisk represents one or more characters in a pattern.
213///  - `?` question mark represents a single character in a pattern.
214///
215/// Given a topic and pattern potentially containing wildcard characters, i.e.
216/// `*` and `?`, where `?` can match any single character in the topic, and `*`
217/// can match any number of characters including zero characters.
218///
219/// The asterisk in a wildcard matches any character zero or more times. For
220/// example, `comp*` matches anything beginning with `comp` which means `comp`,
221/// `complete`, and `computer` are all matched.
222///
223/// A question mark matches a single character once. For example, `c?mp` matches
224/// `camp` and `comp`. The question mark can also be used more than once.
225/// For example, `c??p` would match both of the above examples and `coop`.
226pub struct MessageBus {
227    /// The trader ID associated with the message bus.
228    pub trader_id: TraderId,
229    /// The instance ID associated with the message bus.
230    pub instance_id: UUID4,
231    /// The name for the message bus.
232    pub name: String,
233    /// If the message bus is backed by a database.
234    pub has_backing: bool,
235    pub(crate) switchboard: MessagingSwitchboard,
236    pub(crate) subscriptions: AHashSet<Subscription>,
237    pub(crate) topics: IndexMap<MStr<Topic>, Vec<Subscription>>,
238    pub(crate) endpoints: IndexMap<MStr<Endpoint>, ShareableMessageHandler>,
239    pub(crate) correlation_index: AHashMap<UUID4, ShareableMessageHandler>,
240    pub(crate) router_quotes: TopicRouter<QuoteTick>,
241    pub(crate) router_trades: TopicRouter<TradeTick>,
242    pub(crate) router_bars: TopicRouter<Bar>,
243    pub(crate) router_deltas: TopicRouter<OrderBookDeltas>,
244    pub(crate) router_depth: TopicRouter<OrderBookDepth>,
245    pub(crate) router_book_snapshots: TopicRouter<OrderBook>,
246    pub(crate) router_mark_prices: TopicRouter<MarkPriceUpdate>,
247    pub(crate) router_index_prices: TopicRouter<IndexPriceUpdate>,
248    pub(crate) router_funding_rates: TopicRouter<FundingRateUpdate>,
249    pub(crate) router_order_events: TopicRouter<OrderEventAny>,
250    pub(crate) router_position_events: TopicRouter<PositionEvent>,
251    pub(crate) router_account_state: TopicRouter<AccountState>,
252    pub(crate) router_orders: TopicRouter<OrderAny>,
253    pub(crate) router_positions: TopicRouter<Position>,
254    pub(crate) router_portfolio: TopicRouter<PortfolioSnapshot>,
255    pub(crate) router_greeks: TopicRouter<GreeksData>,
256    pub(crate) router_option_greeks: TopicRouter<OptionGreeks>,
257    pub(crate) router_option_chain: TopicRouter<OptionChainSlice>,
258    pub(crate) router_instruments: TopicRouter<InstrumentAny>,
259    #[cfg(feature = "defi")]
260    pub(crate) router_defi_blocks: TopicRouter<nautilus_model::defi::Block>, // nautilus-import-ok
261    #[cfg(feature = "defi")]
262    pub(crate) router_defi_pools: TopicRouter<nautilus_model::defi::Pool>, // nautilus-import-ok
263    #[cfg(feature = "defi")]
264    pub(crate) router_defi_swaps: TopicRouter<nautilus_model::defi::PoolSwap>, // nautilus-import-ok
265    #[cfg(feature = "defi")]
266    pub(crate) router_defi_liquidity: TopicRouter<nautilus_model::defi::PoolLiquidityUpdate>, // nautilus-import-ok
267    #[cfg(feature = "defi")]
268    pub(crate) router_defi_collects: TopicRouter<nautilus_model::defi::PoolFeeCollect>, // nautilus-import-ok
269    #[cfg(feature = "defi")]
270    pub(crate) router_defi_flash: TopicRouter<nautilus_model::defi::PoolFlash>, // nautilus-import-ok
271    #[cfg(feature = "defi")]
272    pub(crate) endpoints_defi_data: IntoEndpointMap<nautilus_model::defi::DefiData>, // nautilus-import-ok
273    pub(crate) endpoints_quotes: EndpointMap<QuoteTick>,
274    pub(crate) endpoints_trades: EndpointMap<TradeTick>,
275    pub(crate) endpoints_bars: EndpointMap<Bar>,
276    pub(crate) endpoints_account_state: EndpointMap<AccountState>,
277    pub(crate) endpoints_trading_commands: IntoEndpointMap<TradingCommand>,
278    pub(crate) endpoints_data_commands: IntoEndpointMap<DataCommand>,
279    pub(crate) endpoints_data_responses: IntoEndpointMap<DataResponse>,
280    pub(crate) endpoints_exec_reports: IntoEndpointMap<ExecutionReport>,
281    pub(crate) endpoints_order_events: IntoEndpointMap<OrderEventAny>,
282    pub(crate) endpoints_data: IntoEndpointMap<Data>,
283    routers_typed: AHashMap<TypeId, Box<dyn Any>>,
284    endpoints_typed: AHashMap<TypeId, Box<dyn Any>>,
285    sent_count: u64,
286    req_count: u64,
287    res_count: u64,
288    pub_count: u64,
289    external_egress: Option<Rc<RefCell<Box<dyn MessageBusExternalEgress>>>>,
290    has_external_streams: bool,
291    encoding: SerializationEncoding,
292    encoding_market_data: Option<SerializationEncoding>,
293    encoding_builtin: Option<SerializationEncoding>,
294    types_filter: AHashSet<BusPayloadType>,
295    streaming_types: AHashSet<BusPayloadType>,
296}
297
298impl Debug for MessageBus {
299    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
300        f.debug_struct(stringify!(MessageBus))
301            .field("trader_id", &self.trader_id)
302            .field("instance_id", &self.instance_id)
303            .field("name", &self.name)
304            .field("has_backing", &self.has_backing)
305            .field("external_egress", &self.external_egress.is_some())
306            .finish_non_exhaustive()
307    }
308}
309
310impl Default for MessageBus {
311    /// Creates a new default [`MessageBus`] instance.
312    fn default() -> Self {
313        Self::new(TraderId::from("TRADER-001"), UUID4::new(), None, None)
314    }
315}
316
317impl MessageBus {
318    /// Creates a new [`MessageBus`] instance.
319    #[must_use]
320    pub fn new(
321        trader_id: TraderId,
322        instance_id: UUID4,
323        name: Option<String>,
324        _config: Option<HashMap<String, serde_json::Value>>,
325    ) -> Self {
326        Self {
327            trader_id,
328            instance_id,
329            name: name.unwrap_or(stringify!(MessageBus).to_owned()),
330            switchboard: MessagingSwitchboard::default(),
331            subscriptions: AHashSet::new(),
332            topics: IndexMap::new(),
333            endpoints: IndexMap::new(),
334            correlation_index: AHashMap::new(),
335            has_backing: false,
336            router_quotes: TopicRouter::new(),
337            router_trades: TopicRouter::new(),
338            router_bars: TopicRouter::new(),
339            router_deltas: TopicRouter::new(),
340            router_depth: TopicRouter::new(),
341            router_book_snapshots: TopicRouter::new(),
342            router_mark_prices: TopicRouter::new(),
343            router_index_prices: TopicRouter::new(),
344            router_funding_rates: TopicRouter::new(),
345            router_order_events: TopicRouter::new(),
346            router_position_events: TopicRouter::new(),
347            router_account_state: TopicRouter::new(),
348            router_portfolio: TopicRouter::new(),
349            router_orders: TopicRouter::new(),
350            router_positions: TopicRouter::new(),
351            router_greeks: TopicRouter::new(),
352            router_option_greeks: TopicRouter::new(),
353            router_option_chain: TopicRouter::new(),
354            router_instruments: TopicRouter::new(),
355            #[cfg(feature = "defi")]
356            router_defi_blocks: TopicRouter::new(),
357            #[cfg(feature = "defi")]
358            router_defi_pools: TopicRouter::new(),
359            #[cfg(feature = "defi")]
360            router_defi_swaps: TopicRouter::new(),
361            #[cfg(feature = "defi")]
362            router_defi_liquidity: TopicRouter::new(),
363            #[cfg(feature = "defi")]
364            router_defi_collects: TopicRouter::new(),
365            #[cfg(feature = "defi")]
366            router_defi_flash: TopicRouter::new(),
367            #[cfg(feature = "defi")]
368            endpoints_defi_data: IntoEndpointMap::new(),
369            endpoints_quotes: EndpointMap::new(),
370            endpoints_trades: EndpointMap::new(),
371            endpoints_bars: EndpointMap::new(),
372            endpoints_account_state: EndpointMap::new(),
373            endpoints_trading_commands: IntoEndpointMap::new(),
374            endpoints_data_commands: IntoEndpointMap::new(),
375            endpoints_data_responses: IntoEndpointMap::new(),
376            endpoints_exec_reports: IntoEndpointMap::new(),
377            endpoints_order_events: IntoEndpointMap::new(),
378            endpoints_data: IntoEndpointMap::new(),
379            routers_typed: AHashMap::new(),
380            endpoints_typed: AHashMap::new(),
381            sent_count: 0,
382            req_count: 0,
383            res_count: 0,
384            pub_count: 0,
385            external_egress: None,
386            has_external_streams: false,
387            encoding: SerializationEncoding::Json,
388            encoding_market_data: None,
389            encoding_builtin: None,
390            types_filter: AHashSet::new(),
391            streaming_types: AHashSet::new(),
392        }
393    }
394
395    /// Registers message bus for the current thread.
396    pub fn register_message_bus(self) -> Rc<RefCell<Self>> {
397        let msgbus = Rc::new(RefCell::new(self));
398        set_message_bus(msgbus.clone());
399        msgbus
400    }
401
402    /// Gets or creates a typed router for custom message type `T`.
403    ///
404    /// # Panics
405    ///
406    /// Panics if the stored router type doesn't match `T` (internal bug).
407    pub fn router<T: 'static>(&mut self) -> &mut TopicRouter<T> {
408        self.routers_typed
409            .entry(TypeId::of::<T>())
410            .or_insert_with(|| Box::new(TopicRouter::<T>::new()))
411            .downcast_mut::<TopicRouter<T>>()
412            .expect("TopicRouter type mismatch - this is a bug")
413    }
414
415    /// Gets or creates a typed endpoint map for custom message type `T`.
416    ///
417    /// # Panics
418    ///
419    /// Panics if the stored endpoint map type doesn't match `T` (internal bug).
420    pub fn endpoint_map<T: 'static>(&mut self) -> &mut EndpointMap<T> {
421        self.endpoints_typed
422            .entry(TypeId::of::<T>())
423            .or_insert_with(|| Box::new(EndpointMap::<T>::new()))
424            .downcast_mut::<EndpointMap<T>>()
425            .expect("EndpointMap type mismatch - this is a bug")
426    }
427
428    /// Sets external egress for serialized published messages.
429    ///
430    /// Typed message variants remain local because this method does not configure external streams.
431    pub fn set_external_egress(
432        &mut self,
433        external_egress: Box<dyn MessageBusExternalEgress>,
434        encoding: SerializationEncoding,
435    ) {
436        self.external_egress = Some(Rc::new(RefCell::new(external_egress)));
437        self.has_external_streams = false;
438        self.encoding = encoding;
439        self.encoding_market_data = None;
440        self.encoding_builtin = None;
441        self.has_backing = true;
442        HAS_EXTERNAL_EGRESS.with(|flag| flag.set(true));
443    }
444
445    /// Sets external egress and category encoding policy from a validated config.
446    ///
447    /// Typed message variants are enabled only when `config.external_streams` is non-empty.
448    ///
449    /// # Errors
450    ///
451    /// Returns an error if the config selects an unsupported encoding.
452    pub fn set_external_egress_config(
453        &mut self,
454        external_egress: Box<dyn MessageBusExternalEgress>,
455        config: &MessageBusConfig,
456    ) -> crate::config::ConfigResult<()> {
457        config.validate()?;
458
459        self.external_egress = Some(Rc::new(RefCell::new(external_egress)));
460        self.has_external_streams = config
461            .external_streams
462            .as_ref()
463            .is_some_and(|streams| !streams.is_empty());
464        self.encoding = config.encoding;
465        self.encoding_market_data = config.encoding_market_data;
466        self.encoding_builtin = config.encoding_builtin;
467        self.has_backing = true;
468        HAS_EXTERNAL_EGRESS.with(|flag| flag.set(true));
469
470        Ok(())
471    }
472
473    /// Sets the payload type names excluded from external publishing.
474    pub fn set_types_filter(&mut self, filter: Vec<String>) {
475        self.types_filter.clear();
476
477        for type_name in filter {
478            let payload_type = BusPayloadType::from_name(&type_name);
479            if !payload_type.as_str().is_empty() {
480                self.types_filter.insert(payload_type);
481            }
482
483            if let Some(payload_type) = BusPayloadType::from_typed_name(&type_name) {
484                self.types_filter.insert(payload_type);
485            }
486        }
487    }
488
489    /// Registers a payload type for external-to-internal streaming.
490    pub fn add_streaming_type(&mut self, payload_type: BusPayloadType) {
491        if !payload_type.as_str().is_empty() {
492            self.streaming_types.insert(payload_type);
493        }
494    }
495
496    /// Returns whether the payload type is registered for external-to-internal streaming.
497    #[must_use]
498    pub fn is_streaming_type(&self, payload_type: BusPayloadType) -> bool {
499        !payload_type.as_str().is_empty() && self.streaming_types.contains(&payload_type)
500    }
501
502    /// Clears all payload types registered for external-to-internal streaming.
503    pub fn clear_streaming_types(&mut self) {
504        self.streaming_types.clear();
505    }
506
507    #[must_use]
508    pub(crate) fn has_external_egress(&self) -> bool {
509        self.external_egress.is_some()
510    }
511
512    pub(crate) fn has_external_streams(&self) -> bool {
513        self.has_external_streams
514    }
515
516    pub(crate) fn external_egress(&self) -> Option<Rc<RefCell<Box<dyn MessageBusExternalEgress>>>> {
517        self.external_egress.clone()
518    }
519
520    pub(crate) fn encoding_for(&self, payload_type: BusPayloadType) -> SerializationEncoding {
521        match payload_type.category() {
522            BusPayloadCategory::MarketData => self.encoding_market_data.unwrap_or(self.encoding),
523            BusPayloadCategory::BuiltIn => self.encoding_builtin.unwrap_or(self.encoding),
524            BusPayloadCategory::Other => self.encoding,
525        }
526    }
527
528    pub(crate) fn types_filter(&self) -> &AHashSet<BusPayloadType> {
529        &self.types_filter
530    }
531
532    /// Disposes of the message bus, clearing all subscriptions, endpoints,
533    /// and handler references.
534    pub fn dispose(&mut self) {
535        self.subscriptions.clear();
536        self.topics.clear();
537        self.endpoints.clear();
538        self.correlation_index.clear();
539
540        self.router_quotes.clear();
541        self.router_trades.clear();
542        self.router_bars.clear();
543        self.router_deltas.clear();
544        self.router_depth.clear();
545        self.router_book_snapshots.clear();
546        self.router_mark_prices.clear();
547        self.router_index_prices.clear();
548        self.router_funding_rates.clear();
549        self.router_order_events.clear();
550        self.router_position_events.clear();
551        self.router_account_state.clear();
552        self.router_portfolio.clear();
553        self.router_orders.clear();
554        self.router_positions.clear();
555        self.router_greeks.clear();
556        self.router_option_greeks.clear();
557        self.router_option_chain.clear();
558        self.router_instruments.clear();
559
560        #[cfg(feature = "defi")]
561        {
562            self.router_defi_blocks.clear();
563            self.router_defi_pools.clear();
564            self.router_defi_swaps.clear();
565            self.router_defi_liquidity.clear();
566            self.router_defi_collects.clear();
567            self.router_defi_flash.clear();
568            self.endpoints_defi_data.clear();
569        }
570
571        self.endpoints_quotes.clear();
572        self.endpoints_trades.clear();
573        self.endpoints_bars.clear();
574        self.endpoints_account_state.clear();
575        self.endpoints_trading_commands.clear();
576        self.endpoints_data_commands.clear();
577        self.endpoints_data_responses.clear();
578        self.endpoints_exec_reports.clear();
579        self.endpoints_order_events.clear();
580        self.endpoints_data.clear();
581        self.routers_typed.clear();
582        self.endpoints_typed.clear();
583        self.clear_streaming_types();
584        self.sent_count = 0;
585        self.req_count = 0;
586        self.res_count = 0;
587        self.pub_count = 0;
588
589        if let Some(external_egress) = self.external_egress.take() {
590            external_egress.borrow_mut().close();
591        }
592        self.has_external_streams = false;
593        self.has_backing = false;
594        HAS_EXTERNAL_EGRESS.with(|flag| flag.set(false));
595    }
596
597    /// Returns the memory address of this instance as a hexadecimal string.
598    #[must_use]
599    pub fn mem_address(&self) -> String {
600        format!("{self:p}")
601    }
602
603    /// Returns a reference to the switchboard.
604    #[must_use]
605    pub fn switchboard(&self) -> &MessagingSwitchboard {
606        &self.switchboard
607    }
608
609    /// Returns the total count of messages sent to endpoints.
610    #[must_use]
611    pub const fn sent_count(&self) -> u64 {
612        self.sent_count
613    }
614
615    /// Returns the total count of requests sent to endpoints.
616    #[must_use]
617    pub const fn req_count(&self) -> u64 {
618        self.req_count
619    }
620
621    /// Returns the total count of responses sent to registered handlers.
622    #[must_use]
623    pub const fn res_count(&self) -> u64 {
624        self.res_count
625    }
626
627    /// Returns the total count of messages published to topics.
628    #[must_use]
629    pub const fn pub_count(&self) -> u64 {
630        self.pub_count
631    }
632
633    pub(crate) fn increment_sent_count(&mut self) {
634        self.sent_count += 1;
635    }
636
637    pub(crate) fn increment_req_count(&mut self) {
638        self.req_count += 1;
639    }
640
641    pub(crate) fn increment_res_count(&mut self) {
642        self.res_count += 1;
643    }
644
645    pub(crate) fn increment_pub_count(&mut self) {
646        self.pub_count += 1;
647    }
648
649    /// Returns the registered endpoint addresses.
650    #[must_use]
651    pub fn endpoints(&self) -> Vec<&str> {
652        self.endpoints.iter().map(|e| e.0.as_str()).collect()
653    }
654
655    /// Returns actively subscribed patterns.
656    #[must_use]
657    pub fn patterns(&self) -> Vec<&str> {
658        self.subscriptions
659            .iter()
660            .map(|s| s.pattern.as_str())
661            .collect()
662    }
663
664    /// Returns whether there are subscribers for the `topic`.
665    ///
666    /// # Errors
667    ///
668    /// Returns an error if the `topic` is not a valid topic string.
669    pub fn has_subscribers<T: AsRef<str>>(&self, topic: T) -> anyhow::Result<bool> {
670        Ok(self.subscriptions_count(topic)? > 0)
671    }
672
673    /// Returns the count of subscribers for the `topic`.
674    ///
675    /// # Errors
676    ///
677    /// Returns an error if the `topic` is not a valid topic string.
678    pub fn subscriptions_count<T: AsRef<str>>(&self, topic: T) -> anyhow::Result<usize> {
679        let topic = MStr::<Topic>::topic(topic)?;
680        Ok(self
681            .topics
682            .get(&topic)
683            .map_or_else(|| self.find_topic_matches(topic).len(), Vec::len))
684    }
685
686    /// Returns active subscriptions.
687    #[must_use]
688    pub fn subscriptions(&self) -> Vec<&Subscription> {
689        self.subscriptions.iter().collect()
690    }
691
692    /// Returns the handler IDs for actively subscribed patterns.
693    #[must_use]
694    pub fn subscription_handler_ids(&self) -> Vec<&str> {
695        self.subscriptions
696            .iter()
697            .map(|s| s.handler_id.as_str())
698            .collect()
699    }
700
701    /// Returns whether the endpoint is registered.
702    ///
703    /// # Panics
704    ///
705    /// Panics if the `endpoint` conversion to `MStr<Endpoint>` fails.
706    #[must_use]
707    pub fn is_registered<T: Into<MStr<Endpoint>>>(&self, endpoint: T) -> bool {
708        let endpoint: MStr<Endpoint> = endpoint.into();
709        self.endpoints.contains_key(&endpoint)
710    }
711
712    /// Returns whether the `handler` is subscribed to the `pattern`.
713    #[must_use]
714    pub fn is_subscribed<T: AsRef<str>>(
715        &self,
716        pattern: T,
717        handler: ShareableMessageHandler,
718    ) -> bool {
719        let pattern = MStr::<Pattern>::pattern(pattern);
720        let sub = Subscription::new(pattern, handler, None);
721        self.subscriptions.contains(&sub)
722    }
723
724    /// Close the message bus which will close the sender channel and join the thread.
725    ///
726    /// This function does not return an error.
727    #[allow(
728        clippy::missing_errors_doc,
729        reason = "closing the in-memory message bus returns no errors"
730    )]
731    pub fn close(&mut self) -> anyhow::Result<()> {
732        if let Some(external_egress) = self.external_egress.take() {
733            external_egress.borrow_mut().close();
734        }
735        self.has_external_streams = false;
736        self.has_backing = false;
737        HAS_EXTERNAL_EGRESS.with(|flag| flag.set(false));
738        Ok(())
739    }
740
741    /// Returns the handler for the `endpoint`.
742    #[must_use]
743    pub fn get_endpoint(&self, endpoint: MStr<Endpoint>) -> Option<&ShareableMessageHandler> {
744        self.endpoints.get(&endpoint)
745    }
746
747    /// Returns the handler for the `correlation_id`.
748    #[must_use]
749    pub fn get_response_handler(&self, correlation_id: &UUID4) -> Option<&ShareableMessageHandler> {
750        self.correlation_index.get(correlation_id)
751    }
752
753    /// Removes and returns the handler for the `correlation_id`.
754    pub(crate) fn take_response_handler(
755        &mut self,
756        correlation_id: &UUID4,
757    ) -> Option<ShareableMessageHandler> {
758        self.correlation_index.remove(correlation_id)
759    }
760
761    /// Finds the subscriptions with pattern matching the `topic`.
762    fn find_topic_matches(&self, topic: MStr<Topic>) -> Vec<Subscription> {
763        self.subscriptions
764            .iter()
765            .filter_map(|sub| {
766                if is_matching_backtracking(topic, sub.pattern) {
767                    Some(sub.clone())
768                } else {
769                    None
770                }
771            })
772            .collect()
773    }
774
775    /// Finds the subscriptions which match the `topic` and caches the
776    /// results in the `patterns` map.
777    #[must_use]
778    pub fn matching_subscriptions<T: Into<MStr<Topic>>>(&mut self, topic: T) -> Vec<Subscription> {
779        self.inner_matching_subscriptions(topic.into())
780    }
781
782    fn inner_matching_subscriptions(&mut self, topic: MStr<Topic>) -> Vec<Subscription> {
783        self.topics.get(&topic).cloned().unwrap_or_else(|| {
784            let mut matches = self.find_topic_matches(topic);
785            matches.sort_by(Subscription::delivery_order);
786            self.topics.insert(topic, matches.clone());
787            matches
788        })
789    }
790
791    /// Fills a buffer with handlers matching a topic.
792    pub(crate) fn fill_matching_any_handlers(
793        &mut self,
794        topic: MStr<Topic>,
795        buf: &mut SmallVec<[ShareableMessageHandler; 64]>,
796    ) {
797        if let Some(subs) = self.topics.get(&topic) {
798            for sub in subs {
799                buf.push(sub.handler.clone());
800            }
801        } else {
802            let mut matches = self.find_topic_matches(topic);
803            matches.sort_by(Subscription::delivery_order);
804
805            for sub in &matches {
806                buf.push(sub.handler.clone());
807            }
808
809            self.topics.insert(topic, matches);
810        }
811    }
812
813    /// Registers a response handler for a specific correlation ID.
814    ///
815    /// # Errors
816    ///
817    /// Returns an error if `handler` is already registered for the `correlation_id`.
818    pub fn register_response_handler(
819        &mut self,
820        correlation_id: &UUID4,
821        handler: ShareableMessageHandler,
822    ) -> anyhow::Result<()> {
823        if self.correlation_index.contains_key(correlation_id) {
824            anyhow::bail!("Correlation ID <{correlation_id}> already has a registered handler");
825        }
826
827        self.correlation_index.insert(*correlation_id, handler);
828
829        Ok(())
830    }
831
832    pub(crate) fn unsubscribe_any(
833        &mut self,
834        pattern: MStr<Pattern>,
835        handler: &ShareableMessageHandler,
836    ) {
837        log::debug!("Unsubscribing {handler:?} from pattern '{pattern}'");
838
839        let handler_id = handler.0.id();
840
841        let count_before = self.subscriptions.len();
842
843        self.topics.values_mut().for_each(|subs| {
844            subs.retain(|s| !(s.pattern == pattern && s.handler_id == handler_id));
845        });
846
847        self.subscriptions
848            .retain(|s| !(s.pattern == pattern && s.handler_id == handler_id));
849
850        let removed = self.subscriptions.len() < count_before;
851
852        if removed {
853            log::debug!("Handler for pattern '{pattern}' was removed");
854        } else {
855            log::debug!("No matching handler for pattern '{pattern}' was found");
856        }
857    }
858}
859
860#[cfg(test)]
861mod tests {
862    use std::{
863        any::Any,
864        cell::RefCell,
865        collections::hash_map::DefaultHasher,
866        fmt::Debug,
867        hash::{Hash, Hasher},
868        rc::Rc,
869    };
870
871    use rand::{RngExt, SeedableRng, rngs::StdRng};
872    use rstest::rstest;
873    use ustr::Ustr;
874
875    use super::*;
876    use crate::msgbus::{
877        self, Handler, ShareableMessageHandler, get_message_bus,
878        matching::is_matching_backtracking,
879        stubs::{get_any_saving_handler, get_call_check_handler, get_stub_shareable_handler},
880        subscriptions_count_any,
881        typed_handler::shareable_handler,
882    };
883
884    #[derive(Debug)]
885    struct RecordingAnyHandler {
886        id: Ustr,
887        label: &'static str,
888        order: Rc<RefCell<Vec<&'static str>>>,
889    }
890
891    impl Handler<dyn Any> for RecordingAnyHandler {
892        fn id(&self) -> Ustr {
893            self.id
894        }
895
896        fn handle(&self, _message: &dyn Any) {
897            self.order.borrow_mut().push(self.label);
898        }
899    }
900
901    fn recording_any_handler(
902        id: &'static str,
903        label: &'static str,
904        order: Rc<RefCell<Vec<&'static str>>>,
905    ) -> ShareableMessageHandler {
906        shareable_handler(Rc::new(RecordingAnyHandler {
907            id: Ustr::from(id),
908            label,
909            order,
910        }))
911    }
912
913    #[rstest]
914    fn test_subscription_ordering_laws() {
915        let handler = get_stub_shareable_handler(Some(Ustr::from("handler-b")));
916        let base = Subscription::new("pattern-b".into(), handler.clone(), Some(1));
917        let different_priority = Subscription::new("pattern-b".into(), handler, Some(2));
918        let different_pattern = Subscription::new(
919            "pattern-a".into(),
920            get_stub_shareable_handler(Some(Ustr::from("handler-b"))),
921            Some(1),
922        );
923        let different_handler = Subscription::new(
924            "pattern-b".into(),
925            get_stub_shareable_handler(Some(Ustr::from("handler-a"))),
926            Some(1),
927        );
928
929        assert_eq!(base, different_priority);
930        assert_eq!(base.cmp(&different_priority), std::cmp::Ordering::Equal);
931
932        let mut base_hasher = DefaultHasher::new();
933        base.hash(&mut base_hasher);
934        let mut different_priority_hasher = DefaultHasher::new();
935        different_priority.hash(&mut different_priority_hasher);
936        assert_eq!(base_hasher.finish(), different_priority_hasher.finish());
937
938        let variants = [
939            base,
940            different_priority,
941            different_pattern,
942            different_handler,
943        ];
944
945        for a in &variants {
946            for b in &variants {
947                assert_eq!(a == b, a.cmp(b).is_eq());
948                assert_eq!(a.partial_cmp(b), Some(a.cmp(b)));
949                assert_eq!(a.cmp(b), b.cmp(a).reverse());
950            }
951        }
952    }
953
954    #[rstest]
955    fn test_new() {
956        let trader_id = TraderId::default();
957        let msgbus = MessageBus::new(trader_id, UUID4::new(), None, None);
958
959        assert_eq!(msgbus.trader_id, trader_id);
960        assert_eq!(msgbus.name, stringify!(MessageBus));
961    }
962
963    #[rstest]
964    fn encoding_for_uses_market_data_override() {
965        let msgbus = MessageBus {
966            encoding: SerializationEncoding::Json,
967            encoding_market_data: Some(SerializationEncoding::MsgPack),
968            ..Default::default()
969        };
970
971        assert_eq!(
972            msgbus.encoding_for(BusPayloadType::QuoteTick),
973            SerializationEncoding::MsgPack
974        );
975        assert_eq!(
976            msgbus.encoding_for(BusPayloadType::Custom(Ustr::from("CustomPayload"))),
977            SerializationEncoding::Json
978        );
979    }
980
981    #[rstest]
982    fn encoding_for_uses_builtin_override() {
983        let msgbus = MessageBus {
984            encoding: SerializationEncoding::Json,
985            encoding_builtin: Some(SerializationEncoding::MsgPack),
986            ..Default::default()
987        };
988
989        assert_eq!(
990            msgbus.encoding_for(BusPayloadType::OrderEvent),
991            SerializationEncoding::MsgPack
992        );
993        assert_eq!(
994            msgbus.encoding_for(BusPayloadType::Instrument),
995            SerializationEncoding::Json
996        );
997    }
998
999    #[rstest]
1000    fn encoding_for_uses_default_without_category_override() {
1001        let msgbus = MessageBus {
1002            encoding: SerializationEncoding::MsgPack,
1003            ..Default::default()
1004        };
1005
1006        assert_eq!(
1007            msgbus.encoding_for(BusPayloadType::QuoteTick),
1008            SerializationEncoding::MsgPack
1009        );
1010        assert_eq!(
1011            msgbus.encoding_for(BusPayloadType::OrderEvent),
1012            SerializationEncoding::MsgPack
1013        );
1014        assert_eq!(
1015            msgbus.encoding_for(BusPayloadType::Custom(Ustr::from("CustomPayload"))),
1016            SerializationEncoding::MsgPack
1017        );
1018    }
1019
1020    #[rstest]
1021    fn set_types_filter_resolves_untyped_and_typed_names() {
1022        let mut msgbus = MessageBus::default();
1023
1024        msgbus.set_types_filter(vec![
1025            "QuoteTick".to_string(),
1026            "ExternalCustomPayload".to_string(),
1027            "OrderStatusReport".to_string(),
1028            String::new(),
1029        ]);
1030
1031        let filter = msgbus.types_filter();
1032        assert_eq!(filter.len(), 4);
1033        assert!(filter.contains(&BusPayloadType::QuoteTick));
1034        assert!(filter.contains(&BusPayloadType::Custom(Ustr::from("ExternalCustomPayload"))));
1035        assert!(filter.contains(&BusPayloadType::Custom(Ustr::from("OrderStatusReport"))));
1036        assert!(filter.contains(&BusPayloadType::OrderStatusReport));
1037        assert!(!filter.contains(&BusPayloadType::Custom(Ustr::default())));
1038    }
1039
1040    #[rstest]
1041    fn streaming_type_registration_uses_canonical_payload_names() {
1042        let mut msgbus = MessageBus::default();
1043
1044        msgbus.add_streaming_type(BusPayloadType::QuoteTick);
1045        msgbus.add_streaming_type(BusPayloadType::Custom(Ustr::from("CustomPayload")));
1046
1047        assert!(msgbus.is_streaming_type(BusPayloadType::QuoteTick));
1048        assert!(msgbus.is_streaming_type(BusPayloadType::Custom(Ustr::from("CustomPayload"))));
1049        assert!(msgbus.streaming_types.contains(&BusPayloadType::QuoteTick));
1050        assert!(
1051            msgbus
1052                .streaming_types
1053                .contains(&BusPayloadType::Custom(Ustr::from("CustomPayload")))
1054        );
1055        assert!(!msgbus.is_streaming_type(BusPayloadType::TradeTick));
1056    }
1057
1058    #[rstest]
1059    fn streaming_type_registration_ignores_empty_custom_payload_type() {
1060        let mut msgbus = MessageBus::default();
1061
1062        msgbus.add_streaming_type(BusPayloadType::Custom(Ustr::default()));
1063
1064        assert!(!msgbus.is_streaming_type(BusPayloadType::Custom(Ustr::default())));
1065        assert!(msgbus.streaming_types.is_empty());
1066    }
1067
1068    #[rstest]
1069    fn clear_streaming_types_removes_registered_types() {
1070        let mut msgbus = MessageBus::default();
1071        msgbus.add_streaming_type(BusPayloadType::QuoteTick);
1072
1073        msgbus.clear_streaming_types();
1074
1075        assert!(!msgbus.is_streaming_type(BusPayloadType::QuoteTick));
1076    }
1077
1078    #[rstest]
1079    fn dispose_clears_streaming_types() {
1080        let mut msgbus = MessageBus::default();
1081        msgbus.add_streaming_type(BusPayloadType::QuoteTick);
1082
1083        msgbus.dispose();
1084
1085        assert!(!msgbus.is_streaming_type(BusPayloadType::QuoteTick));
1086    }
1087
1088    #[rstest]
1089    fn test_dispose_resets_counters() {
1090        let mut msgbus = MessageBus::default();
1091
1092        msgbus.increment_sent_count();
1093        msgbus.increment_req_count();
1094        msgbus.increment_res_count();
1095        msgbus.increment_pub_count();
1096        msgbus.dispose();
1097
1098        assert_eq!(msgbus.sent_count(), 0);
1099        assert_eq!(msgbus.req_count(), 0);
1100        assert_eq!(msgbus.res_count(), 0);
1101        assert_eq!(msgbus.pub_count(), 0);
1102    }
1103
1104    #[rstest]
1105    fn test_endpoints_when_no_endpoints() {
1106        let msgbus = get_message_bus();
1107        assert!(msgbus.borrow().endpoints().is_empty());
1108    }
1109
1110    #[rstest]
1111    fn test_topics_when_no_subscriptions() {
1112        let msgbus = get_message_bus();
1113        assert!(msgbus.borrow().patterns().is_empty());
1114        assert!(!msgbus.borrow().has_subscribers("my-topic").unwrap());
1115    }
1116
1117    #[rstest]
1118    fn test_is_subscribed_when_no_subscriptions() {
1119        let msgbus = get_message_bus();
1120        let handler = get_stub_shareable_handler(None);
1121
1122        assert!(!msgbus.borrow().is_subscribed("my-topic", handler));
1123    }
1124
1125    #[rstest]
1126    fn test_get_response_handler_when_no_handler() {
1127        let msgbus = get_message_bus();
1128        let msgbus_ref = msgbus.borrow();
1129        let handler = msgbus_ref.get_response_handler(&UUID4::new());
1130        assert!(handler.is_none());
1131    }
1132
1133    #[rstest]
1134    fn test_get_response_handler_when_already_registered() {
1135        let msgbus = get_message_bus();
1136        let mut msgbus_ref = msgbus.borrow_mut();
1137        let handler = get_stub_shareable_handler(None);
1138
1139        let request_id = UUID4::new();
1140        msgbus_ref
1141            .register_response_handler(&request_id, handler.clone())
1142            .unwrap();
1143
1144        let result = msgbus_ref.register_response_handler(&request_id, handler);
1145        assert!(result.is_err());
1146    }
1147
1148    #[rstest]
1149    fn test_get_response_handler_when_registered() {
1150        let msgbus = get_message_bus();
1151        let mut msgbus_ref = msgbus.borrow_mut();
1152        let handler = get_stub_shareable_handler(None);
1153
1154        let request_id = UUID4::new();
1155        msgbus_ref
1156            .register_response_handler(&request_id, handler)
1157            .unwrap();
1158
1159        let handler = msgbus_ref.get_response_handler(&request_id).unwrap();
1160        assert_eq!(handler.id(), handler.id());
1161    }
1162
1163    #[rstest]
1164    fn test_take_response_handler_removes_registration_and_allows_reregistration() {
1165        let mut msgbus = MessageBus::default();
1166        let request_id = UUID4::new();
1167        let handler = get_stub_shareable_handler(None);
1168        let handler_id = handler.id();
1169        msgbus
1170            .register_response_handler(&request_id, handler)
1171            .unwrap();
1172
1173        let taken = msgbus.take_response_handler(&request_id).unwrap();
1174
1175        assert_eq!(taken.id(), handler_id);
1176        assert!(msgbus.get_response_handler(&request_id).is_none());
1177        assert!(msgbus.register_response_handler(&request_id, taken).is_ok());
1178    }
1179
1180    #[rstest]
1181    fn test_is_registered_when_no_registrations() {
1182        let msgbus = get_message_bus();
1183        assert!(!msgbus.borrow().is_registered("MyEndpoint"));
1184    }
1185
1186    #[rstest]
1187    fn test_register_endpoint() {
1188        let msgbus = get_message_bus();
1189        let endpoint = "MyEndpoint".into();
1190        let handler = get_stub_shareable_handler(None);
1191
1192        msgbus::register_any(endpoint, handler);
1193
1194        assert_eq!(msgbus.borrow().endpoints(), vec![endpoint.to_string()]);
1195        assert!(msgbus.borrow().get_endpoint(endpoint).is_some());
1196    }
1197
1198    #[rstest]
1199    fn test_endpoint_send() {
1200        let msgbus = get_message_bus();
1201        let endpoint = "MyEndpoint".into();
1202        let (handler, checker) = get_call_check_handler(None);
1203        let sent_count = msgbus.borrow().sent_count();
1204
1205        msgbus::register_any(endpoint, handler);
1206        assert!(msgbus.borrow().get_endpoint(endpoint).is_some());
1207        assert!(!checker.was_called());
1208
1209        msgbus::send_any(endpoint, &"Test Message");
1210
1211        assert!(checker.was_called());
1212        assert_eq!(msgbus.borrow().sent_count(), sent_count + 1);
1213    }
1214
1215    #[rstest]
1216    fn test_endpoint_send_value_increments_sent_count() {
1217        let msgbus = get_message_bus();
1218        let endpoint = "MyValueEndpoint".into();
1219        let (handler, checker) = get_call_check_handler(None);
1220        let sent_count = msgbus.borrow().sent_count();
1221
1222        msgbus::register_any(endpoint, handler);
1223        msgbus::send_any_value(endpoint, &"Test Message");
1224
1225        assert!(checker.was_called());
1226        assert_eq!(msgbus.borrow().sent_count(), sent_count + 1);
1227    }
1228
1229    #[rstest]
1230    fn test_publish_any_increments_publish_count() {
1231        let msgbus = get_message_bus();
1232        let topic = "my-published-topic";
1233        let (handler, checker) = get_call_check_handler(None);
1234        let pub_count = msgbus.borrow().pub_count();
1235
1236        msgbus::subscribe_any(topic.into(), handler, None);
1237        msgbus::publish_any(topic.into(), &"Test Message");
1238
1239        assert!(checker.was_called());
1240        assert_eq!(msgbus.borrow().pub_count(), pub_count + 1);
1241    }
1242
1243    #[rstest]
1244    fn test_deregsiter_endpoint() {
1245        let msgbus = get_message_bus();
1246        let endpoint = "MyEndpoint".into();
1247        let handler = get_stub_shareable_handler(None);
1248
1249        msgbus::register_any(endpoint, handler);
1250        msgbus::deregister_any(endpoint);
1251
1252        assert!(msgbus.borrow().endpoints().is_empty());
1253    }
1254
1255    #[rstest]
1256    fn test_subscribe() {
1257        let msgbus = get_message_bus();
1258        let topic = "my-topic";
1259        let handler = get_stub_shareable_handler(None);
1260
1261        msgbus::subscribe_any(topic.into(), handler, Some(1));
1262
1263        assert!(msgbus.borrow().has_subscribers(topic).unwrap());
1264        assert_eq!(msgbus.borrow().patterns(), vec![topic]);
1265    }
1266
1267    #[rstest]
1268    fn test_unsubscribe() {
1269        let msgbus = get_message_bus();
1270        let topic = "my-topic";
1271        let handler = get_stub_shareable_handler(None);
1272
1273        msgbus::subscribe_any(topic.into(), handler.clone(), None);
1274        msgbus::unsubscribe_any(topic.into(), &handler);
1275
1276        assert!(!msgbus.borrow().has_subscribers(topic).unwrap());
1277        assert!(msgbus.borrow().patterns().is_empty());
1278    }
1279
1280    #[rstest]
1281    fn test_subscriptions_count_rejects_invalid_topic() {
1282        let msgbus = get_message_bus();
1283
1284        let err = msgbus
1285            .borrow()
1286            .subscriptions_count("data.*")
1287            .expect_err("wildcards are invalid in topics");
1288
1289        assert_eq!(
1290            err.to_string(),
1291            "Topic `value` contained invalid characters, was data.*"
1292        );
1293    }
1294
1295    #[rstest]
1296    fn test_has_subscribers_rejects_invalid_topic() {
1297        let msgbus = get_message_bus();
1298
1299        let err = msgbus
1300            .borrow()
1301            .has_subscribers("data.*")
1302            .expect_err("wildcards are invalid in topics");
1303
1304        assert_eq!(
1305            err.to_string(),
1306            "Topic `value` contained invalid characters, was data.*"
1307        );
1308    }
1309
1310    #[rstest]
1311    fn test_subscriptions_count_any_rejects_invalid_topic() {
1312        let err = subscriptions_count_any("data.*").expect_err("wildcards are invalid in topics");
1313
1314        assert_eq!(
1315            err.to_string(),
1316            "Topic `value` contained invalid characters, was data.*"
1317        );
1318    }
1319
1320    #[rstest]
1321    fn test_matching_subscriptions() {
1322        let msgbus = get_message_bus();
1323        let pattern = "my-pattern";
1324
1325        let handler_id1 = Ustr::from("1");
1326        let handler1 = get_stub_shareable_handler(Some(handler_id1));
1327
1328        let handler_id2 = Ustr::from("2");
1329        let handler2 = get_stub_shareable_handler(Some(handler_id2));
1330
1331        let handler_id3 = Ustr::from("3");
1332        let handler3 = get_stub_shareable_handler(Some(handler_id3));
1333
1334        let handler_id4 = Ustr::from("4");
1335        let handler4 = get_stub_shareable_handler(Some(handler_id4));
1336
1337        msgbus::subscribe_any(pattern.into(), handler1, None);
1338        msgbus::subscribe_any(pattern.into(), handler2, None);
1339        msgbus::subscribe_any(pattern.into(), handler3, Some(1));
1340        msgbus::subscribe_any(pattern.into(), handler4, Some(2));
1341
1342        assert_eq!(
1343            msgbus.borrow().patterns(),
1344            vec![pattern, pattern, pattern, pattern]
1345        );
1346        assert_eq!(subscriptions_count_any(pattern).unwrap(), 4);
1347
1348        let topic = pattern;
1349        let subs = msgbus.borrow_mut().matching_subscriptions(topic);
1350        assert_eq!(subs.len(), 4);
1351        assert_eq!(subs[0].handler_id, handler_id4);
1352        assert_eq!(subs[1].handler_id, handler_id3);
1353        assert_eq!(subs[2].handler_id, handler_id1);
1354        assert_eq!(subs[3].handler_id, handler_id2);
1355    }
1356
1357    #[rstest]
1358    fn test_matching_subscriptions_orders_by_full_delivery_key_on_cache_miss() {
1359        MessageBus::default().register_message_bus();
1360        let order = Rc::new(RefCell::new(Vec::new()));
1361
1362        msgbus::subscribe_any(
1363            "delivery.*".into(),
1364            recording_any_handler("handler-z", "low-z", order.clone()),
1365            Some(1),
1366        );
1367        msgbus::subscribe_any(
1368            "delivery.topic".into(),
1369            recording_any_handler("handler-b", "exact-b", order.clone()),
1370            Some(10),
1371        );
1372        msgbus::subscribe_any(
1373            "delivery.topic".into(),
1374            recording_any_handler("handler-a", "exact-a", order.clone()),
1375            Some(10),
1376        );
1377        msgbus::subscribe_any(
1378            "delivery.*".into(),
1379            recording_any_handler("handler-b", "wildcard-b", order),
1380            Some(10),
1381        );
1382
1383        let subscriptions = get_message_bus()
1384            .borrow_mut()
1385            .matching_subscriptions("delivery.topic");
1386        let actual = subscriptions
1387            .iter()
1388            .map(|sub| (sub.priority, sub.pattern.as_str(), sub.handler_id.as_str()))
1389            .collect::<Vec<_>>();
1390
1391        assert_eq!(
1392            actual,
1393            vec![
1394                (10, "delivery.*", "handler-b"),
1395                (10, "delivery.topic", "handler-a"),
1396                (10, "delivery.topic", "handler-b"),
1397                (1, "delivery.*", "handler-z"),
1398            ]
1399        );
1400    }
1401
1402    #[rstest]
1403    fn test_first_uncached_publish_any_orders_by_full_delivery_key() {
1404        MessageBus::default().register_message_bus();
1405        let order = Rc::new(RefCell::new(Vec::new()));
1406
1407        msgbus::subscribe_any(
1408            "delivery.*".into(),
1409            recording_any_handler("handler-z", "low-z", order.clone()),
1410            Some(1),
1411        );
1412        msgbus::subscribe_any(
1413            "delivery.topic".into(),
1414            recording_any_handler("handler-b", "exact-b", order.clone()),
1415            Some(10),
1416        );
1417        msgbus::subscribe_any(
1418            "delivery.topic".into(),
1419            recording_any_handler("handler-a", "exact-a", order.clone()),
1420            Some(10),
1421        );
1422        msgbus::subscribe_any(
1423            "delivery.*".into(),
1424            recording_any_handler("handler-b", "wildcard-b", order.clone()),
1425            Some(10),
1426        );
1427
1428        msgbus::publish_any("delivery.topic".into(), &());
1429
1430        assert_eq!(
1431            *order.borrow(),
1432            vec!["wildcard-b", "exact-a", "exact-b", "low-z"]
1433        );
1434    }
1435
1436    #[rstest]
1437    fn test_late_subscription_orders_cached_any_topic_by_full_delivery_key() {
1438        MessageBus::default().register_message_bus();
1439        let order = Rc::new(RefCell::new(Vec::new()));
1440
1441        msgbus::subscribe_any(
1442            "delivery.*".into(),
1443            recording_any_handler("handler-z", "low-z", order.clone()),
1444            Some(1),
1445        );
1446        msgbus::subscribe_any(
1447            "delivery.topic".into(),
1448            recording_any_handler("handler-b", "exact-b", order.clone()),
1449            Some(10),
1450        );
1451        msgbus::subscribe_any(
1452            "delivery.topic".into(),
1453            recording_any_handler("handler-a", "exact-a", order.clone()),
1454            Some(10),
1455        );
1456        msgbus::publish_any("delivery.topic".into(), &());
1457        order.borrow_mut().clear();
1458
1459        msgbus::subscribe_any(
1460            "delivery.*".into(),
1461            recording_any_handler("handler-b", "wildcard-b", order.clone()),
1462            Some(10),
1463        );
1464        msgbus::publish_any("delivery.topic".into(), &());
1465
1466        assert_eq!(
1467            *order.borrow(),
1468            vec!["wildcard-b", "exact-a", "exact-b", "low-z"]
1469        );
1470    }
1471
1472    #[rstest]
1473    fn test_subscription_pattern_matching() {
1474        let msgbus = get_message_bus();
1475        let handler1 = get_stub_shareable_handler(Some(Ustr::from("1")));
1476        let handler2 = get_stub_shareable_handler(Some(Ustr::from("2")));
1477        let handler3 = get_stub_shareable_handler(Some(Ustr::from("3")));
1478
1479        msgbus::subscribe_any("data.quotes.*".into(), handler1, None);
1480        msgbus::subscribe_any("data.trades.*".into(), handler2, None);
1481        msgbus::subscribe_any("data.*.BINANCE.*".into(), handler3, None);
1482        assert_eq!(msgbus.borrow().subscriptions().len(), 3);
1483
1484        let topic = "data.quotes.BINANCE.ETHUSDT";
1485        assert_eq!(msgbus.borrow().find_topic_matches(topic.into()).len(), 2);
1486
1487        let matches = msgbus.borrow_mut().matching_subscriptions(topic);
1488        assert_eq!(matches.len(), 2);
1489        assert_eq!(matches[0].handler_id, Ustr::from("3"));
1490        assert_eq!(matches[1].handler_id, Ustr::from("1"));
1491    }
1492
1493    #[rstest]
1494    fn test_late_wildcard_subscription_receives_cached_topic() {
1495        let msgbus = get_message_bus();
1496        let topic = "data.instrument.POLYMARKET.TEST-SYMBOL";
1497
1498        let (early_handler, early_saver) =
1499            get_any_saving_handler::<String>(Some(Ustr::from("early")));
1500        msgbus::subscribe_any("data.*.POLYMARKET.*".into(), early_handler, None);
1501
1502        msgbus::publish_any(topic.into(), &"ONE".to_string());
1503
1504        let (late_handler, late_saver) = get_any_saving_handler::<String>(Some(Ustr::from("late")));
1505        msgbus::subscribe_any("data.instrument.POLYMARKET.*".into(), late_handler, None);
1506
1507        msgbus::publish_any(topic.into(), &"TWO".to_string());
1508
1509        assert_eq!(early_saver.get_messages(), vec!["ONE", "TWO"]);
1510        assert_eq!(late_saver.get_messages(), vec!["TWO"]);
1511
1512        let topic_mstr: MStr<Topic> = topic.into();
1513        let cached = msgbus.borrow_mut().matching_subscriptions(topic_mstr);
1514        assert_eq!(cached.len(), 2);
1515    }
1516
1517    #[rstest]
1518    fn test_late_wildcard_backfills_into_multiple_cached_topics() {
1519        let msgbus = get_message_bus();
1520        let topics = ["data.A", "data.B", "data.C"];
1521
1522        let (early_handler, early_saver) =
1523            get_any_saving_handler::<String>(Some(Ustr::from("early")));
1524        msgbus::subscribe_any("data.*".into(), early_handler, None);
1525
1526        for topic in &topics {
1527            msgbus::publish_any((*topic).into(), &(*topic).to_string());
1528        }
1529
1530        let (late_handler, late_saver) = get_any_saving_handler::<String>(Some(Ustr::from("late")));
1531        msgbus::subscribe_any("data.*".into(), late_handler, None);
1532
1533        for topic in &topics {
1534            msgbus::publish_any((*topic).into(), &format!("{topic}-2"));
1535        }
1536
1537        assert_eq!(
1538            early_saver.get_messages(),
1539            vec![
1540                "data.A", "data.B", "data.C", "data.A-2", "data.B-2", "data.C-2"
1541            ],
1542        );
1543        assert_eq!(
1544            late_saver.get_messages(),
1545            vec!["data.A-2", "data.B-2", "data.C-2"]
1546        );
1547
1548        for topic in &topics {
1549            let topic_mstr: MStr<Topic> = (*topic).into();
1550            assert_eq!(
1551                msgbus.borrow_mut().matching_subscriptions(topic_mstr).len(),
1552                2,
1553                "topic {topic} should have both subscribers cached",
1554            );
1555        }
1556    }
1557
1558    /// A simple reference model for subscription behavior.
1559    struct SimpleSubscriptionModel {
1560        /// Stores (pattern, `handler_id`) tuples for active subscriptions.
1561        subscriptions: Vec<(String, String)>,
1562    }
1563
1564    impl SimpleSubscriptionModel {
1565        fn new() -> Self {
1566            Self {
1567                subscriptions: Vec::new(),
1568            }
1569        }
1570
1571        fn subscribe(&mut self, pattern: &str, handler_id: &str) {
1572            let subscription = (pattern.to_string(), handler_id.to_string());
1573            if !self.subscriptions.contains(&subscription) {
1574                self.subscriptions.push(subscription);
1575            }
1576        }
1577
1578        fn unsubscribe(&mut self, pattern: &str, handler_id: &str) -> bool {
1579            let subscription = (pattern.to_string(), handler_id.to_string());
1580            if let Some(idx) = self.subscriptions.iter().position(|s| s == &subscription) {
1581                self.subscriptions.remove(idx);
1582                true
1583            } else {
1584                false
1585            }
1586        }
1587
1588        fn is_subscribed(&self, pattern: &str, handler_id: &str) -> bool {
1589            self.subscriptions
1590                .contains(&(pattern.to_string(), handler_id.to_string()))
1591        }
1592
1593        fn matching_subscriptions(&self, topic: &str) -> Vec<(String, String)> {
1594            let topic = topic.into();
1595
1596            self.subscriptions
1597                .iter()
1598                .filter(|(pat, _)| is_matching_backtracking(topic, pat.into()))
1599                .map(|(pat, id)| (pat.clone(), id.clone()))
1600                .collect()
1601        }
1602
1603        fn subscription_count(&self) -> usize {
1604            self.subscriptions.len()
1605        }
1606    }
1607
1608    #[rstest]
1609    fn subscription_model_fuzz_testing() {
1610        let mut rng = StdRng::seed_from_u64(42);
1611
1612        let msgbus = get_message_bus();
1613        let mut model = SimpleSubscriptionModel::new();
1614
1615        // Map from handler_id to handler
1616        let mut handlers: Vec<(String, ShareableMessageHandler)> = Vec::new();
1617
1618        // Generate some patterns
1619        let patterns = generate_test_patterns(&mut rng);
1620
1621        // Generate some handler IDs
1622        let handler_ids: Vec<String> = (0..50).map(|i| format!("handler_{i}")).collect();
1623
1624        // Initialize handlers
1625        for id in &handler_ids {
1626            let handler = get_stub_shareable_handler(Some(Ustr::from(id)));
1627            handlers.push((id.clone(), handler));
1628        }
1629
1630        let num_operations = 50_000;
1631        for op_num in 0..num_operations {
1632            let operation = rng.random_range(0..4);
1633
1634            match operation {
1635                // Subscribe
1636                0 => {
1637                    let pattern_idx = rng.random_range(0..patterns.len());
1638                    let handler_idx = rng.random_range(0..handlers.len());
1639                    let pattern = &patterns[pattern_idx];
1640                    let (handler_id, handler) = &handlers[handler_idx];
1641
1642                    // Apply to reference model
1643                    model.subscribe(pattern, handler_id);
1644
1645                    // Apply to message bus
1646                    msgbus::subscribe_any(pattern.as_str().into(), handler.clone(), None);
1647
1648                    assert_eq!(
1649                        model.subscription_count(),
1650                        msgbus.borrow().subscriptions().len()
1651                    );
1652
1653                    assert!(
1654                        msgbus.borrow().is_subscribed(pattern, handler.clone()),
1655                        "Op {op_num}: is_subscribed should return true after subscribe"
1656                    );
1657                }
1658
1659                // Unsubscribe
1660                1 => {
1661                    if model.subscription_count() > 0 {
1662                        let sub_idx = rng.random_range(0..model.subscription_count());
1663                        let (pattern, handler_id) = model.subscriptions[sub_idx].clone();
1664
1665                        // Apply to reference model
1666                        model.unsubscribe(&pattern, &handler_id);
1667
1668                        // Find handler
1669                        let handler = handlers
1670                            .iter()
1671                            .find(|(id, _)| id == &handler_id)
1672                            .map(|(_, h)| h.clone())
1673                            .unwrap();
1674
1675                        // Apply to message bus
1676                        msgbus::unsubscribe_any(pattern.as_str().into(), &handler);
1677
1678                        assert_eq!(
1679                            model.subscription_count(),
1680                            msgbus.borrow().subscriptions().len()
1681                        );
1682                        assert!(
1683                            !msgbus.borrow().is_subscribed(pattern, handler.clone()),
1684                            "Op {op_num}: is_subscribed should return false after unsubscribe"
1685                        );
1686                    }
1687                }
1688
1689                // Check is_subscribed
1690                2 => {
1691                    // Get a random pattern and handler
1692                    let pattern_idx = rng.random_range(0..patterns.len());
1693                    let handler_idx = rng.random_range(0..handlers.len());
1694                    let pattern = &patterns[pattern_idx];
1695                    let (handler_id, handler) = &handlers[handler_idx];
1696
1697                    let expected = model.is_subscribed(pattern, handler_id);
1698                    let actual = msgbus.borrow().is_subscribed(pattern, handler.clone());
1699
1700                    assert_eq!(
1701                        expected, actual,
1702                        "Op {op_num}: Subscription state mismatch for pattern '{pattern}', handler '{handler_id}': expected={expected}, actual={actual}"
1703                    );
1704                }
1705
1706                // Check matching_subscriptions
1707                3 => {
1708                    // Generate a topic
1709                    let topic = create_topic(&mut rng);
1710
1711                    let actual_matches = msgbus.borrow_mut().matching_subscriptions(topic);
1712                    let expected_matches = model.matching_subscriptions(&topic);
1713
1714                    assert_eq!(
1715                        expected_matches.len(),
1716                        actual_matches.len(),
1717                        "Op {}: Match count mismatch for topic '{}': expected={}, actual={}",
1718                        op_num,
1719                        topic,
1720                        expected_matches.len(),
1721                        actual_matches.len()
1722                    );
1723
1724                    for sub in &actual_matches {
1725                        assert!(
1726                            expected_matches
1727                                .contains(&(sub.pattern.to_string(), sub.handler_id.to_string())),
1728                            "Op {}: Expected match not found: pattern='{}', handler_id='{}'",
1729                            op_num,
1730                            sub.pattern,
1731                            sub.handler_id
1732                        );
1733                    }
1734                }
1735                _ => unreachable!(),
1736            }
1737        }
1738    }
1739
1740    fn generate_pattern_from_topic(topic: &str, rng: &mut StdRng) -> String {
1741        let mut pattern = String::new();
1742
1743        for c in topic.chars() {
1744            let val: f64 = rng.random();
1745            if val < 0.1 {
1746                pattern.push('*');
1747            } else if val < 0.3 {
1748                pattern.push('?');
1749            } else if val >= 0.5 {
1750                pattern.push(c);
1751            }
1752        }
1753
1754        pattern
1755    }
1756
1757    fn generate_test_patterns(rng: &mut StdRng) -> Vec<String> {
1758        let mut patterns = vec![
1759            "data.*.*.*".to_string(),
1760            "*.*.BINANCE.*".to_string(),
1761            "events.order.*".to_string(),
1762            "data.*.*.?USDT".to_string(),
1763            "*.trades.*.BTC*".to_string(),
1764            "*.*.*.*".to_string(),
1765        ];
1766
1767        // Add some random patterns
1768        for _ in 0..50 {
1769            match rng.random_range(0..10) {
1770                // Use existing pattern
1771                0..=1 => {
1772                    let idx = rng.random_range(0..patterns.len());
1773                    patterns.push(patterns[idx].clone());
1774                }
1775                // Generate new pattern from topic
1776                _ => {
1777                    let topic = create_topic(rng);
1778                    let pattern = generate_pattern_from_topic(&topic, rng);
1779                    patterns.push(pattern);
1780                }
1781            }
1782        }
1783
1784        patterns
1785    }
1786
1787    fn create_topic(rng: &mut StdRng) -> Ustr {
1788        let cat = ["data", "info", "order"];
1789        let model = ["quotes", "trades", "orderbooks", "depths"];
1790        let venue = ["BINANCE", "BYBIT", "OKX", "FTX", "KRAKEN"];
1791        let instrument = ["BTCUSDT", "ETHUSDT", "SOLUSDT", "XRPUSDT", "DOGEUSDT"];
1792
1793        let cat = cat[rng.random_range(0..cat.len())];
1794        let model = model[rng.random_range(0..model.len())];
1795        let venue = venue[rng.random_range(0..venue.len())];
1796        let instrument = instrument[rng.random_range(0..instrument.len())];
1797        Ustr::from(&format!("{cat}.{model}.{venue}.{instrument}"))
1798    }
1799}
nautilus_common/msgbus/core.rs

nautilus_common/msgbus/
core.rs
