nautilus_common/msgbus/

core.rs

// -------------------------------------------------------------------------------------------------
//  Copyright (C) 2015-2026 Nautech Systems Pty Ltd. All rights reserved.
//  https://nautechsystems.io
//
//  Licensed under the GNU Lesser General Public License Version 3.0 (the "License");
//  You may not use this file except in compliance with the License.
//  You may obtain a copy of the License at https://www.gnu.org/licenses/lgpl-3.0.en.html
//
//  Unless required by applicable law or agreed to in writing, software
//  distributed under the License is distributed on an "AS IS" BASIS,
//  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
//  See the License for the specific language governing permissions and
//  limitations under the License.
// -------------------------------------------------------------------------------------------------

//! Core message bus implementation.
//!
//! # Design decisions
//!
//! ## Why two routing mechanisms?
//!
//! The message bus provides typed and Any-based routing to balance performance
//! and flexibility:
//!
//! **Typed routing** optimizes for throughput on known data types:
//! - `TopicRouter<T>` for pub/sub, `EndpointMap<T>` for point-to-point.
//! - Handlers implement `Handler<T>`, receive `&T` directly.
//! - No runtime type checking enables inlining and static dispatch.
//! - Built-in routers: `QuoteTick`, `TradeTick`, `Bar`, `OrderBookDeltas`,
//!   `OrderBookDepth`, `OrderEventAny`, `PositionEvent`, `AccountState`.
//!
//! **Any-based routing** provides flexibility for extensibility:
//! - `subscriptions`/`topics` maps with `ShareableMessageHandler`.
//! - Handlers implement `Handler<dyn Any>`, receive `&dyn Any`.
//! - Supports arbitrary message types without modifying the bus.
//! - Required for Python interop where types aren't known at compile time.
//!
//! ## Handler semantics
//!
//! **Typed handlers receive `&T` references:**
//! - Same message delivered to N handlers without cloning.
//! - Handler decides whether to clone (only if storing).
//! - Zero-cost for `Copy` types (`QuoteTick`, `TradeTick`, `Bar`).
//! - Efficient for large types (`OrderBookDeltas`).
//!
//! **Any-based handlers pay per-handler overhead:**
//! - Each handler receives `&dyn Any`, must downcast to `&T`.
//! - N handlers = N downcasts + N potential clones.
//! - Runtime type checking on every dispatch.
//!
//! ## Performance trade-off
//!
//! Typed routing is faster (see `benches/msgbus_typed.rs`, AMD Ryzen 9 7950X):
//!
//! | Scenario                    | Typed vs Any |
//! |-----------------------------|--------------|
//! | Handler dispatch (noop)     | ~10x faster  |
//! | Router with 5 subscribers   | ~3.5x faster |
//! | Router with 10 subscribers  | ~2x faster   |
//! | High volume (1M messages)   | ~7% faster   |
//!
//! Any-based routing pays for flexibility with runtime type checking. Use
//! typed routing for hot-path data; Any-based for custom types and Python.
//!
//! ## Routing paths are separate
//!
//! Typed and Any-based routing use separate data structures:
//! - `publish_quote` routes through `router_quotes`.
//! - `publish_any` routes through `topics`.
//!
//! Publishers and subscribers must use matching APIs. Mixing them causes
//! silent message loss.
//!
//! ## When to use each
//!
//! **Typed** (`publish_quote`, `subscribe_quotes`, etc.):
//! - Market data (quotes, trades, bars, order book updates).
//! - Order and position events.
//! - High-frequency data with known types.
//!
//! **Any-based** (`publish_any`, `subscribe_any`):
//! - Custom or user-defined data types.
//! - Low-frequency messages.
//! - Python callbacks.

use std::{
    any::{Any, TypeId},
    cell::RefCell,
    collections::HashMap,
    fmt::Debug,
    hash::{Hash, Hasher},
    rc::Rc,
};

use ahash::{AHashMap, AHashSet};
use indexmap::IndexMap;
use nautilus_core::UUID4;
use nautilus_model::{
    data::{
        Bar, Data, FundingRateUpdate, GreeksData, IndexPriceUpdate, MarkPriceUpdate,
        OrderBookDeltas, OrderBookDepth, QuoteTick, TradeTick,
        option_chain::{OptionChainSlice, OptionGreeks},
    },
    events::{AccountState, OrderEventAny, PortfolioSnapshot, PositionEvent},
    identifiers::TraderId,
    instruments::InstrumentAny,
    orderbook::OrderBook,
    orders::OrderAny,
    position::Position,
};
use smallvec::SmallVec;
use ustr::Ustr;

use super::{
    HAS_EXTERNAL_EGRESS, ShareableMessageHandler, TypedHandler,
    backing::MessageBusExternalEgress,
    config::MessageBusConfig,
    matching::is_matching_backtracking,
    message::{BusPayloadCategory, BusPayloadType},
    mstr::{Endpoint, MStr, Pattern, Topic},
    set_message_bus,
    switchboard::MessagingSwitchboard,
    typed_endpoints::{EndpointMap, IntoEndpointMap},
    typed_router::TopicRouter,
};
use crate::{
    enums::SerializationEncoding,
    messages::{
        data::{DataCommand, DataResponse, DataResponseEnvelope},
        execution::{ExecutionReport, TradingCommand},
    },
};

/// Represents a subscription to a particular topic.
///
/// This is an internal class intended to be used by the message bus to organize
/// topics and their subscribers.
#[derive(Clone, Debug)]
pub struct Subscription {
    /// The shareable message handler for the subscription.
    pub handler: ShareableMessageHandler,
    /// Store a copy of the handler ID for faster equality checks.
    pub handler_id: Ustr,
    /// The pattern for the subscription.
    pub pattern: MStr<Pattern>,
    /// The priority for the subscription determines the ordering of handlers receiving
    /// messages being processed, higher priority handlers will receive messages before
    /// lower priority handlers.
    pub priority: u32,
}

impl Subscription {
    /// Creates a new [`Subscription`] instance.
    #[must_use]
    pub fn new(
        pattern: MStr<Pattern>,
        handler: ShareableMessageHandler,
        priority: Option<u32>,
    ) -> Self {
        Self {
            handler_id: handler.0.id(),
            pattern,
            handler,
            priority: priority.unwrap_or(0),
        }
    }

    pub(crate) fn delivery_order(&self, other: &Self) -> std::cmp::Ordering {
        other
            .priority
            .cmp(&self.priority)
            .then_with(|| self.pattern.cmp(&other.pattern))
            .then_with(|| self.handler_id.cmp(&other.handler_id))
    }
}

impl PartialEq<Self> for Subscription {
    fn eq(&self, other: &Self) -> bool {
        self.pattern == other.pattern && self.handler_id == other.handler_id
    }
}

impl Eq for Subscription {}

impl PartialOrd for Subscription {
    fn partial_cmp(&self, other: &Self) -> Option<std::cmp::Ordering> {
        Some(self.cmp(other))
    }
}

impl Ord for Subscription {
    fn cmp(&self, other: &Self) -> std::cmp::Ordering {
        self.pattern
            .cmp(&other.pattern)
            .then_with(|| self.handler_id.cmp(&other.handler_id))
    }
}

impl Hash for Subscription {
    fn hash<H: Hasher>(&self, state: &mut H) {
        self.pattern.hash(state);
        self.handler_id.hash(state);
    }
}

