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//!   `OrderBookDepth10`, `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, OrderBookDepth10, 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///
139#[derive(Clone, Debug)]
140pub struct Subscription {
141    /// The shareable message handler for the subscription.
142    pub handler: ShareableMessageHandler,
143    /// Store a copy of the handler ID for faster equality checks.
144    pub handler_id: Ustr,
145    /// The pattern for the subscription.
146    pub pattern: MStr<Pattern>,
147    /// The priority for the subscription determines the ordering of handlers receiving
148    /// messages being processed, higher priority handlers will receive messages before
149    /// lower priority handlers.
150    pub priority: u32,
151}
152
153impl Subscription {
154    /// Creates a new [`Subscription`] instance.
155    #[must_use]
156    pub fn new(
157        pattern: MStr<Pattern>,
158        handler: ShareableMessageHandler,
159        priority: Option<u32>,
160    ) -> Self {
161        Self {
162            handler_id: handler.0.id(),
163            pattern,
164            handler,
165            priority: priority.unwrap_or(0),
166        }
167    }
168}
169
170impl PartialEq<Self> for Subscription {
171    fn eq(&self, other: &Self) -> bool {
172        self.pattern == other.pattern && self.handler_id == other.handler_id
173    }
174}
175
176impl Eq for Subscription {}
177
178impl PartialOrd for Subscription {
179    fn partial_cmp(&self, other: &Self) -> Option<std::cmp::Ordering> {
180        Some(self.cmp(other))
181    }
182}
183
184impl Ord for Subscription {
185    fn cmp(&self, other: &Self) -> std::cmp::Ordering {
186        other
187            .priority
188            .cmp(&self.priority)
189            .then_with(|| self.pattern.cmp(&other.pattern))
190            .then_with(|| self.handler_id.cmp(&other.handler_id))
191    }
192}
193
194impl Hash for Subscription {
195    fn hash<H: Hasher>(&self, state: &mut H) {
196        self.pattern.hash(state);
197        self.handler_id.hash(state);
198    }
199}
200
201/// A generic message bus to facilitate various messaging patterns.
202///
203/// The bus provides both a producer and consumer API for Pub/Sub, Req/Rep, as
204/// well as direct point-to-point messaging to registered endpoints.
205///
206/// Pub/Sub wildcard patterns for hierarchical topics are possible:
207///  - `*` asterisk represents one or more characters in a pattern.
208///  - `?` question mark represents a single character in a pattern.
209///
210/// Given a topic and pattern potentially containing wildcard characters, i.e.
211/// `*` and `?`, where `?` can match any single character in the topic, and `*`
212/// can match any number of characters including zero characters.
213///
214/// The asterisk in a wildcard matches any character zero or more times. For
215/// example, `comp*` matches anything beginning with `comp` which means `comp`,
216/// `complete`, and `computer` are all matched.
217///
218/// A question mark matches a single character once. For example, `c?mp` matches
219/// `camp` and `comp`. The question mark can also be used more than once.
220/// For example, `c??p` would match both of the above examples and `coop`.
221pub struct MessageBus {
222    /// The trader ID associated with the message bus.
223    pub trader_id: TraderId,
224    /// The instance ID associated with the message bus.
225    pub instance_id: UUID4,
226    /// The name for the message bus.
227    pub name: String,
228    /// If the message bus is backed by a database.
229    pub has_backing: bool,
230    pub(crate) switchboard: MessagingSwitchboard,
231    pub(crate) subscriptions: AHashSet<Subscription>,
232    pub(crate) topics: IndexMap<MStr<Topic>, Vec<Subscription>>,
233    pub(crate) endpoints: IndexMap<MStr<Endpoint>, ShareableMessageHandler>,
234    pub(crate) correlation_index: AHashMap<UUID4, ShareableMessageHandler>,
235    pub(crate) router_quotes: TopicRouter<QuoteTick>,
236    pub(crate) router_trades: TopicRouter<TradeTick>,
237    pub(crate) router_bars: TopicRouter<Bar>,
238    pub(crate) router_deltas: TopicRouter<OrderBookDeltas>,
239    pub(crate) router_depth10: TopicRouter<OrderBookDepth10>,
240    pub(crate) router_book_snapshots: TopicRouter<OrderBook>,
241    pub(crate) router_mark_prices: TopicRouter<MarkPriceUpdate>,
242    pub(crate) router_index_prices: TopicRouter<IndexPriceUpdate>,
243    pub(crate) router_funding_rates: TopicRouter<FundingRateUpdate>,
244    pub(crate) router_order_events: TopicRouter<OrderEventAny>,
245    pub(crate) router_position_events: TopicRouter<PositionEvent>,
246    pub(crate) router_account_state: TopicRouter<AccountState>,
247    pub(crate) router_orders: TopicRouter<OrderAny>,
248    pub(crate) router_positions: TopicRouter<Position>,
249    pub(crate) router_portfolio: TopicRouter<PortfolioSnapshot>,
250    pub(crate) router_greeks: TopicRouter<GreeksData>,
251    pub(crate) router_option_greeks: TopicRouter<OptionGreeks>,
252    pub(crate) router_option_chain: TopicRouter<OptionChainSlice>,
253    pub(crate) router_instruments: TopicRouter<InstrumentAny>,
254    #[cfg(feature = "defi")]
255    pub(crate) router_defi_blocks: TopicRouter<nautilus_model::defi::Block>, // nautilus-import-ok
256    #[cfg(feature = "defi")]
257    pub(crate) router_defi_pools: TopicRouter<nautilus_model::defi::Pool>, // nautilus-import-ok
258    #[cfg(feature = "defi")]
259    pub(crate) router_defi_swaps: TopicRouter<nautilus_model::defi::PoolSwap>, // nautilus-import-ok
260    #[cfg(feature = "defi")]
261    pub(crate) router_defi_liquidity: TopicRouter<nautilus_model::defi::PoolLiquidityUpdate>, // nautilus-import-ok
262    #[cfg(feature = "defi")]
263    pub(crate) router_defi_collects: TopicRouter<nautilus_model::defi::PoolFeeCollect>, // nautilus-import-ok
264    #[cfg(feature = "defi")]
265    pub(crate) router_defi_flash: TopicRouter<nautilus_model::defi::PoolFlash>, // nautilus-import-ok
266    #[cfg(feature = "defi")]
267    pub(crate) endpoints_defi_data: IntoEndpointMap<nautilus_model::defi::DefiData>, // nautilus-import-ok
268    pub(crate) endpoints_quotes: EndpointMap<QuoteTick>,
269    pub(crate) endpoints_trades: EndpointMap<TradeTick>,
270    pub(crate) endpoints_bars: EndpointMap<Bar>,
271    pub(crate) endpoints_account_state: EndpointMap<AccountState>,
272    pub(crate) endpoints_trading_commands: IntoEndpointMap<TradingCommand>,
273    pub(crate) endpoints_data_commands: IntoEndpointMap<DataCommand>,
274    pub(crate) endpoints_data_responses: IntoEndpointMap<DataResponse>,
275    pub(crate) endpoints_exec_reports: IntoEndpointMap<ExecutionReport>,
276    pub(crate) endpoints_order_events: IntoEndpointMap<OrderEventAny>,
277    pub(crate) endpoints_data: IntoEndpointMap<Data>,
278    routers_typed: AHashMap<TypeId, Box<dyn Any>>,
279    endpoints_typed: AHashMap<TypeId, Box<dyn Any>>,
280    sent_count: u64,
281    req_count: u64,
282    res_count: u64,
283    pub_count: u64,
284    external_egress: Option<Box<dyn MessageBusExternalEgress>>,
285    encoding: SerializationEncoding,
286    encoding_market_data: Option<SerializationEncoding>,
287    encoding_builtin: Option<SerializationEncoding>,
288    types_filter: AHashSet<BusPayloadType>,
289    streaming_types: AHashSet<BusPayloadType>,
290}
291
292impl Debug for MessageBus {
293    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
294        f.debug_struct(stringify!(MessageBus))
295            .field("trader_id", &self.trader_id)
296            .field("instance_id", &self.instance_id)
297            .field("name", &self.name)
298            .field("has_backing", &self.has_backing)
299            .field("external_egress", &self.external_egress.is_some())
300            .finish_non_exhaustive()
301    }
302}
303
304impl Default for MessageBus {
305    /// Creates a new default [`MessageBus`] instance.
306    fn default() -> Self {
307        Self::new(TraderId::from("TRADER-001"), UUID4::new(), None, None)
308    }
309}
310
311impl MessageBus {
312    /// Creates a new [`MessageBus`] instance.
313    #[must_use]
314    pub fn new(
315        trader_id: TraderId,
316        instance_id: UUID4,
317        name: Option<String>,
318        _config: Option<HashMap<String, serde_json::Value>>,
319    ) -> Self {
320        Self {
321            trader_id,
322            instance_id,
323            name: name.unwrap_or(stringify!(MessageBus).to_owned()),
324            switchboard: MessagingSwitchboard::default(),
325            subscriptions: AHashSet::new(),
326            topics: IndexMap::new(),
327            endpoints: IndexMap::new(),
328            correlation_index: AHashMap::new(),
329            has_backing: false,
330            router_quotes: TopicRouter::new(),
331            router_trades: TopicRouter::new(),
332            router_bars: TopicRouter::new(),
333            router_deltas: TopicRouter::new(),
334            router_depth10: TopicRouter::new(),
335            router_book_snapshots: TopicRouter::new(),
336            router_mark_prices: TopicRouter::new(),
337            router_index_prices: TopicRouter::new(),
338            router_funding_rates: TopicRouter::new(),
339            router_order_events: TopicRouter::new(),
340            router_position_events: TopicRouter::new(),
341            router_account_state: TopicRouter::new(),
342            router_portfolio: TopicRouter::new(),
343            router_orders: TopicRouter::new(),
344            router_positions: TopicRouter::new(),
345            router_greeks: TopicRouter::new(),
346            router_option_greeks: TopicRouter::new(),
347            router_option_chain: TopicRouter::new(),
348            router_instruments: TopicRouter::new(),
349            #[cfg(feature = "defi")]
350            router_defi_blocks: TopicRouter::new(),
351            #[cfg(feature = "defi")]
352            router_defi_pools: TopicRouter::new(),
353            #[cfg(feature = "defi")]
354            router_defi_swaps: TopicRouter::new(),
355            #[cfg(feature = "defi")]
356            router_defi_liquidity: TopicRouter::new(),
357            #[cfg(feature = "defi")]
358            router_defi_collects: TopicRouter::new(),
359            #[cfg(feature = "defi")]
360            router_defi_flash: TopicRouter::new(),
361            #[cfg(feature = "defi")]
362            endpoints_defi_data: IntoEndpointMap::new(),
363            endpoints_quotes: EndpointMap::new(),
364            endpoints_trades: EndpointMap::new(),
365            endpoints_bars: EndpointMap::new(),
366            endpoints_account_state: EndpointMap::new(),
367            endpoints_trading_commands: IntoEndpointMap::new(),
368            endpoints_data_commands: IntoEndpointMap::new(),
369            endpoints_data_responses: IntoEndpointMap::new(),
370            endpoints_exec_reports: IntoEndpointMap::new(),
371            endpoints_order_events: IntoEndpointMap::new(),
372            endpoints_data: IntoEndpointMap::new(),
373            routers_typed: AHashMap::new(),
374            endpoints_typed: AHashMap::new(),
375            sent_count: 0,
376            req_count: 0,
377            res_count: 0,
378            pub_count: 0,
379            external_egress: None,
380            encoding: SerializationEncoding::Json,
381            encoding_market_data: None,
382            encoding_builtin: None,
383            types_filter: AHashSet::new(),
384            streaming_types: AHashSet::new(),
385        }
386    }
387
388    /// Registers message bus for the current thread.
389    pub fn register_message_bus(self) -> Rc<RefCell<Self>> {
390        let msgbus = Rc::new(RefCell::new(self));
391        set_message_bus(msgbus.clone());
392        msgbus
393    }
394
395    /// Gets or creates a typed router for custom message type `T`.
396    ///
397    /// # Panics
398    ///
399    /// Panics if the stored router type doesn't match `T` (internal bug).
400    pub fn router<T: 'static>(&mut self) -> &mut TopicRouter<T> {
401        self.routers_typed
402            .entry(TypeId::of::<T>())
403            .or_insert_with(|| Box::new(TopicRouter::<T>::new()))
404            .downcast_mut::<TopicRouter<T>>()
405            .expect("TopicRouter type mismatch - this is a bug")
406    }
407
408    /// Gets or creates a typed endpoint map for custom message type `T`.
409    ///
410    /// # Panics
411    ///
412    /// Panics if the stored endpoint map type doesn't match `T` (internal bug).
413    pub fn endpoint_map<T: 'static>(&mut self) -> &mut EndpointMap<T> {
414        self.endpoints_typed
415            .entry(TypeId::of::<T>())
416            .or_insert_with(|| Box::new(EndpointMap::<T>::new()))
417            .downcast_mut::<EndpointMap<T>>()
418            .expect("EndpointMap type mismatch - this is a bug")
419    }
420
421    /// Sets external egress for serialized published messages.
422    pub fn set_external_egress(
423        &mut self,
424        external_egress: Box<dyn MessageBusExternalEgress>,
425        encoding: SerializationEncoding,
426    ) {
427        self.external_egress = Some(external_egress);
428        self.encoding = encoding;
429        self.encoding_market_data = None;
430        self.encoding_builtin = None;
431        self.has_backing = true;
432        HAS_EXTERNAL_EGRESS.with(|flag| flag.set(true));
433    }
434
435    /// Sets external egress and category encoding policy from a validated config.
436    ///
437    /// # Errors
438    ///
439    /// Returns a [`crate::config::ConfigError`] if the config selects an unsupported encoding.
440    pub fn set_external_egress_config(
441        &mut self,
442        external_egress: Box<dyn MessageBusExternalEgress>,
443        config: &MessageBusConfig,
444    ) -> crate::config::ConfigResult<()> {
445        config.validate()?;
446
447        self.external_egress = Some(external_egress);
448        self.encoding = config.encoding;
449        self.encoding_market_data = config.encoding_market_data;
450        self.encoding_builtin = config.encoding_builtin;
451        self.has_backing = true;
452        HAS_EXTERNAL_EGRESS.with(|flag| flag.set(true));
453
454        Ok(())
455    }
456
457    /// Sets the type names excluded from external publishing.
458    pub fn set_types_filter(&mut self, filter: Vec<String>) {
459        self.types_filter = filter
460            .into_iter()
461            .map(|type_name| BusPayloadType::from_name(&type_name))
462            .filter(|payload_type| !payload_type.as_str().is_empty())
463            .collect();
464    }
465
466    /// Registers a payload type for external-to-internal streaming.
467    pub fn add_streaming_type(&mut self, payload_type: BusPayloadType) {
468        if !payload_type.as_str().is_empty() {
469            self.streaming_types.insert(payload_type);
470        }
471    }
472
473    /// Returns whether the payload type is registered for external-to-internal streaming.
474    #[must_use]
475    pub fn is_streaming_type(&self, payload_type: BusPayloadType) -> bool {
476        !payload_type.as_str().is_empty() && self.streaming_types.contains(&payload_type)
477    }
478
479    /// Clears all payload types registered for external-to-internal streaming.
480    pub fn clear_streaming_types(&mut self) {
481        self.streaming_types.clear();
482    }
483
484    #[must_use]
485    pub(crate) fn has_external_egress(&self) -> bool {
486        self.external_egress.is_some()
487    }
488
489    pub(crate) fn external_egress(&self) -> Option<&dyn MessageBusExternalEgress> {
490        self.external_egress.as_deref()
491    }
492
493    pub(crate) fn encoding_for(&self, payload_type: BusPayloadType) -> SerializationEncoding {
494        match payload_type.category() {
495            BusPayloadCategory::MarketData => self.encoding_market_data.unwrap_or(self.encoding),
496            BusPayloadCategory::BuiltIn => self.encoding_builtin.unwrap_or(self.encoding),
497            BusPayloadCategory::Other => self.encoding,
498        }
499    }
500
501    pub(crate) fn types_filter(&self) -> &AHashSet<BusPayloadType> {
502        &self.types_filter
503    }
504
505    /// Disposes of the message bus, clearing all subscriptions, endpoints,
506    /// and handler references.
507    pub fn dispose(&mut self) {
508        self.subscriptions.clear();
509        self.topics.clear();
510        self.endpoints.clear();
511        self.correlation_index.clear();
512
513        self.router_quotes.clear();
514        self.router_trades.clear();
515        self.router_bars.clear();
516        self.router_deltas.clear();
517        self.router_depth10.clear();
518        self.router_book_snapshots.clear();
519        self.router_mark_prices.clear();
520        self.router_index_prices.clear();
521        self.router_funding_rates.clear();
522        self.router_order_events.clear();
523        self.router_position_events.clear();
524        self.router_account_state.clear();
525        self.router_portfolio.clear();
526        self.router_orders.clear();
527        self.router_positions.clear();
528        self.router_greeks.clear();
529        self.router_option_greeks.clear();
530        self.router_option_chain.clear();
531        self.router_instruments.clear();
532
533        #[cfg(feature = "defi")]
534        {
535            self.router_defi_blocks.clear();
536            self.router_defi_pools.clear();
537            self.router_defi_swaps.clear();
538            self.router_defi_liquidity.clear();
539            self.router_defi_collects.clear();
540            self.router_defi_flash.clear();
541            self.endpoints_defi_data.clear();
542        }
543
544        self.endpoints_quotes.clear();
545        self.endpoints_trades.clear();
546        self.endpoints_bars.clear();
547        self.endpoints_account_state.clear();
548        self.endpoints_trading_commands.clear();
549        self.endpoints_data_commands.clear();
550        self.endpoints_data_responses.clear();
551        self.endpoints_exec_reports.clear();
552        self.endpoints_order_events.clear();
553        self.endpoints_data.clear();
554
555        self.routers_typed.clear();
556        self.endpoints_typed.clear();
557        self.clear_streaming_types();
558        self.sent_count = 0;
559        self.req_count = 0;
560        self.res_count = 0;
561        self.pub_count = 0;
562
563        if let Some(mut external_egress) = self.external_egress.take() {
564            external_egress.close();
565        }
566        self.has_backing = false;
567        HAS_EXTERNAL_EGRESS.with(|flag| flag.set(false));
568    }
569
570    /// Returns the memory address of this instance as a hexadecimal string.
571    #[must_use]
572    pub fn mem_address(&self) -> String {
573        format!("{self:p}")
574    }
575
576    /// Returns a reference to the switchboard.
577    #[must_use]
578    pub fn switchboard(&self) -> &MessagingSwitchboard {
579        &self.switchboard
580    }
581
582    /// Returns the total count of messages sent to endpoints.
583    #[must_use]
584    pub const fn sent_count(&self) -> u64 {
585        self.sent_count
586    }
587
588    /// Returns the total count of requests sent to endpoints.
589    #[must_use]
590    pub const fn req_count(&self) -> u64 {
591        self.req_count
592    }
593
594    /// Returns the total count of responses sent to registered handlers.
595    #[must_use]
596    pub const fn res_count(&self) -> u64 {
597        self.res_count
598    }
599
600    /// Returns the total count of messages published to topics.
601    #[must_use]
602    pub const fn pub_count(&self) -> u64 {
603        self.pub_count
604    }
605
606    pub(crate) fn increment_sent_count(&mut self) {
607        self.sent_count += 1;
608    }
609
610    pub(crate) fn increment_req_count(&mut self) {
611        self.req_count += 1;
612    }
613
614    pub(crate) fn increment_res_count(&mut self) {
615        self.res_count += 1;
616    }
617
618    pub(crate) fn increment_pub_count(&mut self) {
619        self.pub_count += 1;
620    }
621
622    /// Returns the registered endpoint addresses.
623    #[must_use]
624    pub fn endpoints(&self) -> Vec<&str> {
625        self.endpoints.iter().map(|e| e.0.as_str()).collect()
626    }
627
628    /// Returns actively subscribed patterns.
629    #[must_use]
630    pub fn patterns(&self) -> Vec<&str> {
631        self.subscriptions
632            .iter()
633            .map(|s| s.pattern.as_str())
634            .collect()
635    }
636
637    /// Returns whether there are subscribers for the `topic`.
638    ///
639    /// # Errors
640    ///
641    /// Returns an error if the `topic` is not a valid topic string.
642    pub fn has_subscribers<T: AsRef<str>>(&self, topic: T) -> anyhow::Result<bool> {
643        Ok(self.subscriptions_count(topic)? > 0)
644    }
645
646    /// Returns the count of subscribers for the `topic`.
647    ///
648    /// # Errors
649    ///
650    /// Returns an error if the `topic` is not a valid topic string.
651    pub fn subscriptions_count<T: AsRef<str>>(&self, topic: T) -> anyhow::Result<usize> {
652        let topic = MStr::<Topic>::topic(topic)?;
653        Ok(self
654            .topics
655            .get(&topic)
656            .map_or_else(|| self.find_topic_matches(topic).len(), Vec::len))
657    }
658
659    /// Returns active subscriptions.
660    #[must_use]
661    pub fn subscriptions(&self) -> Vec<&Subscription> {
662        self.subscriptions.iter().collect()
663    }
664
665    /// Returns the handler IDs for actively subscribed patterns.
666    #[must_use]
667    pub fn subscription_handler_ids(&self) -> Vec<&str> {
668        self.subscriptions
669            .iter()
670            .map(|s| s.handler_id.as_str())
671            .collect()
672    }
673
674    /// Returns whether the endpoint is registered.
675    ///
676    /// # Panics
677    ///
678    /// Panics if the `endpoint` conversion to `MStr<Endpoint>` fails.
679    #[must_use]
680    pub fn is_registered<T: Into<MStr<Endpoint>>>(&self, endpoint: T) -> bool {
681        let endpoint: MStr<Endpoint> = endpoint.into();
682        self.endpoints.contains_key(&endpoint)
683    }
684
685    /// Returns whether the `handler` is subscribed to the `pattern`.
686    #[must_use]
687    pub fn is_subscribed<T: AsRef<str>>(
688        &self,
689        pattern: T,
690        handler: ShareableMessageHandler,
691    ) -> bool {
692        let pattern = MStr::<Pattern>::pattern(pattern);
693        let sub = Subscription::new(pattern, handler, None);
694        self.subscriptions.contains(&sub)
695    }
696
697    /// Close the message bus which will close the sender channel and join the thread.
698    ///
699    /// # Errors
700    ///
701    /// This function never returns an error (TBD once backing database added).
702    pub fn close(&mut self) -> anyhow::Result<()> {
703        if let Some(mut external_egress) = self.external_egress.take() {
704            external_egress.close();
705        }
706        self.has_backing = false;
707        HAS_EXTERNAL_EGRESS.with(|flag| flag.set(false));
708        Ok(())
709    }
710
711    /// Returns the handler for the `endpoint`.
712    #[must_use]
713    pub fn get_endpoint(&self, endpoint: MStr<Endpoint>) -> Option<&ShareableMessageHandler> {
714        self.endpoints.get(&endpoint)
715    }
716
717    /// Returns the handler for the `correlation_id`.
718    #[must_use]
719    pub fn get_response_handler(&self, correlation_id: &UUID4) -> Option<&ShareableMessageHandler> {
720        self.correlation_index.get(correlation_id)
721    }
722
723    /// Finds the subscriptions with pattern matching the `topic`.
724    pub(crate) fn find_topic_matches(&self, topic: MStr<Topic>) -> Vec<Subscription> {
725        self.subscriptions
726            .iter()
727            .filter_map(|sub| {
728                if is_matching_backtracking(topic, sub.pattern) {
729                    Some(sub.clone())
730                } else {
731                    None
732                }
733            })
734            .collect()
735    }
736
737    /// Finds the subscriptions which match the `topic` and caches the
738    /// results in the `patterns` map.
739    #[must_use]
740    pub fn matching_subscriptions<T: Into<MStr<Topic>>>(&mut self, topic: T) -> Vec<Subscription> {
741        self.inner_matching_subscriptions(topic.into())
742    }
743
744    pub(crate) fn inner_matching_subscriptions(&mut self, topic: MStr<Topic>) -> Vec<Subscription> {
745        self.topics.get(&topic).cloned().unwrap_or_else(|| {
746            let mut matches = self.find_topic_matches(topic);
747            matches.sort();
748            self.topics.insert(topic, matches.clone());
749            matches
750        })
751    }
752
753    /// Fills a buffer with handlers matching a topic.
754    pub(crate) fn fill_matching_any_handlers(
755        &mut self,
756        topic: MStr<Topic>,
757        buf: &mut SmallVec<[ShareableMessageHandler; 64]>,
758    ) {
759        if let Some(subs) = self.topics.get(&topic) {
760            for sub in subs {
761                buf.push(sub.handler.clone());
762            }
763        } else {
764            let mut matches = self.find_topic_matches(topic);
765            matches.sort();
766
767            for sub in &matches {
768                buf.push(sub.handler.clone());
769            }
770
771            self.topics.insert(topic, matches);
772        }
773    }
774
775    /// Registers a response handler for a specific correlation ID.
776    ///
777    /// # Errors
778    ///
779    /// Returns an error if `handler` is already registered for the `correlation_id`.
780    pub fn register_response_handler(
781        &mut self,
782        correlation_id: &UUID4,
783        handler: ShareableMessageHandler,
784    ) -> anyhow::Result<()> {
785        if self.correlation_index.contains_key(correlation_id) {
786            anyhow::bail!("Correlation ID <{correlation_id}> already has a registered handler");
787        }
788
789        self.correlation_index.insert(*correlation_id, handler);
790
791        Ok(())
792    }
793}
794
795#[cfg(test)]
796mod tests {
797    use rand::{RngExt, SeedableRng, rngs::StdRng};
798    use rstest::rstest;
799    use ustr::Ustr;
800
801    use super::*;
802    use crate::msgbus::{
803        self, ShareableMessageHandler, get_message_bus,
804        matching::is_matching_backtracking,
805        stubs::{get_any_saving_handler, get_call_check_handler, get_stub_shareable_handler},
806        subscriptions_count_any,
807    };
808
809    #[rstest]
810    fn test_new() {
811        let trader_id = TraderId::default();
812        let msgbus = MessageBus::new(trader_id, UUID4::new(), None, None);
813
814        assert_eq!(msgbus.trader_id, trader_id);
815        assert_eq!(msgbus.name, stringify!(MessageBus));
816    }
817
818    #[rstest]
819    fn encoding_for_uses_market_data_override() {
820        let msgbus = MessageBus {
821            encoding: SerializationEncoding::Json,
822            encoding_market_data: Some(SerializationEncoding::MsgPack),
823            ..Default::default()
824        };
825
826        assert_eq!(
827            msgbus.encoding_for(BusPayloadType::QuoteTick),
828            SerializationEncoding::MsgPack
829        );
830        assert_eq!(
831            msgbus.encoding_for(BusPayloadType::Custom(Ustr::from("CustomPayload"))),
832            SerializationEncoding::Json
833        );
834    }
835
836    #[rstest]
837    fn encoding_for_uses_builtin_override() {
838        let msgbus = MessageBus {
839            encoding: SerializationEncoding::Json,
840            encoding_builtin: Some(SerializationEncoding::MsgPack),
841            ..Default::default()
842        };
843
844        assert_eq!(
845            msgbus.encoding_for(BusPayloadType::OrderEvent),
846            SerializationEncoding::MsgPack
847        );
848        assert_eq!(
849            msgbus.encoding_for(BusPayloadType::Instrument),
850            SerializationEncoding::Json
851        );
852    }
853
854    #[rstest]
855    fn encoding_for_uses_default_without_category_override() {
856        let msgbus = MessageBus {
857            encoding: SerializationEncoding::MsgPack,
858            ..Default::default()
859        };
860
861        assert_eq!(
862            msgbus.encoding_for(BusPayloadType::QuoteTick),
863            SerializationEncoding::MsgPack
864        );
865        assert_eq!(
866            msgbus.encoding_for(BusPayloadType::OrderEvent),
867            SerializationEncoding::MsgPack
868        );
869        assert_eq!(
870            msgbus.encoding_for(BusPayloadType::Custom(Ustr::from("CustomPayload"))),
871            SerializationEncoding::MsgPack
872        );
873    }
874
875    #[rstest]
876    fn set_types_filter_resolves_canonical_and_custom_names() {
877        let mut msgbus = MessageBus::default();
878
879        msgbus.set_types_filter(vec![
880            "QuoteTick".to_string(),
881            "ExternalCustomPayload".to_string(),
882            String::new(),
883        ]);
884
885        let filter = msgbus.types_filter();
886        assert_eq!(filter.len(), 2);
887        assert!(filter.contains(&BusPayloadType::QuoteTick));
888        assert!(filter.contains(&BusPayloadType::Custom(Ustr::from("ExternalCustomPayload"))));
889        assert!(!filter.contains(&BusPayloadType::Custom(Ustr::default())));
890    }
891
892    #[rstest]
893    fn streaming_type_registration_uses_canonical_payload_names() {
894        let mut msgbus = MessageBus::default();
895
896        msgbus.add_streaming_type(BusPayloadType::QuoteTick);
897        msgbus.add_streaming_type(BusPayloadType::Custom(Ustr::from("CustomPayload")));
898
899        assert!(msgbus.is_streaming_type(BusPayloadType::QuoteTick));
900        assert!(msgbus.is_streaming_type(BusPayloadType::Custom(Ustr::from("CustomPayload"))));
901        assert!(msgbus.streaming_types.contains(&BusPayloadType::QuoteTick));
902        assert!(
903            msgbus
904                .streaming_types
905                .contains(&BusPayloadType::Custom(Ustr::from("CustomPayload")))
906        );
907        assert!(!msgbus.is_streaming_type(BusPayloadType::TradeTick));
908    }
909
910    #[rstest]
911    fn streaming_type_registration_ignores_empty_custom_payload_type() {
912        let mut msgbus = MessageBus::default();
913
914        msgbus.add_streaming_type(BusPayloadType::Custom(Ustr::default()));
915
916        assert!(!msgbus.is_streaming_type(BusPayloadType::Custom(Ustr::default())));
917        assert!(msgbus.streaming_types.is_empty());
918    }
919
920    #[rstest]
921    fn clear_streaming_types_removes_registered_types() {
922        let mut msgbus = MessageBus::default();
923        msgbus.add_streaming_type(BusPayloadType::QuoteTick);
924
925        msgbus.clear_streaming_types();
926
927        assert!(!msgbus.is_streaming_type(BusPayloadType::QuoteTick));
928    }
929
930    #[rstest]
931    fn dispose_clears_streaming_types() {
932        let mut msgbus = MessageBus::default();
933        msgbus.add_streaming_type(BusPayloadType::QuoteTick);
934
935        msgbus.dispose();
936
937        assert!(!msgbus.is_streaming_type(BusPayloadType::QuoteTick));
938    }
939
940    #[rstest]
941    fn test_dispose_resets_counters() {
942        let mut msgbus = MessageBus::default();
943
944        msgbus.increment_sent_count();
945        msgbus.increment_req_count();
946        msgbus.increment_res_count();
947        msgbus.increment_pub_count();
948        msgbus.dispose();
949
950        assert_eq!(msgbus.sent_count(), 0);
951        assert_eq!(msgbus.req_count(), 0);
952        assert_eq!(msgbus.res_count(), 0);
953        assert_eq!(msgbus.pub_count(), 0);
954    }
955
956    #[rstest]
957    fn test_endpoints_when_no_endpoints() {
958        let msgbus = get_message_bus();
959        assert!(msgbus.borrow().endpoints().is_empty());
960    }
961
962    #[rstest]
963    fn test_topics_when_no_subscriptions() {
964        let msgbus = get_message_bus();
965        assert!(msgbus.borrow().patterns().is_empty());
966        assert!(!msgbus.borrow().has_subscribers("my-topic").unwrap());
967    }
968
969    #[rstest]
970    fn test_is_subscribed_when_no_subscriptions() {
971        let msgbus = get_message_bus();
972        let handler = get_stub_shareable_handler(None);
973
974        assert!(!msgbus.borrow().is_subscribed("my-topic", handler));
975    }
976
977    #[rstest]
978    fn test_get_response_handler_when_no_handler() {
979        let msgbus = get_message_bus();
980        let msgbus_ref = msgbus.borrow();
981        let handler = msgbus_ref.get_response_handler(&UUID4::new());
982        assert!(handler.is_none());
983    }
984
985    #[rstest]
986    fn test_get_response_handler_when_already_registered() {
987        let msgbus = get_message_bus();
988        let mut msgbus_ref = msgbus.borrow_mut();
989        let handler = get_stub_shareable_handler(None);
990
991        let request_id = UUID4::new();
992        msgbus_ref
993            .register_response_handler(&request_id, handler.clone())
994            .unwrap();
995
996        let result = msgbus_ref.register_response_handler(&request_id, handler);
997        assert!(result.is_err());
998    }
999
1000    #[rstest]
1001    fn test_get_response_handler_when_registered() {
1002        let msgbus = get_message_bus();
1003        let mut msgbus_ref = msgbus.borrow_mut();
1004        let handler = get_stub_shareable_handler(None);
1005
1006        let request_id = UUID4::new();
1007        msgbus_ref
1008            .register_response_handler(&request_id, handler)
1009            .unwrap();
1010
1011        let handler = msgbus_ref.get_response_handler(&request_id).unwrap();
1012        assert_eq!(handler.id(), handler.id());
1013    }
1014
1015    #[rstest]
1016    fn test_is_registered_when_no_registrations() {
1017        let msgbus = get_message_bus();
1018        assert!(!msgbus.borrow().is_registered("MyEndpoint"));
1019    }
1020
1021    #[rstest]
1022    fn test_register_endpoint() {
1023        let msgbus = get_message_bus();
1024        let endpoint = "MyEndpoint".into();
1025        let handler = get_stub_shareable_handler(None);
1026
1027        msgbus::register_any(endpoint, handler);
1028
1029        assert_eq!(msgbus.borrow().endpoints(), vec![endpoint.to_string()]);
1030        assert!(msgbus.borrow().get_endpoint(endpoint).is_some());
1031    }
1032
1033    #[rstest]
1034    fn test_endpoint_send() {
1035        let msgbus = get_message_bus();
1036        let endpoint = "MyEndpoint".into();
1037        let (handler, checker) = get_call_check_handler(None);
1038        let sent_count = msgbus.borrow().sent_count();
1039
1040        msgbus::register_any(endpoint, handler);
1041        assert!(msgbus.borrow().get_endpoint(endpoint).is_some());
1042        assert!(!checker.was_called());
1043
1044        msgbus::send_any(endpoint, &"Test Message");
1045
1046        assert!(checker.was_called());
1047        assert_eq!(msgbus.borrow().sent_count(), sent_count + 1);
1048    }
1049
1050    #[rstest]
1051    fn test_endpoint_send_value_increments_sent_count() {
1052        let msgbus = get_message_bus();
1053        let endpoint = "MyValueEndpoint".into();
1054        let (handler, checker) = get_call_check_handler(None);
1055        let sent_count = msgbus.borrow().sent_count();
1056
1057        msgbus::register_any(endpoint, handler);
1058        msgbus::send_any_value(endpoint, &"Test Message");
1059
1060        assert!(checker.was_called());
1061        assert_eq!(msgbus.borrow().sent_count(), sent_count + 1);
1062    }
1063
1064    #[rstest]
1065    fn test_publish_any_increments_publish_count() {
1066        let msgbus = get_message_bus();
1067        let topic = "my-published-topic";
1068        let (handler, checker) = get_call_check_handler(None);
1069        let pub_count = msgbus.borrow().pub_count();
1070
1071        msgbus::subscribe_any(topic.into(), handler, None);
1072        msgbus::publish_any(topic.into(), &"Test Message");
1073
1074        assert!(checker.was_called());
1075        assert_eq!(msgbus.borrow().pub_count(), pub_count + 1);
1076    }
1077
1078    #[rstest]
1079    fn test_deregsiter_endpoint() {
1080        let msgbus = get_message_bus();
1081        let endpoint = "MyEndpoint".into();
1082        let handler = get_stub_shareable_handler(None);
1083
1084        msgbus::register_any(endpoint, handler);
1085        msgbus::deregister_any(endpoint);
1086
1087        assert!(msgbus.borrow().endpoints().is_empty());
1088    }
1089
1090    #[rstest]
1091    fn test_subscribe() {
1092        let msgbus = get_message_bus();
1093        let topic = "my-topic";
1094        let handler = get_stub_shareable_handler(None);
1095
1096        msgbus::subscribe_any(topic.into(), handler, Some(1));
1097
1098        assert!(msgbus.borrow().has_subscribers(topic).unwrap());
1099        assert_eq!(msgbus.borrow().patterns(), vec![topic]);
1100    }
1101
1102    #[rstest]
1103    fn test_unsubscribe() {
1104        let msgbus = get_message_bus();
1105        let topic = "my-topic";
1106        let handler = get_stub_shareable_handler(None);
1107
1108        msgbus::subscribe_any(topic.into(), handler.clone(), None);
1109        msgbus::unsubscribe_any(topic.into(), &handler);
1110
1111        assert!(!msgbus.borrow().has_subscribers(topic).unwrap());
1112        assert!(msgbus.borrow().patterns().is_empty());
1113    }
1114
1115    #[rstest]
1116    fn test_subscriptions_count_rejects_invalid_topic() {
1117        let msgbus = get_message_bus();
1118
1119        let err = msgbus
1120            .borrow()
1121            .subscriptions_count("data.*")
1122            .expect_err("wildcards are invalid in topics");
1123
1124        assert_eq!(
1125            err.to_string(),
1126            "Topic `value` contained invalid characters, was data.*"
1127        );
1128    }
1129
1130    #[rstest]
1131    fn test_has_subscribers_rejects_invalid_topic() {
1132        let msgbus = get_message_bus();
1133
1134        let err = msgbus
1135            .borrow()
1136            .has_subscribers("data.*")
1137            .expect_err("wildcards are invalid in topics");
1138
1139        assert_eq!(
1140            err.to_string(),
1141            "Topic `value` contained invalid characters, was data.*"
1142        );
1143    }
1144
1145    #[rstest]
1146    fn test_subscriptions_count_any_rejects_invalid_topic() {
1147        let err = subscriptions_count_any("data.*").expect_err("wildcards are invalid in topics");
1148
1149        assert_eq!(
1150            err.to_string(),
1151            "Topic `value` contained invalid characters, was data.*"
1152        );
1153    }
1154
1155    #[rstest]
1156    fn test_matching_subscriptions() {
1157        let msgbus = get_message_bus();
1158        let pattern = "my-pattern";
1159
1160        let handler_id1 = Ustr::from("1");
1161        let handler1 = get_stub_shareable_handler(Some(handler_id1));
1162
1163        let handler_id2 = Ustr::from("2");
1164        let handler2 = get_stub_shareable_handler(Some(handler_id2));
1165
1166        let handler_id3 = Ustr::from("3");
1167        let handler3 = get_stub_shareable_handler(Some(handler_id3));
1168
1169        let handler_id4 = Ustr::from("4");
1170        let handler4 = get_stub_shareable_handler(Some(handler_id4));
1171
1172        msgbus::subscribe_any(pattern.into(), handler1, None);
1173        msgbus::subscribe_any(pattern.into(), handler2, None);
1174        msgbus::subscribe_any(pattern.into(), handler3, Some(1));
1175        msgbus::subscribe_any(pattern.into(), handler4, Some(2));
1176
1177        assert_eq!(
1178            msgbus.borrow().patterns(),
1179            vec![pattern, pattern, pattern, pattern]
1180        );
1181        assert_eq!(subscriptions_count_any(pattern).unwrap(), 4);
1182
1183        let topic = pattern;
1184        let subs = msgbus.borrow_mut().matching_subscriptions(topic);
1185        assert_eq!(subs.len(), 4);
1186        assert_eq!(subs[0].handler_id, handler_id4);
1187        assert_eq!(subs[1].handler_id, handler_id3);
1188        assert_eq!(subs[2].handler_id, handler_id1);
1189        assert_eq!(subs[3].handler_id, handler_id2);
1190    }
1191
1192    #[rstest]
1193    fn test_subscription_pattern_matching() {
1194        let msgbus = get_message_bus();
1195        let handler1 = get_stub_shareable_handler(Some(Ustr::from("1")));
1196        let handler2 = get_stub_shareable_handler(Some(Ustr::from("2")));
1197        let handler3 = get_stub_shareable_handler(Some(Ustr::from("3")));
1198
1199        msgbus::subscribe_any("data.quotes.*".into(), handler1, None);
1200        msgbus::subscribe_any("data.trades.*".into(), handler2, None);
1201        msgbus::subscribe_any("data.*.BINANCE.*".into(), handler3, None);
1202        assert_eq!(msgbus.borrow().subscriptions().len(), 3);
1203
1204        let topic = "data.quotes.BINANCE.ETHUSDT";
1205        assert_eq!(msgbus.borrow().find_topic_matches(topic.into()).len(), 2);
1206
1207        let matches = msgbus.borrow_mut().matching_subscriptions(topic);
1208        assert_eq!(matches.len(), 2);
1209        assert_eq!(matches[0].handler_id, Ustr::from("3"));
1210        assert_eq!(matches[1].handler_id, Ustr::from("1"));
1211    }
1212
1213    #[rstest]
1214    fn test_late_wildcard_subscription_receives_cached_topic() {
1215        let msgbus = get_message_bus();
1216        let topic = "data.instrument.POLYMARKET.TEST-SYMBOL";
1217
1218        let (early_handler, early_saver) =
1219            get_any_saving_handler::<String>(Some(Ustr::from("early")));
1220        msgbus::subscribe_any("data.*.POLYMARKET.*".into(), early_handler, None);
1221
1222        msgbus::publish_any(topic.into(), &"ONE".to_string());
1223
1224        let (late_handler, late_saver) = get_any_saving_handler::<String>(Some(Ustr::from("late")));
1225        msgbus::subscribe_any("data.instrument.POLYMARKET.*".into(), late_handler, None);
1226
1227        msgbus::publish_any(topic.into(), &"TWO".to_string());
1228
1229        assert_eq!(early_saver.get_messages(), vec!["ONE", "TWO"]);
1230        assert_eq!(late_saver.get_messages(), vec!["TWO"]);
1231
1232        let topic_mstr: MStr<Topic> = topic.into();
1233        let cached = msgbus.borrow_mut().matching_subscriptions(topic_mstr);
1234        assert_eq!(cached.len(), 2);
1235    }
1236
1237    #[rstest]
1238    fn test_late_wildcard_backfills_into_multiple_cached_topics() {
1239        let msgbus = get_message_bus();
1240        let topics = ["data.A", "data.B", "data.C"];
1241
1242        let (early_handler, early_saver) =
1243            get_any_saving_handler::<String>(Some(Ustr::from("early")));
1244        msgbus::subscribe_any("data.*".into(), early_handler, None);
1245
1246        for topic in &topics {
1247            msgbus::publish_any((*topic).into(), &(*topic).to_string());
1248        }
1249
1250        let (late_handler, late_saver) = get_any_saving_handler::<String>(Some(Ustr::from("late")));
1251        msgbus::subscribe_any("data.*".into(), late_handler, None);
1252
1253        for topic in &topics {
1254            msgbus::publish_any((*topic).into(), &format!("{topic}-2"));
1255        }
1256
1257        assert_eq!(
1258            early_saver.get_messages(),
1259            vec![
1260                "data.A", "data.B", "data.C", "data.A-2", "data.B-2", "data.C-2"
1261            ],
1262        );
1263        assert_eq!(
1264            late_saver.get_messages(),
1265            vec!["data.A-2", "data.B-2", "data.C-2"]
1266        );
1267
1268        for topic in &topics {
1269            let topic_mstr: MStr<Topic> = (*topic).into();
1270            assert_eq!(
1271                msgbus.borrow_mut().matching_subscriptions(topic_mstr).len(),
1272                2,
1273                "topic {topic} should have both subscribers cached",
1274            );
1275        }
1276    }
1277
1278    /// A simple reference model for subscription behavior.
1279    struct SimpleSubscriptionModel {
1280        /// Stores (pattern, `handler_id`) tuples for active subscriptions.
1281        subscriptions: Vec<(String, String)>,
1282    }
1283
1284    impl SimpleSubscriptionModel {
1285        fn new() -> Self {
1286            Self {
1287                subscriptions: Vec::new(),
1288            }
1289        }
1290
1291        fn subscribe(&mut self, pattern: &str, handler_id: &str) {
1292            let subscription = (pattern.to_string(), handler_id.to_string());
1293            if !self.subscriptions.contains(&subscription) {
1294                self.subscriptions.push(subscription);
1295            }
1296        }
1297
1298        fn unsubscribe(&mut self, pattern: &str, handler_id: &str) -> bool {
1299            let subscription = (pattern.to_string(), handler_id.to_string());
1300            if let Some(idx) = self.subscriptions.iter().position(|s| s == &subscription) {
1301                self.subscriptions.remove(idx);
1302                true
1303            } else {
1304                false
1305            }
1306        }
1307
1308        fn is_subscribed(&self, pattern: &str, handler_id: &str) -> bool {
1309            self.subscriptions
1310                .contains(&(pattern.to_string(), handler_id.to_string()))
1311        }
1312
1313        fn matching_subscriptions(&self, topic: &str) -> Vec<(String, String)> {
1314            let topic = topic.into();
1315
1316            self.subscriptions
1317                .iter()
1318                .filter(|(pat, _)| is_matching_backtracking(topic, pat.into()))
1319                .map(|(pat, id)| (pat.clone(), id.clone()))
1320                .collect()
1321        }
1322
1323        fn subscription_count(&self) -> usize {
1324            self.subscriptions.len()
1325        }
1326    }
1327
1328    #[rstest]
1329    fn subscription_model_fuzz_testing() {
1330        let mut rng = StdRng::seed_from_u64(42);
1331
1332        let msgbus = get_message_bus();
1333        let mut model = SimpleSubscriptionModel::new();
1334
1335        // Map from handler_id to handler
1336        let mut handlers: Vec<(String, ShareableMessageHandler)> = Vec::new();
1337
1338        // Generate some patterns
1339        let patterns = generate_test_patterns(&mut rng);
1340
1341        // Generate some handler IDs
1342        let handler_ids: Vec<String> = (0..50).map(|i| format!("handler_{i}")).collect();
1343
1344        // Initialize handlers
1345        for id in &handler_ids {
1346            let handler = get_stub_shareable_handler(Some(Ustr::from(id)));
1347            handlers.push((id.clone(), handler));
1348        }
1349
1350        let num_operations = 50_000;
1351        for op_num in 0..num_operations {
1352            let operation = rng.random_range(0..4);
1353
1354            match operation {
1355                // Subscribe
1356                0 => {
1357                    let pattern_idx = rng.random_range(0..patterns.len());
1358                    let handler_idx = rng.random_range(0..handlers.len());
1359                    let pattern = &patterns[pattern_idx];
1360                    let (handler_id, handler) = &handlers[handler_idx];
1361
1362                    // Apply to reference model
1363                    model.subscribe(pattern, handler_id);
1364
1365                    // Apply to message bus
1366                    msgbus::subscribe_any(pattern.as_str().into(), handler.clone(), None);
1367
1368                    assert_eq!(
1369                        model.subscription_count(),
1370                        msgbus.borrow().subscriptions().len()
1371                    );
1372
1373                    assert!(
1374                        msgbus.borrow().is_subscribed(pattern, handler.clone()),
1375                        "Op {op_num}: is_subscribed should return true after subscribe"
1376                    );
1377                }
1378
1379                // Unsubscribe
1380                1 => {
1381                    if model.subscription_count() > 0 {
1382                        let sub_idx = rng.random_range(0..model.subscription_count());
1383                        let (pattern, handler_id) = model.subscriptions[sub_idx].clone();
1384
1385                        // Apply to reference model
1386                        model.unsubscribe(&pattern, &handler_id);
1387
1388                        // Find handler
1389                        let handler = handlers
1390                            .iter()
1391                            .find(|(id, _)| id == &handler_id)
1392                            .map(|(_, h)| h.clone())
1393                            .unwrap();
1394
1395                        // Apply to message bus
1396                        msgbus::unsubscribe_any(pattern.as_str().into(), &handler);
1397
1398                        assert_eq!(
1399                            model.subscription_count(),
1400                            msgbus.borrow().subscriptions().len()
1401                        );
1402                        assert!(
1403                            !msgbus.borrow().is_subscribed(pattern, handler.clone()),
1404                            "Op {op_num}: is_subscribed should return false after unsubscribe"
1405                        );
1406                    }
1407                }
1408
1409                // Check is_subscribed
1410                2 => {
1411                    // Get a random pattern and handler
1412                    let pattern_idx = rng.random_range(0..patterns.len());
1413                    let handler_idx = rng.random_range(0..handlers.len());
1414                    let pattern = &patterns[pattern_idx];
1415                    let (handler_id, handler) = &handlers[handler_idx];
1416
1417                    let expected = model.is_subscribed(pattern, handler_id);
1418                    let actual = msgbus.borrow().is_subscribed(pattern, handler.clone());
1419
1420                    assert_eq!(
1421                        expected, actual,
1422                        "Op {op_num}: Subscription state mismatch for pattern '{pattern}', handler '{handler_id}': expected={expected}, actual={actual}"
1423                    );
1424                }
1425
1426                // Check matching_subscriptions
1427                3 => {
1428                    // Generate a topic
1429                    let topic = create_topic(&mut rng);
1430
1431                    let actual_matches = msgbus.borrow_mut().matching_subscriptions(topic);
1432                    let expected_matches = model.matching_subscriptions(&topic);
1433
1434                    assert_eq!(
1435                        expected_matches.len(),
1436                        actual_matches.len(),
1437                        "Op {}: Match count mismatch for topic '{}': expected={}, actual={}",
1438                        op_num,
1439                        topic,
1440                        expected_matches.len(),
1441                        actual_matches.len()
1442                    );
1443
1444                    for sub in &actual_matches {
1445                        assert!(
1446                            expected_matches
1447                                .contains(&(sub.pattern.to_string(), sub.handler_id.to_string())),
1448                            "Op {}: Expected match not found: pattern='{}', handler_id='{}'",
1449                            op_num,
1450                            sub.pattern,
1451                            sub.handler_id
1452                        );
1453                    }
1454                }
1455                _ => unreachable!(),
1456            }
1457        }
1458    }
1459
1460    fn generate_pattern_from_topic(topic: &str, rng: &mut StdRng) -> String {
1461        let mut pattern = String::new();
1462
1463        for c in topic.chars() {
1464            let val: f64 = rng.random();
1465            if val < 0.1 {
1466                pattern.push('*');
1467            } else if val < 0.3 {
1468                pattern.push('?');
1469            } else if val >= 0.5 {
1470                pattern.push(c);
1471            }
1472        }
1473
1474        pattern
1475    }
1476
1477    fn generate_test_patterns(rng: &mut StdRng) -> Vec<String> {
1478        let mut patterns = vec![
1479            "data.*.*.*".to_string(),
1480            "*.*.BINANCE.*".to_string(),
1481            "events.order.*".to_string(),
1482            "data.*.*.?USDT".to_string(),
1483            "*.trades.*.BTC*".to_string(),
1484            "*.*.*.*".to_string(),
1485        ];
1486
1487        // Add some random patterns
1488        for _ in 0..50 {
1489            match rng.random_range(0..10) {
1490                // Use existing pattern
1491                0..=1 => {
1492                    let idx = rng.random_range(0..patterns.len());
1493                    patterns.push(patterns[idx].clone());
1494                }
1495                // Generate new pattern from topic
1496                _ => {
1497                    let topic = create_topic(rng);
1498                    let pattern = generate_pattern_from_topic(&topic, rng);
1499                    patterns.push(pattern);
1500                }
1501            }
1502        }
1503
1504        patterns
1505    }
1506
1507    fn create_topic(rng: &mut StdRng) -> Ustr {
1508        let cat = ["data", "info", "order"];
1509        let model = ["quotes", "trades", "orderbooks", "depths"];
1510        let venue = ["BINANCE", "BYBIT", "OKX", "FTX", "KRAKEN"];
1511        let instrument = ["BTCUSDT", "ETHUSDT", "SOLUSDT", "XRPUSDT", "DOGEUSDT"];
1512
1513        let cat = cat[rng.random_range(0..cat.len())];
1514        let model = model[rng.random_range(0..model.len())];
1515        let venue = venue[rng.random_range(0..venue.len())];
1516        let instrument = instrument[rng.random_range(0..instrument.len())];
1517        Ustr::from(&format!("{cat}.{model}.{venue}.{instrument}"))
1518    }
1519}
nautilus_common/msgbus/core.rs

nautilus_common/msgbus/
core.rs
