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#[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_depth10: TopicRouter<OrderBookDepth10>,
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<Box<dyn MessageBusExternalEgress>>,
290    encoding: SerializationEncoding,
291    encoding_market_data: Option<SerializationEncoding>,
292    encoding_builtin: Option<SerializationEncoding>,
293    types_filter: AHashSet<BusPayloadType>,
294    streaming_types: AHashSet<BusPayloadType>,
295}
296
297impl Debug for MessageBus {
298    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
299        f.debug_struct(stringify!(MessageBus))
300            .field("trader_id", &self.trader_id)
301            .field("instance_id", &self.instance_id)
302            .field("name", &self.name)
303            .field("has_backing", &self.has_backing)
304            .field("external_egress", &self.external_egress.is_some())
305            .finish_non_exhaustive()
306    }
307}
308
309impl Default for MessageBus {
310    /// Creates a new default [`MessageBus`] instance.
311    fn default() -> Self {
312        Self::new(TraderId::from("TRADER-001"), UUID4::new(), None, None)
313    }
314}
315
316impl MessageBus {
317    /// Creates a new [`MessageBus`] instance.
318    #[must_use]
319    pub fn new(
320        trader_id: TraderId,
321        instance_id: UUID4,
322        name: Option<String>,
323        _config: Option<HashMap<String, serde_json::Value>>,
324    ) -> Self {
325        Self {
326            trader_id,
327            instance_id,
328            name: name.unwrap_or(stringify!(MessageBus).to_owned()),
329            switchboard: MessagingSwitchboard::default(),
330            subscriptions: AHashSet::new(),
331            topics: IndexMap::new(),
332            endpoints: IndexMap::new(),
333            correlation_index: AHashMap::new(),
334            has_backing: false,
335            router_quotes: TopicRouter::new(),
336            router_trades: TopicRouter::new(),
337            router_bars: TopicRouter::new(),
338            router_deltas: TopicRouter::new(),
339            router_depth10: TopicRouter::new(),
340            router_book_snapshots: TopicRouter::new(),
341            router_mark_prices: TopicRouter::new(),
342            router_index_prices: TopicRouter::new(),
343            router_funding_rates: TopicRouter::new(),
344            router_order_events: TopicRouter::new(),
345            router_position_events: TopicRouter::new(),
346            router_account_state: TopicRouter::new(),
347            router_portfolio: TopicRouter::new(),
348            router_orders: TopicRouter::new(),
349            router_positions: TopicRouter::new(),
350            router_greeks: TopicRouter::new(),
351            router_option_greeks: TopicRouter::new(),
352            router_option_chain: TopicRouter::new(),
353            router_instruments: TopicRouter::new(),
354            #[cfg(feature = "defi")]
355            router_defi_blocks: TopicRouter::new(),
356            #[cfg(feature = "defi")]
357            router_defi_pools: TopicRouter::new(),
358            #[cfg(feature = "defi")]
359            router_defi_swaps: TopicRouter::new(),
360            #[cfg(feature = "defi")]
361            router_defi_liquidity: TopicRouter::new(),
362            #[cfg(feature = "defi")]
363            router_defi_collects: TopicRouter::new(),
364            #[cfg(feature = "defi")]
365            router_defi_flash: TopicRouter::new(),
366            #[cfg(feature = "defi")]
367            endpoints_defi_data: IntoEndpointMap::new(),
368            endpoints_quotes: EndpointMap::new(),
369            endpoints_trades: EndpointMap::new(),
370            endpoints_bars: EndpointMap::new(),
371            endpoints_account_state: EndpointMap::new(),
372            endpoints_trading_commands: IntoEndpointMap::new(),
373            endpoints_data_commands: IntoEndpointMap::new(),
374            endpoints_data_responses: IntoEndpointMap::new(),
375            endpoints_exec_reports: IntoEndpointMap::new(),
376            endpoints_order_events: IntoEndpointMap::new(),
377            endpoints_data: IntoEndpointMap::new(),
378            routers_typed: AHashMap::new(),
379            endpoints_typed: AHashMap::new(),
380            sent_count: 0,
381            req_count: 0,
382            res_count: 0,
383            pub_count: 0,
384            external_egress: None,
385            encoding: SerializationEncoding::Json,
386            encoding_market_data: None,
387            encoding_builtin: None,
388            types_filter: AHashSet::new(),
389            streaming_types: AHashSet::new(),
390        }
391    }
392
393    /// Registers message bus for the current thread.
394    pub fn register_message_bus(self) -> Rc<RefCell<Self>> {
395        let msgbus = Rc::new(RefCell::new(self));
396        set_message_bus(msgbus.clone());
397        msgbus
398    }
399
400    /// Gets or creates a typed router for custom message type `T`.
401    ///
402    /// # Panics
403    ///
404    /// Panics if the stored router type doesn't match `T` (internal bug).
405    pub fn router<T: 'static>(&mut self) -> &mut TopicRouter<T> {
406        self.routers_typed
407            .entry(TypeId::of::<T>())
408            .or_insert_with(|| Box::new(TopicRouter::<T>::new()))
409            .downcast_mut::<TopicRouter<T>>()
410            .expect("TopicRouter type mismatch - this is a bug")
411    }
412
413    /// Gets or creates a typed endpoint map for custom message type `T`.
414    ///
415    /// # Panics
416    ///
417    /// Panics if the stored endpoint map type doesn't match `T` (internal bug).
418    pub fn endpoint_map<T: 'static>(&mut self) -> &mut EndpointMap<T> {
419        self.endpoints_typed
420            .entry(TypeId::of::<T>())
421            .or_insert_with(|| Box::new(EndpointMap::<T>::new()))
422            .downcast_mut::<EndpointMap<T>>()
423            .expect("EndpointMap type mismatch - this is a bug")
424    }
425
426    /// Sets external egress for serialized published messages.
427    pub fn set_external_egress(
428        &mut self,
429        external_egress: Box<dyn MessageBusExternalEgress>,
430        encoding: SerializationEncoding,
431    ) {
432        self.external_egress = Some(external_egress);
433        self.encoding = encoding;
434        self.encoding_market_data = None;
435        self.encoding_builtin = None;
436        self.has_backing = true;
437        HAS_EXTERNAL_EGRESS.with(|flag| flag.set(true));
438    }
439
440    /// Sets external egress and category encoding policy from a validated config.
441    ///
442    /// # Errors
443    ///
444    /// Returns a [`crate::config::ConfigError`] if the config selects an unsupported encoding.
445    pub fn set_external_egress_config(
446        &mut self,
447        external_egress: Box<dyn MessageBusExternalEgress>,
448        config: &MessageBusConfig,
449    ) -> crate::config::ConfigResult<()> {
450        config.validate()?;
451
452        self.external_egress = Some(external_egress);
453        self.encoding = config.encoding;
454        self.encoding_market_data = config.encoding_market_data;
455        self.encoding_builtin = config.encoding_builtin;
456        self.has_backing = true;
457        HAS_EXTERNAL_EGRESS.with(|flag| flag.set(true));
458
459        Ok(())
460    }
461
462    /// Sets the type names excluded from external publishing.
463    pub fn set_types_filter(&mut self, filter: Vec<String>) {
464        self.types_filter = filter
465            .into_iter()
466            .map(|type_name| BusPayloadType::from_name(&type_name))
467            .filter(|payload_type| !payload_type.as_str().is_empty())
468            .collect();
469    }
470
471    /// Registers a payload type for external-to-internal streaming.
472    pub fn add_streaming_type(&mut self, payload_type: BusPayloadType) {
473        if !payload_type.as_str().is_empty() {
474            self.streaming_types.insert(payload_type);
475        }
476    }
477
478    /// Returns whether the payload type is registered for external-to-internal streaming.
479    #[must_use]
480    pub fn is_streaming_type(&self, payload_type: BusPayloadType) -> bool {
481        !payload_type.as_str().is_empty() && self.streaming_types.contains(&payload_type)
482    }
483
484    /// Clears all payload types registered for external-to-internal streaming.
485    pub fn clear_streaming_types(&mut self) {
486        self.streaming_types.clear();
487    }
488
489    #[must_use]
490    pub(crate) fn has_external_egress(&self) -> bool {
491        self.external_egress.is_some()
492    }
493
494    pub(crate) fn external_egress(&self) -> Option<&dyn MessageBusExternalEgress> {
495        self.external_egress.as_deref()
496    }
497
498    pub(crate) fn encoding_for(&self, payload_type: BusPayloadType) -> SerializationEncoding {
499        match payload_type.category() {
500            BusPayloadCategory::MarketData => self.encoding_market_data.unwrap_or(self.encoding),
501            BusPayloadCategory::BuiltIn => self.encoding_builtin.unwrap_or(self.encoding),
502            BusPayloadCategory::Other => self.encoding,
503        }
504    }
505
506    pub(crate) fn types_filter(&self) -> &AHashSet<BusPayloadType> {
507        &self.types_filter
508    }
509
510    /// Disposes of the message bus, clearing all subscriptions, endpoints,
511    /// and handler references.
512    pub fn dispose(&mut self) {
513        self.subscriptions.clear();
514        self.topics.clear();
515        self.endpoints.clear();
516        self.correlation_index.clear();
517
518        self.router_quotes.clear();
519        self.router_trades.clear();
520        self.router_bars.clear();
521        self.router_deltas.clear();
522        self.router_depth10.clear();
523        self.router_book_snapshots.clear();
524        self.router_mark_prices.clear();
525        self.router_index_prices.clear();
526        self.router_funding_rates.clear();
527        self.router_order_events.clear();
528        self.router_position_events.clear();
529        self.router_account_state.clear();
530        self.router_portfolio.clear();
531        self.router_orders.clear();
532        self.router_positions.clear();
533        self.router_greeks.clear();
534        self.router_option_greeks.clear();
535        self.router_option_chain.clear();
536        self.router_instruments.clear();
537
538        #[cfg(feature = "defi")]
539        {
540            self.router_defi_blocks.clear();
541            self.router_defi_pools.clear();
542            self.router_defi_swaps.clear();
543            self.router_defi_liquidity.clear();
544            self.router_defi_collects.clear();
545            self.router_defi_flash.clear();
546            self.endpoints_defi_data.clear();
547        }
548
549        self.endpoints_quotes.clear();
550        self.endpoints_trades.clear();
551        self.endpoints_bars.clear();
552        self.endpoints_account_state.clear();
553        self.endpoints_trading_commands.clear();
554        self.endpoints_data_commands.clear();
555        self.endpoints_data_responses.clear();
556        self.endpoints_exec_reports.clear();
557        self.endpoints_order_events.clear();
558        self.endpoints_data.clear();
559
560        self.routers_typed.clear();
561        self.endpoints_typed.clear();
562        self.clear_streaming_types();
563        self.sent_count = 0;
564        self.req_count = 0;
565        self.res_count = 0;
566        self.pub_count = 0;
567
568        if let Some(mut external_egress) = self.external_egress.take() {
569            external_egress.close();
570        }
571        self.has_backing = false;
572        HAS_EXTERNAL_EGRESS.with(|flag| flag.set(false));
573    }
574
575    /// Returns the memory address of this instance as a hexadecimal string.
576    #[must_use]
577    pub fn mem_address(&self) -> String {
578        format!("{self:p}")
579    }
580
581    /// Returns a reference to the switchboard.
582    #[must_use]
583    pub fn switchboard(&self) -> &MessagingSwitchboard {
584        &self.switchboard
585    }
586
587    /// Returns the total count of messages sent to endpoints.
588    #[must_use]
589    pub const fn sent_count(&self) -> u64 {
590        self.sent_count
591    }
592
593    /// Returns the total count of requests sent to endpoints.
594    #[must_use]
595    pub const fn req_count(&self) -> u64 {
596        self.req_count
597    }
598
599    /// Returns the total count of responses sent to registered handlers.
600    #[must_use]
601    pub const fn res_count(&self) -> u64 {
602        self.res_count
603    }
604
605    /// Returns the total count of messages published to topics.
606    #[must_use]
607    pub const fn pub_count(&self) -> u64 {
608        self.pub_count
609    }
610
611    pub(crate) fn increment_sent_count(&mut self) {
612        self.sent_count += 1;
613    }
614
615    pub(crate) fn increment_req_count(&mut self) {
616        self.req_count += 1;
617    }
618
619    pub(crate) fn increment_res_count(&mut self) {
620        self.res_count += 1;
621    }
622
623    pub(crate) fn increment_pub_count(&mut self) {
624        self.pub_count += 1;
625    }
626
627    /// Returns the registered endpoint addresses.
628    #[must_use]
629    pub fn endpoints(&self) -> Vec<&str> {
630        self.endpoints.iter().map(|e| e.0.as_str()).collect()
631    }
632
633    /// Returns actively subscribed patterns.
634    #[must_use]
635    pub fn patterns(&self) -> Vec<&str> {
636        self.subscriptions
637            .iter()
638            .map(|s| s.pattern.as_str())
639            .collect()
640    }
641
642    /// Returns whether there are subscribers for the `topic`.
643    ///
644    /// # Errors
645    ///
646    /// Returns an error if the `topic` is not a valid topic string.
647    pub fn has_subscribers<T: AsRef<str>>(&self, topic: T) -> anyhow::Result<bool> {
648        Ok(self.subscriptions_count(topic)? > 0)
649    }
650
651    /// Returns the count of subscribers for the `topic`.
652    ///
653    /// # Errors
654    ///
655    /// Returns an error if the `topic` is not a valid topic string.
656    pub fn subscriptions_count<T: AsRef<str>>(&self, topic: T) -> anyhow::Result<usize> {
657        let topic = MStr::<Topic>::topic(topic)?;
658        Ok(self
659            .topics
660            .get(&topic)
661            .map_or_else(|| self.find_topic_matches(topic).len(), Vec::len))
662    }
663
664    /// Returns active subscriptions.
665    #[must_use]
666    pub fn subscriptions(&self) -> Vec<&Subscription> {
667        self.subscriptions.iter().collect()
668    }
669
670    /// Returns the handler IDs for actively subscribed patterns.
671    #[must_use]
672    pub fn subscription_handler_ids(&self) -> Vec<&str> {
673        self.subscriptions
674            .iter()
675            .map(|s| s.handler_id.as_str())
676            .collect()
677    }
678
679    /// Returns whether the endpoint is registered.
680    ///
681    /// # Panics
682    ///
683    /// Panics if the `endpoint` conversion to `MStr<Endpoint>` fails.
684    #[must_use]
685    pub fn is_registered<T: Into<MStr<Endpoint>>>(&self, endpoint: T) -> bool {
686        let endpoint: MStr<Endpoint> = endpoint.into();
687        self.endpoints.contains_key(&endpoint)
688    }
689
690    /// Returns whether the `handler` is subscribed to the `pattern`.
691    #[must_use]
692    pub fn is_subscribed<T: AsRef<str>>(
693        &self,
694        pattern: T,
695        handler: ShareableMessageHandler,
696    ) -> bool {
697        let pattern = MStr::<Pattern>::pattern(pattern);
698        let sub = Subscription::new(pattern, handler, None);
699        self.subscriptions.contains(&sub)
700    }
701
702    /// Close the message bus which will close the sender channel and join the thread.
703    ///
704    /// # Errors
705    ///
706    /// This function never returns an error (TBD once backing database added).
707    pub fn close(&mut self) -> anyhow::Result<()> {
708        if let Some(mut external_egress) = self.external_egress.take() {
709            external_egress.close();
710        }
711        self.has_backing = false;
712        HAS_EXTERNAL_EGRESS.with(|flag| flag.set(false));
713        Ok(())
714    }
715
716    /// Returns the handler for the `endpoint`.
717    #[must_use]
718    pub fn get_endpoint(&self, endpoint: MStr<Endpoint>) -> Option<&ShareableMessageHandler> {
719        self.endpoints.get(&endpoint)
720    }
721
722    /// Returns the handler for the `correlation_id`.
723    #[must_use]
724    pub fn get_response_handler(&self, correlation_id: &UUID4) -> Option<&ShareableMessageHandler> {
725        self.correlation_index.get(correlation_id)
726    }
727
728    /// Removes and returns the handler for the `correlation_id`.
729    pub(crate) fn take_response_handler(
730        &mut self,
731        correlation_id: &UUID4,
732    ) -> Option<ShareableMessageHandler> {
733        self.correlation_index.remove(correlation_id)
734    }
735
736    /// Finds the subscriptions with pattern matching the `topic`.
737    fn find_topic_matches(&self, topic: MStr<Topic>) -> Vec<Subscription> {
738        self.subscriptions
739            .iter()
740            .filter_map(|sub| {
741                if is_matching_backtracking(topic, sub.pattern) {
742                    Some(sub.clone())
743                } else {
744                    None
745                }
746            })
747            .collect()
748    }
749
750    /// Finds the subscriptions which match the `topic` and caches the
751    /// results in the `patterns` map.
752    #[must_use]
753    pub fn matching_subscriptions<T: Into<MStr<Topic>>>(&mut self, topic: T) -> Vec<Subscription> {
754        self.inner_matching_subscriptions(topic.into())
755    }
756
757    fn inner_matching_subscriptions(&mut self, topic: MStr<Topic>) -> Vec<Subscription> {
758        self.topics.get(&topic).cloned().unwrap_or_else(|| {
759            let mut matches = self.find_topic_matches(topic);
760            matches.sort_by(Subscription::delivery_order);
761            self.topics.insert(topic, matches.clone());
762            matches
763        })
764    }
765
766    /// Fills a buffer with handlers matching a topic.
767    pub(crate) fn fill_matching_any_handlers(
768        &mut self,
769        topic: MStr<Topic>,
770        buf: &mut SmallVec<[ShareableMessageHandler; 64]>,
771    ) {
772        if let Some(subs) = self.topics.get(&topic) {
773            for sub in subs {
774                buf.push(sub.handler.clone());
775            }
776        } else {
777            let mut matches = self.find_topic_matches(topic);
778            matches.sort_by(Subscription::delivery_order);
779
780            for sub in &matches {
781                buf.push(sub.handler.clone());
782            }
783
784            self.topics.insert(topic, matches);
785        }
786    }
787
788    /// Registers a response handler for a specific correlation ID.
789    ///
790    /// # Errors
791    ///
792    /// Returns an error if `handler` is already registered for the `correlation_id`.
793    pub fn register_response_handler(
794        &mut self,
795        correlation_id: &UUID4,
796        handler: ShareableMessageHandler,
797    ) -> anyhow::Result<()> {
798        if self.correlation_index.contains_key(correlation_id) {
799            anyhow::bail!("Correlation ID <{correlation_id}> already has a registered handler");
800        }
801
802        self.correlation_index.insert(*correlation_id, handler);
803
804        Ok(())
805    }
806}
807
808#[cfg(test)]
809mod tests {
810    use std::{
811        any::Any,
812        cell::RefCell,
813        collections::hash_map::DefaultHasher,
814        fmt::Debug,
815        hash::{Hash, Hasher},
816        rc::Rc,
817    };
818
819    use rand::{RngExt, SeedableRng, rngs::StdRng};
820    use rstest::rstest;
821    use ustr::Ustr;
822
823    use super::*;
824    use crate::msgbus::{
825        self, Handler, ShareableMessageHandler, get_message_bus,
826        matching::is_matching_backtracking,
827        stubs::{get_any_saving_handler, get_call_check_handler, get_stub_shareable_handler},
828        subscriptions_count_any,
829        typed_handler::shareable_handler,
830    };
831
832    #[derive(Debug)]
833    struct RecordingAnyHandler {
834        id: Ustr,
835        label: &'static str,
836        order: Rc<RefCell<Vec<&'static str>>>,
837    }
838
839    impl Handler<dyn Any> for RecordingAnyHandler {
840        fn id(&self) -> Ustr {
841            self.id
842        }
843
844        fn handle(&self, _message: &dyn Any) {
845            self.order.borrow_mut().push(self.label);
846        }
847    }
848
849    fn recording_any_handler(
850        id: &'static str,
851        label: &'static str,
852        order: Rc<RefCell<Vec<&'static str>>>,
853    ) -> ShareableMessageHandler {
854        shareable_handler(Rc::new(RecordingAnyHandler {
855            id: Ustr::from(id),
856            label,
857            order,
858        }))
859    }
860
861    #[rstest]
862    fn test_subscription_ordering_laws() {
863        let handler = get_stub_shareable_handler(Some(Ustr::from("handler-b")));
864        let base = Subscription::new("pattern-b".into(), handler.clone(), Some(1));
865        let different_priority = Subscription::new("pattern-b".into(), handler, Some(2));
866        let different_pattern = Subscription::new(
867            "pattern-a".into(),
868            get_stub_shareable_handler(Some(Ustr::from("handler-b"))),
869            Some(1),
870        );
871        let different_handler = Subscription::new(
872            "pattern-b".into(),
873            get_stub_shareable_handler(Some(Ustr::from("handler-a"))),
874            Some(1),
875        );
876
877        assert_eq!(base, different_priority);
878        assert_eq!(base.cmp(&different_priority), std::cmp::Ordering::Equal);
879
880        let mut base_hasher = DefaultHasher::new();
881        base.hash(&mut base_hasher);
882        let mut different_priority_hasher = DefaultHasher::new();
883        different_priority.hash(&mut different_priority_hasher);
884        assert_eq!(base_hasher.finish(), different_priority_hasher.finish());
885
886        let variants = [
887            base,
888            different_priority,
889            different_pattern,
890            different_handler,
891        ];
892
893        for a in &variants {
894            for b in &variants {
895                assert_eq!(a == b, a.cmp(b).is_eq());
896                assert_eq!(a.partial_cmp(b), Some(a.cmp(b)));
897                assert_eq!(a.cmp(b), b.cmp(a).reverse());
898            }
899        }
900    }
901
902    #[rstest]
903    fn test_new() {
904        let trader_id = TraderId::default();
905        let msgbus = MessageBus::new(trader_id, UUID4::new(), None, None);
906
907        assert_eq!(msgbus.trader_id, trader_id);
908        assert_eq!(msgbus.name, stringify!(MessageBus));
909    }
910
911    #[rstest]
912    fn encoding_for_uses_market_data_override() {
913        let msgbus = MessageBus {
914            encoding: SerializationEncoding::Json,
915            encoding_market_data: Some(SerializationEncoding::MsgPack),
916            ..Default::default()
917        };
918
919        assert_eq!(
920            msgbus.encoding_for(BusPayloadType::QuoteTick),
921            SerializationEncoding::MsgPack
922        );
923        assert_eq!(
924            msgbus.encoding_for(BusPayloadType::Custom(Ustr::from("CustomPayload"))),
925            SerializationEncoding::Json
926        );
927    }
928
929    #[rstest]
930    fn encoding_for_uses_builtin_override() {
931        let msgbus = MessageBus {
932            encoding: SerializationEncoding::Json,
933            encoding_builtin: Some(SerializationEncoding::MsgPack),
934            ..Default::default()
935        };
936
937        assert_eq!(
938            msgbus.encoding_for(BusPayloadType::OrderEvent),
939            SerializationEncoding::MsgPack
940        );
941        assert_eq!(
942            msgbus.encoding_for(BusPayloadType::Instrument),
943            SerializationEncoding::Json
944        );
945    }
946
947    #[rstest]
948    fn encoding_for_uses_default_without_category_override() {
949        let msgbus = MessageBus {
950            encoding: SerializationEncoding::MsgPack,
951            ..Default::default()
952        };
953
954        assert_eq!(
955            msgbus.encoding_for(BusPayloadType::QuoteTick),
956            SerializationEncoding::MsgPack
957        );
958        assert_eq!(
959            msgbus.encoding_for(BusPayloadType::OrderEvent),
960            SerializationEncoding::MsgPack
961        );
962        assert_eq!(
963            msgbus.encoding_for(BusPayloadType::Custom(Ustr::from("CustomPayload"))),
964            SerializationEncoding::MsgPack
965        );
966    }
967
968    #[rstest]
969    fn set_types_filter_resolves_canonical_and_custom_names() {
970        let mut msgbus = MessageBus::default();
971
972        msgbus.set_types_filter(vec![
973            "QuoteTick".to_string(),
974            "ExternalCustomPayload".to_string(),
975            String::new(),
976        ]);
977
978        let filter = msgbus.types_filter();
979        assert_eq!(filter.len(), 2);
980        assert!(filter.contains(&BusPayloadType::QuoteTick));
981        assert!(filter.contains(&BusPayloadType::Custom(Ustr::from("ExternalCustomPayload"))));
982        assert!(!filter.contains(&BusPayloadType::Custom(Ustr::default())));
983    }
984
985    #[rstest]
986    fn streaming_type_registration_uses_canonical_payload_names() {
987        let mut msgbus = MessageBus::default();
988
989        msgbus.add_streaming_type(BusPayloadType::QuoteTick);
990        msgbus.add_streaming_type(BusPayloadType::Custom(Ustr::from("CustomPayload")));
991
992        assert!(msgbus.is_streaming_type(BusPayloadType::QuoteTick));
993        assert!(msgbus.is_streaming_type(BusPayloadType::Custom(Ustr::from("CustomPayload"))));
994        assert!(msgbus.streaming_types.contains(&BusPayloadType::QuoteTick));
995        assert!(
996            msgbus
997                .streaming_types
998                .contains(&BusPayloadType::Custom(Ustr::from("CustomPayload")))
999        );
1000        assert!(!msgbus.is_streaming_type(BusPayloadType::TradeTick));
1001    }
1002
1003    #[rstest]
1004    fn streaming_type_registration_ignores_empty_custom_payload_type() {
1005        let mut msgbus = MessageBus::default();
1006
1007        msgbus.add_streaming_type(BusPayloadType::Custom(Ustr::default()));
1008
1009        assert!(!msgbus.is_streaming_type(BusPayloadType::Custom(Ustr::default())));
1010        assert!(msgbus.streaming_types.is_empty());
1011    }
1012
1013    #[rstest]
1014    fn clear_streaming_types_removes_registered_types() {
1015        let mut msgbus = MessageBus::default();
1016        msgbus.add_streaming_type(BusPayloadType::QuoteTick);
1017
1018        msgbus.clear_streaming_types();
1019
1020        assert!(!msgbus.is_streaming_type(BusPayloadType::QuoteTick));
1021    }
1022
1023    #[rstest]
1024    fn dispose_clears_streaming_types() {
1025        let mut msgbus = MessageBus::default();
1026        msgbus.add_streaming_type(BusPayloadType::QuoteTick);
1027
1028        msgbus.dispose();
1029
1030        assert!(!msgbus.is_streaming_type(BusPayloadType::QuoteTick));
1031    }
1032
1033    #[rstest]
1034    fn test_dispose_resets_counters() {
1035        let mut msgbus = MessageBus::default();
1036
1037        msgbus.increment_sent_count();
1038        msgbus.increment_req_count();
1039        msgbus.increment_res_count();
1040        msgbus.increment_pub_count();
1041        msgbus.dispose();
1042
1043        assert_eq!(msgbus.sent_count(), 0);
1044        assert_eq!(msgbus.req_count(), 0);
1045        assert_eq!(msgbus.res_count(), 0);
1046        assert_eq!(msgbus.pub_count(), 0);
1047    }
1048
1049    #[rstest]
1050    fn test_endpoints_when_no_endpoints() {
1051        let msgbus = get_message_bus();
1052        assert!(msgbus.borrow().endpoints().is_empty());
1053    }
1054
1055    #[rstest]
1056    fn test_topics_when_no_subscriptions() {
1057        let msgbus = get_message_bus();
1058        assert!(msgbus.borrow().patterns().is_empty());
1059        assert!(!msgbus.borrow().has_subscribers("my-topic").unwrap());
1060    }
1061
1062    #[rstest]
1063    fn test_is_subscribed_when_no_subscriptions() {
1064        let msgbus = get_message_bus();
1065        let handler = get_stub_shareable_handler(None);
1066
1067        assert!(!msgbus.borrow().is_subscribed("my-topic", handler));
1068    }
1069
1070    #[rstest]
1071    fn test_get_response_handler_when_no_handler() {
1072        let msgbus = get_message_bus();
1073        let msgbus_ref = msgbus.borrow();
1074        let handler = msgbus_ref.get_response_handler(&UUID4::new());
1075        assert!(handler.is_none());
1076    }
1077
1078    #[rstest]
1079    fn test_get_response_handler_when_already_registered() {
1080        let msgbus = get_message_bus();
1081        let mut msgbus_ref = msgbus.borrow_mut();
1082        let handler = get_stub_shareable_handler(None);
1083
1084        let request_id = UUID4::new();
1085        msgbus_ref
1086            .register_response_handler(&request_id, handler.clone())
1087            .unwrap();
1088
1089        let result = msgbus_ref.register_response_handler(&request_id, handler);
1090        assert!(result.is_err());
1091    }
1092
1093    #[rstest]
1094    fn test_get_response_handler_when_registered() {
1095        let msgbus = get_message_bus();
1096        let mut msgbus_ref = msgbus.borrow_mut();
1097        let handler = get_stub_shareable_handler(None);
1098
1099        let request_id = UUID4::new();
1100        msgbus_ref
1101            .register_response_handler(&request_id, handler)
1102            .unwrap();
1103
1104        let handler = msgbus_ref.get_response_handler(&request_id).unwrap();
1105        assert_eq!(handler.id(), handler.id());
1106    }
1107
1108    #[rstest]
1109    fn test_take_response_handler_removes_registration_and_allows_reregistration() {
1110        let mut msgbus = MessageBus::default();
1111        let request_id = UUID4::new();
1112        let handler = get_stub_shareable_handler(None);
1113        let handler_id = handler.id();
1114        msgbus
1115            .register_response_handler(&request_id, handler)
1116            .unwrap();
1117
1118        let taken = msgbus.take_response_handler(&request_id).unwrap();
1119
1120        assert_eq!(taken.id(), handler_id);
1121        assert!(msgbus.get_response_handler(&request_id).is_none());
1122        assert!(msgbus.register_response_handler(&request_id, taken).is_ok());
1123    }
1124
1125    #[rstest]
1126    fn test_is_registered_when_no_registrations() {
1127        let msgbus = get_message_bus();
1128        assert!(!msgbus.borrow().is_registered("MyEndpoint"));
1129    }
1130
1131    #[rstest]
1132    fn test_register_endpoint() {
1133        let msgbus = get_message_bus();
1134        let endpoint = "MyEndpoint".into();
1135        let handler = get_stub_shareable_handler(None);
1136
1137        msgbus::register_any(endpoint, handler);
1138
1139        assert_eq!(msgbus.borrow().endpoints(), vec![endpoint.to_string()]);
1140        assert!(msgbus.borrow().get_endpoint(endpoint).is_some());
1141    }
1142
1143    #[rstest]
1144    fn test_endpoint_send() {
1145        let msgbus = get_message_bus();
1146        let endpoint = "MyEndpoint".into();
1147        let (handler, checker) = get_call_check_handler(None);
1148        let sent_count = msgbus.borrow().sent_count();
1149
1150        msgbus::register_any(endpoint, handler);
1151        assert!(msgbus.borrow().get_endpoint(endpoint).is_some());
1152        assert!(!checker.was_called());
1153
1154        msgbus::send_any(endpoint, &"Test Message");
1155
1156        assert!(checker.was_called());
1157        assert_eq!(msgbus.borrow().sent_count(), sent_count + 1);
1158    }
1159
1160    #[rstest]
1161    fn test_endpoint_send_value_increments_sent_count() {
1162        let msgbus = get_message_bus();
1163        let endpoint = "MyValueEndpoint".into();
1164        let (handler, checker) = get_call_check_handler(None);
1165        let sent_count = msgbus.borrow().sent_count();
1166
1167        msgbus::register_any(endpoint, handler);
1168        msgbus::send_any_value(endpoint, &"Test Message");
1169
1170        assert!(checker.was_called());
1171        assert_eq!(msgbus.borrow().sent_count(), sent_count + 1);
1172    }
1173
1174    #[rstest]
1175    fn test_publish_any_increments_publish_count() {
1176        let msgbus = get_message_bus();
1177        let topic = "my-published-topic";
1178        let (handler, checker) = get_call_check_handler(None);
1179        let pub_count = msgbus.borrow().pub_count();
1180
1181        msgbus::subscribe_any(topic.into(), handler, None);
1182        msgbus::publish_any(topic.into(), &"Test Message");
1183
1184        assert!(checker.was_called());
1185        assert_eq!(msgbus.borrow().pub_count(), pub_count + 1);
1186    }
1187
1188    #[rstest]
1189    fn test_deregsiter_endpoint() {
1190        let msgbus = get_message_bus();
1191        let endpoint = "MyEndpoint".into();
1192        let handler = get_stub_shareable_handler(None);
1193
1194        msgbus::register_any(endpoint, handler);
1195        msgbus::deregister_any(endpoint);
1196
1197        assert!(msgbus.borrow().endpoints().is_empty());
1198    }
1199
1200    #[rstest]
1201    fn test_subscribe() {
1202        let msgbus = get_message_bus();
1203        let topic = "my-topic";
1204        let handler = get_stub_shareable_handler(None);
1205
1206        msgbus::subscribe_any(topic.into(), handler, Some(1));
1207
1208        assert!(msgbus.borrow().has_subscribers(topic).unwrap());
1209        assert_eq!(msgbus.borrow().patterns(), vec![topic]);
1210    }
1211
1212    #[rstest]
1213    fn test_unsubscribe() {
1214        let msgbus = get_message_bus();
1215        let topic = "my-topic";
1216        let handler = get_stub_shareable_handler(None);
1217
1218        msgbus::subscribe_any(topic.into(), handler.clone(), None);
1219        msgbus::unsubscribe_any(topic.into(), &handler);
1220
1221        assert!(!msgbus.borrow().has_subscribers(topic).unwrap());
1222        assert!(msgbus.borrow().patterns().is_empty());
1223    }
1224
1225    #[rstest]
1226    fn test_subscriptions_count_rejects_invalid_topic() {
1227        let msgbus = get_message_bus();
1228
1229        let err = msgbus
1230            .borrow()
1231            .subscriptions_count("data.*")
1232            .expect_err("wildcards are invalid in topics");
1233
1234        assert_eq!(
1235            err.to_string(),
1236            "Topic `value` contained invalid characters, was data.*"
1237        );
1238    }
1239
1240    #[rstest]
1241    fn test_has_subscribers_rejects_invalid_topic() {
1242        let msgbus = get_message_bus();
1243
1244        let err = msgbus
1245            .borrow()
1246            .has_subscribers("data.*")
1247            .expect_err("wildcards are invalid in topics");
1248
1249        assert_eq!(
1250            err.to_string(),
1251            "Topic `value` contained invalid characters, was data.*"
1252        );
1253    }
1254
1255    #[rstest]
1256    fn test_subscriptions_count_any_rejects_invalid_topic() {
1257        let err = subscriptions_count_any("data.*").expect_err("wildcards are invalid in topics");
1258
1259        assert_eq!(
1260            err.to_string(),
1261            "Topic `value` contained invalid characters, was data.*"
1262        );
1263    }
1264
1265    #[rstest]
1266    fn test_matching_subscriptions() {
1267        let msgbus = get_message_bus();
1268        let pattern = "my-pattern";
1269
1270        let handler_id1 = Ustr::from("1");
1271        let handler1 = get_stub_shareable_handler(Some(handler_id1));
1272
1273        let handler_id2 = Ustr::from("2");
1274        let handler2 = get_stub_shareable_handler(Some(handler_id2));
1275
1276        let handler_id3 = Ustr::from("3");
1277        let handler3 = get_stub_shareable_handler(Some(handler_id3));
1278
1279        let handler_id4 = Ustr::from("4");
1280        let handler4 = get_stub_shareable_handler(Some(handler_id4));
1281
1282        msgbus::subscribe_any(pattern.into(), handler1, None);
1283        msgbus::subscribe_any(pattern.into(), handler2, None);
1284        msgbus::subscribe_any(pattern.into(), handler3, Some(1));
1285        msgbus::subscribe_any(pattern.into(), handler4, Some(2));
1286
1287        assert_eq!(
1288            msgbus.borrow().patterns(),
1289            vec![pattern, pattern, pattern, pattern]
1290        );
1291        assert_eq!(subscriptions_count_any(pattern).unwrap(), 4);
1292
1293        let topic = pattern;
1294        let subs = msgbus.borrow_mut().matching_subscriptions(topic);
1295        assert_eq!(subs.len(), 4);
1296        assert_eq!(subs[0].handler_id, handler_id4);
1297        assert_eq!(subs[1].handler_id, handler_id3);
1298        assert_eq!(subs[2].handler_id, handler_id1);
1299        assert_eq!(subs[3].handler_id, handler_id2);
1300    }
1301
1302    #[rstest]
1303    fn test_matching_subscriptions_orders_by_full_delivery_key_on_cache_miss() {
1304        MessageBus::default().register_message_bus();
1305        let order = Rc::new(RefCell::new(Vec::new()));
1306
1307        msgbus::subscribe_any(
1308            "delivery.*".into(),
1309            recording_any_handler("handler-z", "low-z", order.clone()),
1310            Some(1),
1311        );
1312        msgbus::subscribe_any(
1313            "delivery.topic".into(),
1314            recording_any_handler("handler-b", "exact-b", order.clone()),
1315            Some(10),
1316        );
1317        msgbus::subscribe_any(
1318            "delivery.topic".into(),
1319            recording_any_handler("handler-a", "exact-a", order.clone()),
1320            Some(10),
1321        );
1322        msgbus::subscribe_any(
1323            "delivery.*".into(),
1324            recording_any_handler("handler-b", "wildcard-b", order),
1325            Some(10),
1326        );
1327
1328        let subscriptions = get_message_bus()
1329            .borrow_mut()
1330            .matching_subscriptions("delivery.topic");
1331        let actual = subscriptions
1332            .iter()
1333            .map(|sub| (sub.priority, sub.pattern.as_str(), sub.handler_id.as_str()))
1334            .collect::<Vec<_>>();
1335
1336        assert_eq!(
1337            actual,
1338            vec![
1339                (10, "delivery.*", "handler-b"),
1340                (10, "delivery.topic", "handler-a"),
1341                (10, "delivery.topic", "handler-b"),
1342                (1, "delivery.*", "handler-z"),
1343            ]
1344        );
1345    }
1346
1347    #[rstest]
1348    fn test_first_uncached_publish_any_orders_by_full_delivery_key() {
1349        MessageBus::default().register_message_bus();
1350        let order = Rc::new(RefCell::new(Vec::new()));
1351
1352        msgbus::subscribe_any(
1353            "delivery.*".into(),
1354            recording_any_handler("handler-z", "low-z", order.clone()),
1355            Some(1),
1356        );
1357        msgbus::subscribe_any(
1358            "delivery.topic".into(),
1359            recording_any_handler("handler-b", "exact-b", order.clone()),
1360            Some(10),
1361        );
1362        msgbus::subscribe_any(
1363            "delivery.topic".into(),
1364            recording_any_handler("handler-a", "exact-a", order.clone()),
1365            Some(10),
1366        );
1367        msgbus::subscribe_any(
1368            "delivery.*".into(),
1369            recording_any_handler("handler-b", "wildcard-b", order.clone()),
1370            Some(10),
1371        );
1372
1373        msgbus::publish_any("delivery.topic".into(), &());
1374
1375        assert_eq!(
1376            *order.borrow(),
1377            vec!["wildcard-b", "exact-a", "exact-b", "low-z"]
1378        );
1379    }
1380
1381    #[rstest]
1382    fn test_late_subscription_orders_cached_any_topic_by_full_delivery_key() {
1383        MessageBus::default().register_message_bus();
1384        let order = Rc::new(RefCell::new(Vec::new()));
1385
1386        msgbus::subscribe_any(
1387            "delivery.*".into(),
1388            recording_any_handler("handler-z", "low-z", order.clone()),
1389            Some(1),
1390        );
1391        msgbus::subscribe_any(
1392            "delivery.topic".into(),
1393            recording_any_handler("handler-b", "exact-b", order.clone()),
1394            Some(10),
1395        );
1396        msgbus::subscribe_any(
1397            "delivery.topic".into(),
1398            recording_any_handler("handler-a", "exact-a", order.clone()),
1399            Some(10),
1400        );
1401        msgbus::publish_any("delivery.topic".into(), &());
1402        order.borrow_mut().clear();
1403
1404        msgbus::subscribe_any(
1405            "delivery.*".into(),
1406            recording_any_handler("handler-b", "wildcard-b", order.clone()),
1407            Some(10),
1408        );
1409        msgbus::publish_any("delivery.topic".into(), &());
1410
1411        assert_eq!(
1412            *order.borrow(),
1413            vec!["wildcard-b", "exact-a", "exact-b", "low-z"]
1414        );
1415    }
1416
1417    #[rstest]
1418    fn test_subscription_pattern_matching() {
1419        let msgbus = get_message_bus();
1420        let handler1 = get_stub_shareable_handler(Some(Ustr::from("1")));
1421        let handler2 = get_stub_shareable_handler(Some(Ustr::from("2")));
1422        let handler3 = get_stub_shareable_handler(Some(Ustr::from("3")));
1423
1424        msgbus::subscribe_any("data.quotes.*".into(), handler1, None);
1425        msgbus::subscribe_any("data.trades.*".into(), handler2, None);
1426        msgbus::subscribe_any("data.*.BINANCE.*".into(), handler3, None);
1427        assert_eq!(msgbus.borrow().subscriptions().len(), 3);
1428
1429        let topic = "data.quotes.BINANCE.ETHUSDT";
1430        assert_eq!(msgbus.borrow().find_topic_matches(topic.into()).len(), 2);
1431
1432        let matches = msgbus.borrow_mut().matching_subscriptions(topic);
1433        assert_eq!(matches.len(), 2);
1434        assert_eq!(matches[0].handler_id, Ustr::from("3"));
1435        assert_eq!(matches[1].handler_id, Ustr::from("1"));
1436    }
1437
1438    #[rstest]
1439    fn test_late_wildcard_subscription_receives_cached_topic() {
1440        let msgbus = get_message_bus();
1441        let topic = "data.instrument.POLYMARKET.TEST-SYMBOL";
1442
1443        let (early_handler, early_saver) =
1444            get_any_saving_handler::<String>(Some(Ustr::from("early")));
1445        msgbus::subscribe_any("data.*.POLYMARKET.*".into(), early_handler, None);
1446
1447        msgbus::publish_any(topic.into(), &"ONE".to_string());
1448
1449        let (late_handler, late_saver) = get_any_saving_handler::<String>(Some(Ustr::from("late")));
1450        msgbus::subscribe_any("data.instrument.POLYMARKET.*".into(), late_handler, None);
1451
1452        msgbus::publish_any(topic.into(), &"TWO".to_string());
1453
1454        assert_eq!(early_saver.get_messages(), vec!["ONE", "TWO"]);
1455        assert_eq!(late_saver.get_messages(), vec!["TWO"]);
1456
1457        let topic_mstr: MStr<Topic> = topic.into();
1458        let cached = msgbus.borrow_mut().matching_subscriptions(topic_mstr);
1459        assert_eq!(cached.len(), 2);
1460    }
1461
1462    #[rstest]
1463    fn test_late_wildcard_backfills_into_multiple_cached_topics() {
1464        let msgbus = get_message_bus();
1465        let topics = ["data.A", "data.B", "data.C"];
1466
1467        let (early_handler, early_saver) =
1468            get_any_saving_handler::<String>(Some(Ustr::from("early")));
1469        msgbus::subscribe_any("data.*".into(), early_handler, None);
1470
1471        for topic in &topics {
1472            msgbus::publish_any((*topic).into(), &(*topic).to_string());
1473        }
1474
1475        let (late_handler, late_saver) = get_any_saving_handler::<String>(Some(Ustr::from("late")));
1476        msgbus::subscribe_any("data.*".into(), late_handler, None);
1477
1478        for topic in &topics {
1479            msgbus::publish_any((*topic).into(), &format!("{topic}-2"));
1480        }
1481
1482        assert_eq!(
1483            early_saver.get_messages(),
1484            vec![
1485                "data.A", "data.B", "data.C", "data.A-2", "data.B-2", "data.C-2"
1486            ],
1487        );
1488        assert_eq!(
1489            late_saver.get_messages(),
1490            vec!["data.A-2", "data.B-2", "data.C-2"]
1491        );
1492
1493        for topic in &topics {
1494            let topic_mstr: MStr<Topic> = (*topic).into();
1495            assert_eq!(
1496                msgbus.borrow_mut().matching_subscriptions(topic_mstr).len(),
1497                2,
1498                "topic {topic} should have both subscribers cached",
1499            );
1500        }
1501    }
1502
1503    /// A simple reference model for subscription behavior.
1504    struct SimpleSubscriptionModel {
1505        /// Stores (pattern, `handler_id`) tuples for active subscriptions.
1506        subscriptions: Vec<(String, String)>,
1507    }
1508
1509    impl SimpleSubscriptionModel {
1510        fn new() -> Self {
1511            Self {
1512                subscriptions: Vec::new(),
1513            }
1514        }
1515
1516        fn subscribe(&mut self, pattern: &str, handler_id: &str) {
1517            let subscription = (pattern.to_string(), handler_id.to_string());
1518            if !self.subscriptions.contains(&subscription) {
1519                self.subscriptions.push(subscription);
1520            }
1521        }
1522
1523        fn unsubscribe(&mut self, pattern: &str, handler_id: &str) -> bool {
1524            let subscription = (pattern.to_string(), handler_id.to_string());
1525            if let Some(idx) = self.subscriptions.iter().position(|s| s == &subscription) {
1526                self.subscriptions.remove(idx);
1527                true
1528            } else {
1529                false
1530            }
1531        }
1532
1533        fn is_subscribed(&self, pattern: &str, handler_id: &str) -> bool {
1534            self.subscriptions
1535                .contains(&(pattern.to_string(), handler_id.to_string()))
1536        }
1537
1538        fn matching_subscriptions(&self, topic: &str) -> Vec<(String, String)> {
1539            let topic = topic.into();
1540
1541            self.subscriptions
1542                .iter()
1543                .filter(|(pat, _)| is_matching_backtracking(topic, pat.into()))
1544                .map(|(pat, id)| (pat.clone(), id.clone()))
1545                .collect()
1546        }
1547
1548        fn subscription_count(&self) -> usize {
1549            self.subscriptions.len()
1550        }
1551    }
1552
1553    #[rstest]
1554    fn subscription_model_fuzz_testing() {
1555        let mut rng = StdRng::seed_from_u64(42);
1556
1557        let msgbus = get_message_bus();
1558        let mut model = SimpleSubscriptionModel::new();
1559
1560        // Map from handler_id to handler
1561        let mut handlers: Vec<(String, ShareableMessageHandler)> = Vec::new();
1562
1563        // Generate some patterns
1564        let patterns = generate_test_patterns(&mut rng);
1565
1566        // Generate some handler IDs
1567        let handler_ids: Vec<String> = (0..50).map(|i| format!("handler_{i}")).collect();
1568
1569        // Initialize handlers
1570        for id in &handler_ids {
1571            let handler = get_stub_shareable_handler(Some(Ustr::from(id)));
1572            handlers.push((id.clone(), handler));
1573        }
1574
1575        let num_operations = 50_000;
1576        for op_num in 0..num_operations {
1577            let operation = rng.random_range(0..4);
1578
1579            match operation {
1580                // Subscribe
1581                0 => {
1582                    let pattern_idx = rng.random_range(0..patterns.len());
1583                    let handler_idx = rng.random_range(0..handlers.len());
1584                    let pattern = &patterns[pattern_idx];
1585                    let (handler_id, handler) = &handlers[handler_idx];
1586
1587                    // Apply to reference model
1588                    model.subscribe(pattern, handler_id);
1589
1590                    // Apply to message bus
1591                    msgbus::subscribe_any(pattern.as_str().into(), handler.clone(), None);
1592
1593                    assert_eq!(
1594                        model.subscription_count(),
1595                        msgbus.borrow().subscriptions().len()
1596                    );
1597
1598                    assert!(
1599                        msgbus.borrow().is_subscribed(pattern, handler.clone()),
1600                        "Op {op_num}: is_subscribed should return true after subscribe"
1601                    );
1602                }
1603
1604                // Unsubscribe
1605                1 => {
1606                    if model.subscription_count() > 0 {
1607                        let sub_idx = rng.random_range(0..model.subscription_count());
1608                        let (pattern, handler_id) = model.subscriptions[sub_idx].clone();
1609
1610                        // Apply to reference model
1611                        model.unsubscribe(&pattern, &handler_id);
1612
1613                        // Find handler
1614                        let handler = handlers
1615                            .iter()
1616                            .find(|(id, _)| id == &handler_id)
1617                            .map(|(_, h)| h.clone())
1618                            .unwrap();
1619
1620                        // Apply to message bus
1621                        msgbus::unsubscribe_any(pattern.as_str().into(), &handler);
1622
1623                        assert_eq!(
1624                            model.subscription_count(),
1625                            msgbus.borrow().subscriptions().len()
1626                        );
1627                        assert!(
1628                            !msgbus.borrow().is_subscribed(pattern, handler.clone()),
1629                            "Op {op_num}: is_subscribed should return false after unsubscribe"
1630                        );
1631                    }
1632                }
1633
1634                // Check is_subscribed
1635                2 => {
1636                    // Get a random pattern and handler
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                    let expected = model.is_subscribed(pattern, handler_id);
1643                    let actual = msgbus.borrow().is_subscribed(pattern, handler.clone());
1644
1645                    assert_eq!(
1646                        expected, actual,
1647                        "Op {op_num}: Subscription state mismatch for pattern '{pattern}', handler '{handler_id}': expected={expected}, actual={actual}"
1648                    );
1649                }
1650
1651                // Check matching_subscriptions
1652                3 => {
1653                    // Generate a topic
1654                    let topic = create_topic(&mut rng);
1655
1656                    let actual_matches = msgbus.borrow_mut().matching_subscriptions(topic);
1657                    let expected_matches = model.matching_subscriptions(&topic);
1658
1659                    assert_eq!(
1660                        expected_matches.len(),
1661                        actual_matches.len(),
1662                        "Op {}: Match count mismatch for topic '{}': expected={}, actual={}",
1663                        op_num,
1664                        topic,
1665                        expected_matches.len(),
1666                        actual_matches.len()
1667                    );
1668
1669                    for sub in &actual_matches {
1670                        assert!(
1671                            expected_matches
1672                                .contains(&(sub.pattern.to_string(), sub.handler_id.to_string())),
1673                            "Op {}: Expected match not found: pattern='{}', handler_id='{}'",
1674                            op_num,
1675                            sub.pattern,
1676                            sub.handler_id
1677                        );
1678                    }
1679                }
1680                _ => unreachable!(),
1681            }
1682        }
1683    }
1684
1685    fn generate_pattern_from_topic(topic: &str, rng: &mut StdRng) -> String {
1686        let mut pattern = String::new();
1687
1688        for c in topic.chars() {
1689            let val: f64 = rng.random();
1690            if val < 0.1 {
1691                pattern.push('*');
1692            } else if val < 0.3 {
1693                pattern.push('?');
1694            } else if val >= 0.5 {
1695                pattern.push(c);
1696            }
1697        }
1698
1699        pattern
1700    }
1701
1702    fn generate_test_patterns(rng: &mut StdRng) -> Vec<String> {
1703        let mut patterns = vec![
1704            "data.*.*.*".to_string(),
1705            "*.*.BINANCE.*".to_string(),
1706            "events.order.*".to_string(),
1707            "data.*.*.?USDT".to_string(),
1708            "*.trades.*.BTC*".to_string(),
1709            "*.*.*.*".to_string(),
1710        ];
1711
1712        // Add some random patterns
1713        for _ in 0..50 {
1714            match rng.random_range(0..10) {
1715                // Use existing pattern
1716                0..=1 => {
1717                    let idx = rng.random_range(0..patterns.len());
1718                    patterns.push(patterns[idx].clone());
1719                }
1720                // Generate new pattern from topic
1721                _ => {
1722                    let topic = create_topic(rng);
1723                    let pattern = generate_pattern_from_topic(&topic, rng);
1724                    patterns.push(pattern);
1725                }
1726            }
1727        }
1728
1729        patterns
1730    }
1731
1732    fn create_topic(rng: &mut StdRng) -> Ustr {
1733        let cat = ["data", "info", "order"];
1734        let model = ["quotes", "trades", "orderbooks", "depths"];
1735        let venue = ["BINANCE", "BYBIT", "OKX", "FTX", "KRAKEN"];
1736        let instrument = ["BTCUSDT", "ETHUSDT", "SOLUSDT", "XRPUSDT", "DOGEUSDT"];
1737
1738        let cat = cat[rng.random_range(0..cat.len())];
1739        let model = model[rng.random_range(0..model.len())];
1740        let venue = venue[rng.random_range(0..venue.len())];
1741        let instrument = instrument[rng.random_range(0..instrument.len())];
1742        Ustr::from(&format!("{cat}.{model}.{venue}.{instrument}"))
1743    }
1744}
nautilus_common/msgbus/core.rs

nautilus_common/msgbus/
core.rs