/// A generic message bus to facilitate various messaging patterns.
///
/// The bus provides both a producer and consumer API for Pub/Sub, Req/Rep, as
/// well as direct point-to-point messaging to registered endpoints.
///
/// Pub/Sub wildcard patterns for hierarchical topics are possible:
///  - `*` asterisk represents one or more characters in a pattern.
///  - `?` question mark represents a single character in a pattern.
///
/// Given a topic and pattern potentially containing wildcard characters, i.e.
/// `*` and `?`, where `?` can match any single character in the topic, and `*`
/// can match any number of characters including zero characters.
///
/// The asterisk in a wildcard matches any character zero or more times. For
/// example, `comp*` matches anything beginning with `comp` which means `comp`,
/// `complete`, and `computer` are all matched.
///
/// A question mark matches a single character once. For example, `c?mp` matches
/// `camp` and `comp`. The question mark can also be used more than once.
/// For example, `c??p` would match both of the above examples and `coop`.
pub struct MessageBus {
    /// The trader ID associated with the message bus.
    pub trader_id: TraderId,
    /// The instance ID associated with the message bus.
    pub instance_id: UUID4,
    /// The name for the message bus.
    pub name: String,
    /// If the message bus is backed by a database.
    pub has_backing: bool,
    pub(crate) switchboard: MessagingSwitchboard,
    pub(crate) subscriptions: AHashSet<Subscription>,
    pub(crate) topics: IndexMap<MStr<Topic>, Vec<Subscription>>,
    pub(crate) endpoints: IndexMap<MStr<Endpoint>, ShareableMessageHandler>,
    pub(crate) correlation_index: AHashMap<UUID4, TypedHandler<DataResponseEnvelope>>,
    pub(crate) router_quotes: TopicRouter<QuoteTick>,
    pub(crate) router_trades: TopicRouter<TradeTick>,
    pub(crate) router_bars: TopicRouter<Bar>,
    pub(crate) router_deltas: TopicRouter<OrderBookDeltas>,
    pub(crate) router_depth: TopicRouter<OrderBookDepth>,
    pub(crate) router_book_snapshots: TopicRouter<OrderBook>,
    pub(crate) router_mark_prices: TopicRouter<MarkPriceUpdate>,
    pub(crate) router_index_prices: TopicRouter<IndexPriceUpdate>,
    pub(crate) router_funding_rates: TopicRouter<FundingRateUpdate>,
    pub(crate) router_order_events: TopicRouter<OrderEventAny>,
    pub(crate) router_position_events: TopicRouter<PositionEvent>,
    pub(crate) router_account_state: TopicRouter<AccountState>,
    pub(crate) router_orders: TopicRouter<OrderAny>,
    pub(crate) router_positions: TopicRouter<Position>,
    pub(crate) router_portfolio: TopicRouter<PortfolioSnapshot>,
    pub(crate) router_greeks: TopicRouter<GreeksData>,
    pub(crate) router_option_greeks: TopicRouter<OptionGreeks>,
    pub(crate) router_option_chain: TopicRouter<OptionChainSlice>,
    pub(crate) router_instruments: TopicRouter<InstrumentAny>,
    #[cfg(feature = "defi")]
    pub(crate) router_defi_blocks: TopicRouter<nautilus_model::defi::Block>, // nautilus-import-ok
    #[cfg(feature = "defi")]
    pub(crate) router_defi_pools: TopicRouter<nautilus_model::defi::Pool>, // nautilus-import-ok
    #[cfg(feature = "defi")]
    pub(crate) router_defi_swaps: TopicRouter<nautilus_model::defi::PoolSwap>, // nautilus-import-ok
    #[cfg(feature = "defi")]
    pub(crate) router_defi_liquidity: TopicRouter<nautilus_model::defi::PoolLiquidityUpdate>, // nautilus-import-ok
    #[cfg(feature = "defi")]
    pub(crate) router_defi_collects: TopicRouter<nautilus_model::defi::PoolFeeCollect>, // nautilus-import-ok
    #[cfg(feature = "defi")]
    pub(crate) router_defi_flash: TopicRouter<nautilus_model::defi::PoolFlash>, // nautilus-import-ok
    #[cfg(feature = "defi")]
    pub(crate) endpoints_defi_data: IntoEndpointMap<nautilus_model::defi::DefiData>, // nautilus-import-ok
    pub(crate) endpoints_quotes: EndpointMap<QuoteTick>,
    pub(crate) endpoints_trades: EndpointMap<TradeTick>,
    pub(crate) endpoints_bars: EndpointMap<Bar>,
    pub(crate) endpoints_account_state: EndpointMap<AccountState>,
    pub(crate) endpoints_trading_commands: IntoEndpointMap<TradingCommand>,
    pub(crate) endpoints_data_commands: IntoEndpointMap<DataCommand>,
    pub(crate) endpoints_data_responses: IntoEndpointMap<DataResponse>,
    pub(crate) endpoints_exec_reports: IntoEndpointMap<ExecutionReport>,
    pub(crate) endpoints_order_events: IntoEndpointMap<OrderEventAny>,
    pub(crate) endpoints_data: IntoEndpointMap<Data>,
    routers_typed: AHashMap<TypeId, Box<dyn Any>>,
    endpoints_typed: AHashMap<TypeId, Box<dyn Any>>,
    sent_count: u64,
    req_count: u64,
    res_count: u64,
    pub_count: u64,
    external_egress: Option<Rc<RefCell<Box<dyn MessageBusExternalEgress>>>>,
    has_external_streams: bool,
    encoding: SerializationEncoding,
    encoding_market_data: Option<SerializationEncoding>,
    encoding_builtin: Option<SerializationEncoding>,
    types_filter: AHashSet<BusPayloadType>,
    streaming_types: AHashSet<BusPayloadType>,
}

impl Debug for MessageBus {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct(stringify!(MessageBus))
            .field("trader_id", &self.trader_id)
            .field("instance_id", &self.instance_id)
            .field("name", &self.name)
            .field("has_backing", &self.has_backing)
            .field("external_egress", &self.external_egress.is_some())
            .finish_non_exhaustive()
    }
}

impl Default for MessageBus {
    /// Creates a new default [`MessageBus`] instance.
    fn default() -> Self {
        Self::new(TraderId::from("TRADER-001"), UUID4::new(), None, None)
    }
}

impl MessageBus {
    /// Creates a new [`MessageBus`] instance.
    #[must_use]
    pub fn new(
        trader_id: TraderId,
        instance_id: UUID4,
        name: Option<String>,
        _config: Option<HashMap<String, serde_json::Value>>,
    ) -> Self {
        Self {
            trader_id,
            instance_id,
            name: name.unwrap_or(stringify!(MessageBus).to_owned()),
            switchboard: MessagingSwitchboard::default(),
            subscriptions: AHashSet::new(),
            topics: IndexMap::new(),
            endpoints: IndexMap::new(),
            correlation_index: AHashMap::new(),
            has_backing: false,
            router_quotes: TopicRouter::new(),
            router_trades: TopicRouter::new(),
            router_bars: TopicRouter::new(),
            router_deltas: TopicRouter::new(),
            router_depth: TopicRouter::new(),
            router_book_snapshots: TopicRouter::new(),
            router_mark_prices: TopicRouter::new(),
            router_index_prices: TopicRouter::new(),
            router_funding_rates: TopicRouter::new(),
            router_order_events: TopicRouter::new(),
            router_position_events: TopicRouter::new(),
            router_account_state: TopicRouter::new(),
            router_portfolio: TopicRouter::new(),
            router_orders: TopicRouter::new(),
            router_positions: TopicRouter::new(),
            router_greeks: TopicRouter::new(),
            router_option_greeks: TopicRouter::new(),
            router_option_chain: TopicRouter::new(),
            router_instruments: TopicRouter::new(),
            #[cfg(feature = "defi")]
            router_defi_blocks: TopicRouter::new(),
            #[cfg(feature = "defi")]
            router_defi_pools: TopicRouter::new(),
            #[cfg(feature = "defi")]
            router_defi_swaps: TopicRouter::new(),
            #[cfg(feature = "defi")]
            router_defi_liquidity: TopicRouter::new(),
            #[cfg(feature = "defi")]
            router_defi_collects: TopicRouter::new(),
            #[cfg(feature = "defi")]
            router_defi_flash: TopicRouter::new(),
            #[cfg(feature = "defi")]
            endpoints_defi_data: IntoEndpointMap::new(),
            endpoints_quotes: EndpointMap::new(),
            endpoints_trades: EndpointMap::new(),
            endpoints_bars: EndpointMap::new(),
            endpoints_account_state: EndpointMap::new(),
            endpoints_trading_commands: IntoEndpointMap::new(),
            endpoints_data_commands: IntoEndpointMap::new(),
            endpoints_data_responses: IntoEndpointMap::new(),
            endpoints_exec_reports: IntoEndpointMap::new(),
            endpoints_order_events: IntoEndpointMap::new(),
            endpoints_data: IntoEndpointMap::new(),
            routers_typed: AHashMap::new(),
            endpoints_typed: AHashMap::new(),
            sent_count: 0,
            req_count: 0,
            res_count: 0,
            pub_count: 0,
            external_egress: None,
            has_external_streams: false,
            encoding: SerializationEncoding::Json,
            encoding_market_data: None,
            encoding_builtin: None,
            types_filter: AHashSet::new(),
            streaming_types: AHashSet::new(),
        }
    }

    /// Registers message bus for the current thread.
    pub fn register_message_bus(self) -> Rc<RefCell<Self>> {
        let msgbus = Rc::new(RefCell::new(self));
        set_message_bus(msgbus.clone());
        msgbus
    }

    /// Gets or creates a typed router for custom message type `T`.
    ///
    /// # Panics
    ///
    /// Panics if the stored router type doesn't match `T` (internal bug).
    pub fn router<T: 'static>(&mut self) -> &mut TopicRouter<T> {
        self.routers_typed
            .entry(TypeId::of::<T>())
            .or_insert_with(|| Box::new(TopicRouter::<T>::new()))
            .downcast_mut::<TopicRouter<T>>()
            .expect("TopicRouter type mismatch - this is a bug")
    }

    /// Gets or creates a typed endpoint map for custom message type `T`.
    ///
    /// # Panics
    ///
    /// Panics if the stored endpoint map type doesn't match `T` (internal bug).
    pub fn endpoint_map<T: 'static>(&mut self) -> &mut EndpointMap<T> {
        self.endpoints_typed
            .entry(TypeId::of::<T>())
            .or_insert_with(|| Box::new(EndpointMap::<T>::new()))
            .downcast_mut::<EndpointMap<T>>()
            .expect("EndpointMap type mismatch - this is a bug")
    }

    /// Sets external egress for serialized published messages.
    ///
    /// Typed message variants remain local because this method does not configure external streams.
    pub fn set_external_egress(
        &mut self,
        external_egress: Box<dyn MessageBusExternalEgress>,
        encoding: SerializationEncoding,
    ) {
        self.external_egress = Some(Rc::new(RefCell::new(external_egress)));
        self.has_external_streams = false;
        self.encoding = encoding;
        self.encoding_market_data = None;
        self.encoding_builtin = None;
        self.has_backing = true;
        HAS_EXTERNAL_EGRESS.with(|flag| flag.set(true));
    }

    /// Sets external egress and category encoding policy from a validated config.
    ///
    /// Typed message variants are enabled only when `config.external_streams` is non-empty.
    ///
    /// # Errors
    ///
    /// Returns an error if the config selects an unsupported encoding.
    pub fn set_external_egress_config(
        &mut self,
        external_egress: Box<dyn MessageBusExternalEgress>,
        config: &MessageBusConfig,
    ) -> crate::config::ConfigResult<()> {
        config.validate()?;

        self.external_egress = Some(Rc::new(RefCell::new(external_egress)));
        self.has_external_streams = config
            .external_streams
            .as_ref()
            .is_some_and(|streams| !streams.is_empty());
        self.encoding = config.encoding;
        self.encoding_market_data = config.encoding_market_data;
        self.encoding_builtin = config.encoding_builtin;
        self.has_backing = true;
        HAS_EXTERNAL_EGRESS.with(|flag| flag.set(true));

        Ok(())
    }

    /// Sets the payload type names excluded from external publishing.
    pub fn set_types_filter(&mut self, filter: Vec<String>) {
        self.types_filter.clear();

        for type_name in filter {
            let payload_type = BusPayloadType::from_name(&type_name);
            if !payload_type.as_str().is_empty() {
                self.types_filter.insert(payload_type);
            }

            if let Some(payload_type) = BusPayloadType::from_typed_name(&type_name) {
                self.types_filter.insert(payload_type);
            }
        }
    }

    /// Registers a payload type for external-to-internal streaming.
    pub fn add_streaming_type(&mut self, payload_type: BusPayloadType) {
        if !payload_type.as_str().is_empty() {
            self.streaming_types.insert(payload_type);
        }
    }

    /// Returns whether the payload type is registered for external-to-internal streaming.
    #[must_use]
    pub fn is_streaming_type(&self, payload_type: BusPayloadType) -> bool {
        !payload_type.as_str().is_empty() && self.streaming_types.contains(&payload_type)
    }

    /// Clears all payload types registered for external-to-internal streaming.
    pub fn clear_streaming_types(&mut self) {
        self.streaming_types.clear();
    }

    #[must_use]
    pub(crate) fn has_external_egress(&self) -> bool {
        self.external_egress.is_some()
    }

    pub(crate) fn has_external_streams(&self) -> bool {
        self.has_external_streams
    }

    pub(crate) fn external_egress(&self) -> Option<Rc<RefCell<Box<dyn MessageBusExternalEgress>>>> {
        self.external_egress.clone()
    }

    pub(crate) fn encoding_for(&self, payload_type: BusPayloadType) -> SerializationEncoding {
        match payload_type.category() {
            BusPayloadCategory::MarketData => self.encoding_market_data.unwrap_or(self.encoding),
            BusPayloadCategory::BuiltIn => self.encoding_builtin.unwrap_or(self.encoding),
            BusPayloadCategory::Other => self.encoding,
        }
    }

    pub(crate) fn types_filter(&self) -> &AHashSet<BusPayloadType> {
        &self.types_filter
    }

    /// Disposes of the message bus, clearing all subscriptions, endpoints,
    /// and handler references.
    pub fn dispose(&mut self) {
        self.subscriptions.clear();
        self.topics.clear();
        self.endpoints.clear();
        self.correlation_index.clear();

        self.router_quotes.clear();
        self.router_trades.clear();
        self.router_bars.clear();
        self.router_deltas.clear();
        self.router_depth.clear();
        self.router_book_snapshots.clear();
        self.router_mark_prices.clear();
        self.router_index_prices.clear();
        self.router_funding_rates.clear();
        self.router_order_events.clear();
        self.router_position_events.clear();
        self.router_account_state.clear();
        self.router_portfolio.clear();
        self.router_orders.clear();
        self.router_positions.clear();
        self.router_greeks.clear();
        self.router_option_greeks.clear();
        self.router_option_chain.clear();
        self.router_instruments.clear();

        #[cfg(feature = "defi")]
        {
            self.router_defi_blocks.clear();
            self.router_defi_pools.clear();
            self.router_defi_swaps.clear();
            self.router_defi_liquidity.clear();
            self.router_defi_collects.clear();
            self.router_defi_flash.clear();
            self.endpoints_defi_data.clear();
        }

        self.endpoints_quotes.clear();
        self.endpoints_trades.clear();
        self.endpoints_bars.clear();
        self.endpoints_account_state.clear();
        self.endpoints_trading_commands.clear();
        self.endpoints_data_commands.clear();
        self.endpoints_data_responses.clear();
        self.endpoints_exec_reports.clear();
        self.endpoints_order_events.clear();
        self.endpoints_data.clear();
        self.routers_typed.clear();
        self.endpoints_typed.clear();
        self.clear_streaming_types();
        self.sent_count = 0;
        self.req_count = 0;
        self.res_count = 0;
        self.pub_count = 0;

        if let Some(external_egress) = self.external_egress.take() {
            external_egress.borrow_mut().close();
        }
        self.has_external_streams = false;
        self.has_backing = false;
        HAS_EXTERNAL_EGRESS.with(|flag| flag.set(false));
    }

    /// Returns the memory address of this instance as a hexadecimal string.
    #[must_use]
    pub fn mem_address(&self) -> String {
        format!("{self:p}")
    }

    /// Returns a reference to the switchboard.
    #[must_use]
    pub fn switchboard(&self) -> &MessagingSwitchboard {
        &self.switchboard
    }

    /// Returns the total count of messages sent to endpoints.
    #[must_use]
    pub const fn sent_count(&self) -> u64 {
        self.sent_count
    }

    /// Returns the total count of requests sent to endpoints.
    #[must_use]
    pub const fn req_count(&self) -> u64 {
        self.req_count
    }

    /// Returns the total count of responses sent to registered handlers.
    #[must_use]
    pub const fn res_count(&self) -> u64 {
        self.res_count
    }

    /// Returns the total count of messages published to topics.
    #[must_use]
    pub const fn pub_count(&self) -> u64 {
        self.pub_count
    }

    pub(crate) fn increment_sent_count(&mut self) {
        self.sent_count += 1;
    }

    pub(crate) fn increment_req_count(&mut self) {
        self.req_count += 1;
    }

    pub(crate) fn increment_res_count(&mut self) {
        self.res_count += 1;
    }

    pub(crate) fn increment_pub_count(&mut self) {
        self.pub_count += 1;
    }

    /// Returns the registered endpoint addresses.
    #[must_use]
    pub fn endpoints(&self) -> Vec<&str> {
        self.endpoints.iter().map(|e| e.0.as_str()).collect()
    }

    /// Returns actively subscribed patterns.
    #[must_use]
    pub fn patterns(&self) -> Vec<&str> {
        self.subscriptions
            .iter()
            .map(|s| s.pattern.as_str())
            .collect()
    }

    /// Returns whether there are subscribers for the `topic`.
    ///
    /// # Errors
    ///
    /// Returns an error if the `topic` is not a valid topic string.
    pub fn has_subscribers<T: AsRef<str>>(&self, topic: T) -> anyhow::Result<bool> {
        Ok(self.subscriptions_count(topic)? > 0)
    }

    /// Returns the count of subscribers for the `topic`.
    ///
    /// # Errors
    ///
    /// Returns an error if the `topic` is not a valid topic string.
    pub fn subscriptions_count<T: AsRef<str>>(&self, topic: T) -> anyhow::Result<usize> {
        let topic = MStr::<Topic>::topic(topic)?;
        Ok(self
            .topics
            .get(&topic)
            .map_or_else(|| self.find_topic_matches(topic).len(), Vec::len))
    }

    /// Returns active subscriptions.
    #[must_use]
    pub fn subscriptions(&self) -> Vec<&Subscription> {
        self.subscriptions.iter().collect()
    }

    /// Returns the handler IDs for actively subscribed patterns.
    #[must_use]
    pub fn subscription_handler_ids(&self) -> Vec<&str> {
        self.subscriptions
            .iter()
            .map(|s| s.handler_id.as_str())
            .collect()
    }

    /// Returns whether the endpoint is registered.
    ///
    /// # Panics
    ///
    /// Panics if the `endpoint` conversion to `MStr<Endpoint>` fails.
    #[must_use]
    pub fn is_registered<T: Into<MStr<Endpoint>>>(&self, endpoint: T) -> bool {
        let endpoint: MStr<Endpoint> = endpoint.into();
        self.endpoints.contains_key(&endpoint)
    }

    /// Returns whether the `handler` is subscribed to the `pattern`.
    #[must_use]
    pub fn is_subscribed<T: AsRef<str>>(
        &self,
        pattern: T,
        handler: ShareableMessageHandler,
    ) -> bool {
        let pattern = MStr::<Pattern>::pattern(pattern);
        let sub = Subscription::new(pattern, handler, None);
        self.subscriptions.contains(&sub)
    }

    /// Close the message bus which will close the sender channel and join the thread.
    ///
    /// This function does not return an error.
    #[allow(
        clippy::missing_errors_doc,
        reason = "closing the in-memory message bus returns no errors"
    )]
    pub fn close(&mut self) -> anyhow::Result<()> {
        if let Some(external_egress) = self.external_egress.take() {
            external_egress.borrow_mut().close();
        }
        self.has_external_streams = false;
        self.has_backing = false;
        HAS_EXTERNAL_EGRESS.with(|flag| flag.set(false));
        Ok(())
    }

    /// Returns the handler for the `endpoint`.
    #[must_use]
    pub fn get_endpoint(&self, endpoint: MStr<Endpoint>) -> Option<&ShareableMessageHandler> {
        self.endpoints.get(&endpoint)
    }

    /// Returns the handler for the `correlation_id`.
    #[must_use]
    pub fn get_response_handler(
        &self,
        correlation_id: &UUID4,
    ) -> Option<&TypedHandler<DataResponseEnvelope>> {
        self.correlation_index.get(correlation_id)
    }

    /// Removes and returns the handler for the `correlation_id`.
    pub fn take_response_handler(
        &mut self,
        correlation_id: &UUID4,
    ) -> Option<TypedHandler<DataResponseEnvelope>> {
        self.correlation_index.remove(correlation_id)
    }

    /// Finds the subscriptions with pattern matching the `topic`.
    fn find_topic_matches(&self, topic: MStr<Topic>) -> Vec<Subscription> {
        self.subscriptions
            .iter()
            .filter_map(|sub| {
                if is_matching_backtracking(topic, sub.pattern) {
                    Some(sub.clone())
                } else {
                    None
                }
            })
            .collect()
    }

    /// Finds the subscriptions which match the `topic` and caches the
    /// results in the `patterns` map.
    #[must_use]
    pub fn matching_subscriptions<T: Into<MStr<Topic>>>(&mut self, topic: T) -> Vec<Subscription> {
        self.inner_matching_subscriptions(topic.into())
    }

    fn inner_matching_subscriptions(&mut self, topic: MStr<Topic>) -> Vec<Subscription> {
        self.topics.get(&topic).cloned().unwrap_or_else(|| {
            let mut matches = self.find_topic_matches(topic);
            matches.sort_by(Subscription::delivery_order);
            self.topics.insert(topic, matches.clone());
            matches
        })
    }

    /// Fills a buffer with handlers matching a topic.
    pub(crate) fn fill_matching_any_handlers(
        &mut self,
        topic: MStr<Topic>,
        buf: &mut SmallVec<[ShareableMessageHandler; 64]>,
    ) {
        if let Some(subs) = self.topics.get(&topic) {
            for sub in subs {
                buf.push(sub.handler.clone());
            }
        } else {
            let mut matches = self.find_topic_matches(topic);
            matches.sort_by(Subscription::delivery_order);

            for sub in &matches {
                buf.push(sub.handler.clone());
            }

            self.topics.insert(topic, matches);
        }
    }

    /// Registers a response handler for a specific correlation ID.
    ///
    /// # Errors
    ///
    /// Returns an error if `handler` is already registered for the `correlation_id`.
    pub fn register_response_handler(
        &mut self,
        correlation_id: &UUID4,
        handler: TypedHandler<DataResponseEnvelope>,
    ) -> anyhow::Result<()> {
        if self.correlation_index.contains_key(correlation_id) {
            anyhow::bail!("Correlation ID <{correlation_id}> already has a registered handler");
        }

        self.correlation_index.insert(*correlation_id, handler);

        Ok(())
    }

    pub(crate) fn unsubscribe_any(
        &mut self,
        pattern: MStr<Pattern>,
        handler: &ShareableMessageHandler,
    ) {
        log::debug!("Unsubscribing {handler:?} from pattern '{pattern}'");

        let handler_id = handler.0.id();

        let count_before = self.subscriptions.len();

        self.topics.values_mut().for_each(|subs| {
            subs.retain(|s| !(s.pattern == pattern && s.handler_id == handler_id));
        });

        self.subscriptions
            .retain(|s| !(s.pattern == pattern && s.handler_id == handler_id));

        let removed = self.subscriptions.len() < count_before;

        if removed {
            log::debug!("Handler for pattern '{pattern}' was removed");
        } else {
            log::debug!("No matching handler for pattern '{pattern}' was found");
        }
    }
}

#[cfg(test)]
mod tests {
    use std::{
        any::Any,
        cell::RefCell,
        collections::hash_map::DefaultHasher,
        fmt::Debug,
        hash::{Hash, Hasher},
        rc::Rc,
    };

    use rand::{RngExt, SeedableRng, rngs::StdRng};
    use rstest::rstest;
    use ustr::Ustr;

    use super::*;
    use crate::msgbus::{
        self, Handler, ShareableMessageHandler, get_message_bus,
        matching::is_matching_backtracking,
        stubs::{get_any_saving_handler, get_call_check_handler, get_stub_shareable_handler},
        subscriptions_count_any,
        typed_handler::shareable_handler,
    };

    #[derive(Debug)]
    struct RecordingAnyHandler {
        id: Ustr,
        label: &'static str,
        order: Rc<RefCell<Vec<&'static str>>>,
    }

    impl Handler<dyn Any> for RecordingAnyHandler {
        fn id(&self) -> Ustr {
            self.id
        }

        fn handle(&self, _message: &dyn Any) {
            self.order.borrow_mut().push(self.label);
        }
    }

    fn recording_any_handler(
        id: &'static str,
        label: &'static str,
        order: Rc<RefCell<Vec<&'static str>>>,
    ) -> ShareableMessageHandler {
        shareable_handler(Rc::new(RecordingAnyHandler {
            id: Ustr::from(id),
            label,
            order,
        }))
    }

    #[rstest]
    fn test_subscription_ordering_laws() {
        let handler = get_stub_shareable_handler(Some(Ustr::from("handler-b")));
        let base = Subscription::new("pattern-b".into(), handler.clone(), Some(1));
        let different_priority = Subscription::new("pattern-b".into(), handler, Some(2));
        let different_pattern = Subscription::new(
            "pattern-a".into(),
            get_stub_shareable_handler(Some(Ustr::from("handler-b"))),
            Some(1),
        );
        let different_handler = Subscription::new(
            "pattern-b".into(),
            get_stub_shareable_handler(Some(Ustr::from("handler-a"))),
            Some(1),
        );

        assert_eq!(base, different_priority);
        assert_eq!(base.cmp(&different_priority), std::cmp::Ordering::Equal);

        let mut base_hasher = DefaultHasher::new();
        base.hash(&mut base_hasher);
        let mut different_priority_hasher = DefaultHasher::new();
        different_priority.hash(&mut different_priority_hasher);
        assert_eq!(base_hasher.finish(), different_priority_hasher.finish());

        let variants = [
            base,
            different_priority,
            different_pattern,
            different_handler,
        ];

        for a in &variants {
            for b in &variants {
                assert_eq!(a == b, a.cmp(b).is_eq());
                assert_eq!(a.partial_cmp(b), Some(a.cmp(b)));
                assert_eq!(a.cmp(b), b.cmp(a).reverse());
            }
        }
    }

    #[rstest]
    fn test_new() {
        let trader_id = TraderId::default();
        let msgbus = MessageBus::new(trader_id, UUID4::new(), None, None);

        assert_eq!(msgbus.trader_id, trader_id);
        assert_eq!(msgbus.name, stringify!(MessageBus));
    }

    #[rstest]
    fn encoding_for_uses_market_data_override() {
        let msgbus = MessageBus {
            encoding: SerializationEncoding::Json,
            encoding_market_data: Some(SerializationEncoding::MsgPack),
            ..Default::default()
        };

        assert_eq!(
            msgbus.encoding_for(BusPayloadType::QuoteTick),
            SerializationEncoding::MsgPack
        );
        assert_eq!(
            msgbus.encoding_for(BusPayloadType::Custom(Ustr::from("CustomPayload"))),
            SerializationEncoding::Json
        );
    }

    #[rstest]
    fn encoding_for_uses_builtin_override() {
        let msgbus = MessageBus {
            encoding: SerializationEncoding::Json,
            encoding_builtin: Some(SerializationEncoding::MsgPack),
            ..Default::default()
        };

        assert_eq!(
            msgbus.encoding_for(BusPayloadType::OrderEvent),
            SerializationEncoding::MsgPack
        );
        assert_eq!(
            msgbus.encoding_for(BusPayloadType::Instrument),
            SerializationEncoding::Json
        );
    }

    #[rstest]
    fn encoding_for_uses_default_without_category_override() {
        let msgbus = MessageBus {
            encoding: SerializationEncoding::MsgPack,
            ..Default::default()
        };

        assert_eq!(
            msgbus.encoding_for(BusPayloadType::QuoteTick),
            SerializationEncoding::MsgPack
        );
        assert_eq!(
            msgbus.encoding_for(BusPayloadType::OrderEvent),
            SerializationEncoding::MsgPack
        );
        assert_eq!(
            msgbus.encoding_for(BusPayloadType::Custom(Ustr::from("CustomPayload"))),
            SerializationEncoding::MsgPack
        );
    }

    #[rstest]
    fn set_types_filter_resolves_untyped_and_typed_names() {
        let mut msgbus = MessageBus::default();

        msgbus.set_types_filter(vec![
            "QuoteTick".to_string(),
            "ExternalCustomPayload".to_string(),
            "OrderStatusReport".to_string(),
            String::new(),
        ]);

        let filter = msgbus.types_filter();
        assert_eq!(filter.len(), 4);
        assert!(filter.contains(&BusPayloadType::QuoteTick));
        assert!(filter.contains(&BusPayloadType::Custom(Ustr::from("ExternalCustomPayload"))));
        assert!(filter.contains(&BusPayloadType::Custom(Ustr::from("OrderStatusReport"))));
        assert!(filter.contains(&BusPayloadType::OrderStatusReport));
        assert!(!filter.contains(&BusPayloadType::Custom(Ustr::default())));
    }

    #[rstest]
    fn streaming_type_registration_uses_canonical_payload_names() {
        let mut msgbus = MessageBus::default();

        msgbus.add_streaming_type(BusPayloadType::QuoteTick);
        msgbus.add_streaming_type(BusPayloadType::Custom(Ustr::from("CustomPayload")));

        assert!(msgbus.is_streaming_type(BusPayloadType::QuoteTick));
        assert!(msgbus.is_streaming_type(BusPayloadType::Custom(Ustr::from("CustomPayload"))));
        assert!(msgbus.streaming_types.contains(&BusPayloadType::QuoteTick));
        assert!(
            msgbus
                .streaming_types
                .contains(&BusPayloadType::Custom(Ustr::from("CustomPayload")))
        );
        assert!(!msgbus.is_streaming_type(BusPayloadType::TradeTick));
    }

    #[rstest]
    fn streaming_type_registration_ignores_empty_custom_payload_type() {
        let mut msgbus = MessageBus::default();

        msgbus.add_streaming_type(BusPayloadType::Custom(Ustr::default()));

        assert!(!msgbus.is_streaming_type(BusPayloadType::Custom(Ustr::default())));
        assert!(msgbus.streaming_types.is_empty());
    }

    #[rstest]
    fn clear_streaming_types_removes_registered_types() {
        let mut msgbus = MessageBus::default();
        msgbus.add_streaming_type(BusPayloadType::QuoteTick);

        msgbus.clear_streaming_types();

        assert!(!msgbus.is_streaming_type(BusPayloadType::QuoteTick));
    }

    #[rstest]
    fn dispose_clears_streaming_types() {
        let mut msgbus = MessageBus::default();
        msgbus.add_streaming_type(BusPayloadType::QuoteTick);

        msgbus.dispose();

        assert!(!msgbus.is_streaming_type(BusPayloadType::QuoteTick));
    }

    #[rstest]
    fn test_dispose_resets_counters() {
        let mut msgbus = MessageBus::default();

        msgbus.increment_sent_count();
        msgbus.increment_req_count();
        msgbus.increment_res_count();
        msgbus.increment_pub_count();
        msgbus.dispose();

        assert_eq!(msgbus.sent_count(), 0);
        assert_eq!(msgbus.req_count(), 0);
        assert_eq!(msgbus.res_count(), 0);
        assert_eq!(msgbus.pub_count(), 0);
    }

    #[rstest]
    fn test_endpoints_when_no_endpoints() {
        let msgbus = get_message_bus();
        assert!(msgbus.borrow().endpoints().is_empty());
    }

    #[rstest]
    fn test_topics_when_no_subscriptions() {
        let msgbus = get_message_bus();
        assert!(msgbus.borrow().patterns().is_empty());
        assert!(!msgbus.borrow().has_subscribers("my-topic").unwrap());
    }

    #[rstest]
    fn test_is_subscribed_when_no_subscriptions() {
        let msgbus = get_message_bus();
        let handler = get_stub_shareable_handler(None);

        assert!(!msgbus.borrow().is_subscribed("my-topic", handler));
    }

    #[rstest]
    fn test_get_response_handler_when_no_handler() {
        let msgbus = get_message_bus();
        let msgbus_ref = msgbus.borrow();
        let handler = msgbus_ref.get_response_handler(&UUID4::new());
        assert!(handler.is_none());
    }

    #[rstest]
    fn test_get_response_handler_when_already_registered() {
        let msgbus = get_message_bus();
        let mut msgbus_ref = msgbus.borrow_mut();
        let handler = TypedHandler::from(|_: &DataResponseEnvelope| {});

        let request_id = UUID4::new();
        msgbus_ref
            .register_response_handler(&request_id, handler.clone())
            .unwrap();

        let result = msgbus_ref.register_response_handler(&request_id, handler);
        assert!(result.is_err());
    }

    #[rstest]
    fn test_get_response_handler_when_registered() {
        let msgbus = get_message_bus();
        let mut msgbus_ref = msgbus.borrow_mut();
        let handler = TypedHandler::from(|_: &DataResponseEnvelope| {});

        let expected_id = handler.id();
        let request_id = UUID4::new();
        msgbus_ref
            .register_response_handler(&request_id, handler)
            .unwrap();

        let handler = msgbus_ref.get_response_handler(&request_id).unwrap();
        assert_eq!(handler.id(), expected_id);
    }

    #[rstest]
    fn test_take_response_handler_removes_registration_and_allows_reregistration() {
        let mut msgbus = MessageBus::default();
        let request_id = UUID4::new();
        let handler = TypedHandler::from(|_: &DataResponseEnvelope| {});
        let handler_id = handler.id();
        msgbus
            .register_response_handler(&request_id, handler)
            .unwrap();

        let taken = msgbus.take_response_handler(&request_id).unwrap();

        assert_eq!(taken.id(), handler_id);
        assert!(msgbus.get_response_handler(&request_id).is_none());
        assert!(msgbus.register_response_handler(&request_id, taken).is_ok());
    }

    #[rstest]
    fn test_is_registered_when_no_registrations() {
        let msgbus = get_message_bus();
        assert!(!msgbus.borrow().is_registered("MyEndpoint"));
    }

    #[rstest]
    fn test_register_endpoint() {
        let msgbus = get_message_bus();
        let endpoint = "MyEndpoint".into();
        let handler = get_stub_shareable_handler(None);

        msgbus::register_any(endpoint, handler);

        assert_eq!(msgbus.borrow().endpoints(), vec![endpoint.to_string()]);
        assert!(msgbus.borrow().get_endpoint(endpoint).is_some());
    }

    #[rstest]
    fn test_endpoint_send() {
        let msgbus = get_message_bus();
        let endpoint = "MyEndpoint".into();
        let (handler, checker) = get_call_check_handler(None);
        let sent_count = msgbus.borrow().sent_count();

        msgbus::register_any(endpoint, handler);
        assert!(msgbus.borrow().get_endpoint(endpoint).is_some());
        assert!(!checker.was_called());

        msgbus::send_any(endpoint, &"Test Message");

        assert!(checker.was_called());
        assert_eq!(msgbus.borrow().sent_count(), sent_count + 1);
    }

    #[rstest]
    fn test_endpoint_send_value_increments_sent_count() {
        let msgbus = get_message_bus();
        let endpoint = "MyValueEndpoint".into();
        let (handler, checker) = get_call_check_handler(None);
        let sent_count = msgbus.borrow().sent_count();

        msgbus::register_any(endpoint, handler);
        msgbus::send_any_value(endpoint, &"Test Message");

        assert!(checker.was_called());
        assert_eq!(msgbus.borrow().sent_count(), sent_count + 1);
    }

    #[rstest]
    fn test_publish_any_increments_publish_count() {
        let msgbus = get_message_bus();
        let topic = "my-published-topic";
        let (handler, checker) = get_call_check_handler(None);
        let pub_count = msgbus.borrow().pub_count();

        msgbus::subscribe_any(topic.into(), handler, None);
        msgbus::publish_any(topic.into(), &"Test Message");

        assert!(checker.was_called());
        assert_eq!(msgbus.borrow().pub_count(), pub_count + 1);
    }

    #[rstest]
    fn test_deregsiter_endpoint() {
        let msgbus = get_message_bus();
        let endpoint = "MyEndpoint".into();
        let handler = get_stub_shareable_handler(None);

        msgbus::register_any(endpoint, handler);
        msgbus::deregister_any(endpoint);

        assert!(msgbus.borrow().endpoints().is_empty());
    }

    #[rstest]
    fn test_subscribe() {
        let msgbus = get_message_bus();
        let topic = "my-topic";
        let handler = get_stub_shareable_handler(None);

        msgbus::subscribe_any(topic.into(), handler, Some(1));

        assert!(msgbus.borrow().has_subscribers(topic).unwrap());
        assert_eq!(msgbus.borrow().patterns(), vec![topic]);
    }

    #[rstest]
    fn test_unsubscribe() {
        let msgbus = get_message_bus();
        let topic = "my-topic";
        let handler = get_stub_shareable_handler(None);

        msgbus::subscribe_any(topic.into(), handler.clone(), None);
        msgbus::unsubscribe_any(topic.into(), &handler);

        assert!(!msgbus.borrow().has_subscribers(topic).unwrap());
        assert!(msgbus.borrow().patterns().is_empty());
    }

    #[rstest]
    fn test_subscriptions_count_rejects_invalid_topic() {
        let msgbus = get_message_bus();

        let err = msgbus
            .borrow()
            .subscriptions_count("data.*")
            .expect_err("wildcards are invalid in topics");

        assert_eq!(
            err.to_string(),
            "Topic `value` contained invalid characters, was data.*"
        );
    }

    #[rstest]
    fn test_has_subscribers_rejects_invalid_topic() {
        let msgbus = get_message_bus();

        let err = msgbus
            .borrow()
            .has_subscribers("data.*")
            .expect_err("wildcards are invalid in topics");

        assert_eq!(
            err.to_string(),
            "Topic `value` contained invalid characters, was data.*"
        );
    }

    #[rstest]
    fn test_subscriptions_count_any_rejects_invalid_topic() {
        let err = subscriptions_count_any("data.*").expect_err("wildcards are invalid in topics");

        assert_eq!(
            err.to_string(),
            "Topic `value` contained invalid characters, was data.*"
        );
    }

    #[rstest]
    fn test_matching_subscriptions() {
        let msgbus = get_message_bus();
        let pattern = "my-pattern";

        let handler_id1 = Ustr::from("1");
        let handler1 = get_stub_shareable_handler(Some(handler_id1));

        let handler_id2 = Ustr::from("2");
        let handler2 = get_stub_shareable_handler(Some(handler_id2));

        let handler_id3 = Ustr::from("3");
        let handler3 = get_stub_shareable_handler(Some(handler_id3));

        let handler_id4 = Ustr::from("4");
        let handler4 = get_stub_shareable_handler(Some(handler_id4));

        msgbus::subscribe_any(pattern.into(), handler1, None);
        msgbus::subscribe_any(pattern.into(), handler2, None);
        msgbus::subscribe_any(pattern.into(), handler3, Some(1));
        msgbus::subscribe_any(pattern.into(), handler4, Some(2));

        assert_eq!(
            msgbus.borrow().patterns(),
            vec![pattern, pattern, pattern, pattern]
        );
        assert_eq!(subscriptions_count_any(pattern).unwrap(), 4);

        let topic = pattern;
        let subs = msgbus.borrow_mut().matching_subscriptions(topic);
        assert_eq!(subs.len(), 4);
        assert_eq!(subs[0].handler_id, handler_id4);
        assert_eq!(subs[1].handler_id, handler_id3);
        assert_eq!(subs[2].handler_id, handler_id1);
        assert_eq!(subs[3].handler_id, handler_id2);
    }

    #[rstest]
    fn test_matching_subscriptions_orders_by_full_delivery_key_on_cache_miss() {
        MessageBus::default().register_message_bus();
        let order = Rc::new(RefCell::new(Vec::new()));

        msgbus::subscribe_any(
            "delivery.*".into(),
            recording_any_handler("handler-z", "low-z", order.clone()),
            Some(1),
        );
        msgbus::subscribe_any(
            "delivery.topic".into(),
            recording_any_handler("handler-b", "exact-b", order.clone()),
            Some(10),
        );
        msgbus::subscribe_any(
            "delivery.topic".into(),
            recording_any_handler("handler-a", "exact-a", order.clone()),
            Some(10),
        );
        msgbus::subscribe_any(
            "delivery.*".into(),
            recording_any_handler("handler-b", "wildcard-b", order),
            Some(10),
        );

        let subscriptions = get_message_bus()
            .borrow_mut()
            .matching_subscriptions("delivery.topic");
        let actual = subscriptions
            .iter()
            .map(|sub| (sub.priority, sub.pattern.as_str(), sub.handler_id.as_str()))
            .collect::<Vec<_>>();

        assert_eq!(
            actual,
            vec![
                (10, "delivery.*", "handler-b"),
                (10, "delivery.topic", "handler-a"),
                (10, "delivery.topic", "handler-b"),
                (1, "delivery.*", "handler-z"),
            ]
        );
    }

    #[rstest]
    fn test_first_uncached_publish_any_orders_by_full_delivery_key() {
        MessageBus::default().register_message_bus();
        let order = Rc::new(RefCell::new(Vec::new()));

        msgbus::subscribe_any(
            "delivery.*".into(),
            recording_any_handler("handler-z", "low-z", order.clone()),
            Some(1),
        );
        msgbus::subscribe_any(
            "delivery.topic".into(),
            recording_any_handler("handler-b", "exact-b", order.clone()),
            Some(10),
        );
        msgbus::subscribe_any(
            "delivery.topic".into(),
            recording_any_handler("handler-a", "exact-a", order.clone()),
            Some(10),
        );
        msgbus::subscribe_any(
            "delivery.*".into(),
            recording_any_handler("handler-b", "wildcard-b", order.clone()),
            Some(10),
        );

        msgbus::publish_any("delivery.topic".into(), &());

        assert_eq!(
            *order.borrow(),
            vec!["wildcard-b", "exact-a", "exact-b", "low-z"]
        );
    }

    #[rstest]
    fn test_late_subscription_orders_cached_any_topic_by_full_delivery_key() {
        MessageBus::default().register_message_bus();
        let order = Rc::new(RefCell::new(Vec::new()));

        msgbus::subscribe_any(
            "delivery.*".into(),
            recording_any_handler("handler-z", "low-z", order.clone()),
            Some(1),
        );
        msgbus::subscribe_any(
            "delivery.topic".into(),
            recording_any_handler("handler-b", "exact-b", order.clone()),
            Some(10),
        );
        msgbus::subscribe_any(
            "delivery.topic".into(),
            recording_any_handler("handler-a", "exact-a", order.clone()),
            Some(10),
        );
        msgbus::publish_any("delivery.topic".into(), &());
        order.borrow_mut().clear();

        msgbus::subscribe_any(
            "delivery.*".into(),
            recording_any_handler("handler-b", "wildcard-b", order.clone()),
            Some(10),
        );
        msgbus::publish_any("delivery.topic".into(), &());

        assert_eq!(
            *order.borrow(),
            vec!["wildcard-b", "exact-a", "exact-b", "low-z"]
        );
    }

    #[rstest]
    fn test_subscription_pattern_matching() {
        let msgbus = get_message_bus();
        let handler1 = get_stub_shareable_handler(Some(Ustr::from("1")));
        let handler2 = get_stub_shareable_handler(Some(Ustr::from("2")));
        let handler3 = get_stub_shareable_handler(Some(Ustr::from("3")));

        msgbus::subscribe_any("data.quotes.*".into(), handler1, None);
        msgbus::subscribe_any("data.trades.*".into(), handler2, None);
        msgbus::subscribe_any("data.*.BINANCE.*".into(), handler3, None);
        assert_eq!(msgbus.borrow().subscriptions().len(), 3);

        let topic = "data.quotes.BINANCE.ETHUSDT";
        assert_eq!(msgbus.borrow().find_topic_matches(topic.into()).len(), 2);

        let matches = msgbus.borrow_mut().matching_subscriptions(topic);
        assert_eq!(matches.len(), 2);
        assert_eq!(matches[0].handler_id, Ustr::from("3"));
        assert_eq!(matches[1].handler_id, Ustr::from("1"));
    }

    #[rstest]
    fn test_late_wildcard_subscription_receives_cached_topic() {
        let msgbus = get_message_bus();
        let topic = "data.instrument.POLYMARKET.TEST-SYMBOL";

        let (early_handler, early_saver) =
            get_any_saving_handler::<String>(Some(Ustr::from("early")));
        msgbus::subscribe_any("data.*.POLYMARKET.*".into(), early_handler, None);

        msgbus::publish_any(topic.into(), &"ONE".to_string());

        let (late_handler, late_saver) = get_any_saving_handler::<String>(Some(Ustr::from("late")));
        msgbus::subscribe_any("data.instrument.POLYMARKET.*".into(), late_handler, None);

        msgbus::publish_any(topic.into(), &"TWO".to_string());

        assert_eq!(early_saver.get_messages(), vec!["ONE", "TWO"]);
        assert_eq!(late_saver.get_messages(), vec!["TWO"]);

        let topic_mstr: MStr<Topic> = topic.into();
        let cached = msgbus.borrow_mut().matching_subscriptions(topic_mstr);
        assert_eq!(cached.len(), 2);
    }

    #[rstest]
    fn test_late_wildcard_backfills_into_multiple_cached_topics() {
        let msgbus = get_message_bus();
        let topics = ["data.A", "data.B", "data.C"];

        let (early_handler, early_saver) =
            get_any_saving_handler::<String>(Some(Ustr::from("early")));
        msgbus::subscribe_any("data.*".into(), early_handler, None);

        for topic in &topics {
            msgbus::publish_any((*topic).into(), &(*topic).to_string());
        }

        let (late_handler, late_saver) = get_any_saving_handler::<String>(Some(Ustr::from("late")));
        msgbus::subscribe_any("data.*".into(), late_handler, None);

        for topic in &topics {
            msgbus::publish_any((*topic).into(), &format!("{topic}-2"));
        }

        assert_eq!(
            early_saver.get_messages(),
            vec![
                "data.A", "data.B", "data.C", "data.A-2", "data.B-2", "data.C-2"
            ],
        );
        assert_eq!(
            late_saver.get_messages(),
            vec!["data.A-2", "data.B-2", "data.C-2"]
        );

        for topic in &topics {
            let topic_mstr: MStr<Topic> = (*topic).into();
            assert_eq!(
                msgbus.borrow_mut().matching_subscriptions(topic_mstr).len(),
                2,
                "topic {topic} should have both subscribers cached",
            );
        }
    }

    /// A simple reference model for subscription behavior.
    struct SimpleSubscriptionModel {
        /// Stores (pattern, `handler_id`) tuples for active subscriptions.
        subscriptions: Vec<(String, String)>,
    }

    impl SimpleSubscriptionModel {
        fn new() -> Self {
            Self {
                subscriptions: Vec::new(),
            }
        }

        fn subscribe(&mut self, pattern: &str, handler_id: &str) {
            let subscription = (pattern.to_string(), handler_id.to_string());
            if !self.subscriptions.contains(&subscription) {
                self.subscriptions.push(subscription);
            }
        }

        fn unsubscribe(&mut self, pattern: &str, handler_id: &str) -> bool {
            let subscription = (pattern.to_string(), handler_id.to_string());
            if let Some(idx) = self.subscriptions.iter().position(|s| s == &subscription) {
                self.subscriptions.remove(idx);
                true
            } else {
                false
            }
        }

        fn is_subscribed(&self, pattern: &str, handler_id: &str) -> bool {
            self.subscriptions
                .contains(&(pattern.to_string(), handler_id.to_string()))
        }

        fn matching_subscriptions(&self, topic: &str) -> Vec<(String, String)> {
            let topic = topic.into();

            self.subscriptions
                .iter()
                .filter(|(pat, _)| is_matching_backtracking(topic, pat.into()))
                .map(|(pat, id)| (pat.clone(), id.clone()))
                .collect()
        }

        fn subscription_count(&self) -> usize {
            self.subscriptions.len()
        }
    }

    #[rstest]
    fn subscription_model_fuzz_testing() {
        let mut rng = StdRng::seed_from_u64(42);

        let msgbus = get_message_bus();
        let mut model = SimpleSubscriptionModel::new();

        // Map from handler_id to handler
        let mut handlers: Vec<(String, ShareableMessageHandler)> = Vec::new();

        // Generate some patterns
        let patterns = generate_test_patterns(&mut rng);

        // Generate some handler IDs
        let handler_ids: Vec<String> = (0..50).map(|i| format!("handler_{i}")).collect();

        // Initialize handlers
        for id in &handler_ids {
            let handler = get_stub_shareable_handler(Some(Ustr::from(id)));
            handlers.push((id.clone(), handler));
        }

        let num_operations = 50_000;
        for op_num in 0..num_operations {
            let operation = rng.random_range(0..4);

            match operation {
                // Subscribe
                0 => {
                    let pattern_idx = rng.random_range(0..patterns.len());
                    let handler_idx = rng.random_range(0..handlers.len());
                    let pattern = &patterns[pattern_idx];
                    let (handler_id, handler) = &handlers[handler_idx];

                    // Apply to reference model
                    model.subscribe(pattern, handler_id);

                    // Apply to message bus
                    msgbus::subscribe_any(pattern.as_str().into(), handler.clone(), None);

                    assert_eq!(
                        model.subscription_count(),
                        msgbus.borrow().subscriptions().len()
                    );

                    assert!(
                        msgbus.borrow().is_subscribed(pattern, handler.clone()),
                        "Op {op_num}: is_subscribed should return true after subscribe"
                    );
                }

                // Unsubscribe
                1 => {
                    if model.subscription_count() > 0 {
                        let sub_idx = rng.random_range(0..model.subscription_count());
                        let (pattern, handler_id) = model.subscriptions[sub_idx].clone();

                        // Apply to reference model
                        model.unsubscribe(&pattern, &handler_id);

                        // Find handler
                        let handler = handlers
                            .iter()
                            .find(|(id, _)| id == &handler_id)
                            .map(|(_, h)| h.clone())
                            .unwrap();

                        // Apply to message bus
                        msgbus::unsubscribe_any(pattern.as_str().into(), &handler);

                        assert_eq!(
                            model.subscription_count(),
                            msgbus.borrow().subscriptions().len()
                        );
                        assert!(
                            !msgbus.borrow().is_subscribed(pattern, handler.clone()),
                            "Op {op_num}: is_subscribed should return false after unsubscribe"
                        );
                    }
                }

                // Check is_subscribed
                2 => {
                    // Get a random pattern and handler
                    let pattern_idx = rng.random_range(0..patterns.len());
                    let handler_idx = rng.random_range(0..handlers.len());
                    let pattern = &patterns[pattern_idx];
                    let (handler_id, handler) = &handlers[handler_idx];

                    let expected = model.is_subscribed(pattern, handler_id);
                    let actual = msgbus.borrow().is_subscribed(pattern, handler.clone());

                    assert_eq!(
                        expected, actual,
                        "Op {op_num}: Subscription state mismatch for pattern '{pattern}', handler '{handler_id}': expected={expected}, actual={actual}"
                    );
                }

                // Check matching_subscriptions
                3 => {
                    // Generate a topic
                    let topic = create_topic(&mut rng);

                    let actual_matches = msgbus.borrow_mut().matching_subscriptions(topic);
                    let expected_matches = model.matching_subscriptions(&topic);

                    assert_eq!(
                        expected_matches.len(),
                        actual_matches.len(),
                        "Op {}: Match count mismatch for topic '{}': expected={}, actual={}",
                        op_num,
                        topic,
                        expected_matches.len(),
                        actual_matches.len()
                    );

                    for sub in &actual_matches {
                        assert!(
                            expected_matches
                                .contains(&(sub.pattern.to_string(), sub.handler_id.to_string())),
                            "Op {}: Expected match not found: pattern='{}', handler_id='{}'",
                            op_num,
                            sub.pattern,
                            sub.handler_id
                        );
                    }
                }
                _ => unreachable!(),
            }
        }
    }

    fn generate_pattern_from_topic(topic: &str, rng: &mut StdRng) -> String {
        let mut pattern = String::new();

        for c in topic.chars() {
            let val: f64 = rng.random();
            if val < 0.1 {
                pattern.push('*');
            } else if val < 0.3 {
                pattern.push('?');
            } else if val >= 0.5 {
                pattern.push(c);
            }
        }

        pattern
    }

    fn generate_test_patterns(rng: &mut StdRng) -> Vec<String> {
        let mut patterns = vec![
            "data.*.*.*".to_string(),
            "*.*.BINANCE.*".to_string(),
            "events.order.*".to_string(),
            "data.*.*.?USDT".to_string(),
            "*.trades.*.BTC*".to_string(),
            "*.*.*.*".to_string(),
        ];

        // Add some random patterns
        for _ in 0..50 {
            match rng.random_range(0..10) {
                // Use existing pattern
                0..=1 => {
                    let idx = rng.random_range(0..patterns.len());
                    patterns.push(patterns[idx].clone());
                }
                // Generate new pattern from topic
                _ => {
                    let topic = create_topic(rng);
                    let pattern = generate_pattern_from_topic(&topic, rng);
                    patterns.push(pattern);
                }
            }
        }

        patterns
    }

    fn create_topic(rng: &mut StdRng) -> Ustr {
        let cat = ["data", "info", "order"];
        let model = ["quotes", "trades", "orderbooks", "depths"];
        let venue = ["BINANCE", "BYBIT", "OKX", "FTX", "KRAKEN"];
        let instrument = ["BTCUSDT", "ETHUSDT", "SOLUSDT", "XRPUSDT", "DOGEUSDT"];

        let cat = cat[rng.random_range(0..cat.len())];
        let model = model[rng.random_range(0..model.len())];
        let venue = venue[rng.random_range(0..venue.len())];
        let instrument = instrument[rng.random_range(0..instrument.len())];
        Ustr::from(&format!("{cat}.{model}.{venue}.{instrument}"))
    }
}