nautilus_common/msgbus/

switchboard.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.
// -------------------------------------------------------------------------------------------------

//! Built-in message bus endpoint, topic, and pattern names.
//!
//! The `DataEngine`, `ExecEngine`, and `RiskEngine` command endpoints use a queued entry point
//! plus a direct dispatch endpoint:
//!
//! - `*.queue_execute` is the normal entry point for runtime command producers. It routes through
//!   the current runner queue or channel before the engine sees the command.
//! - `*.execute` is the lower-level dispatch point used by runner drains and internal paths that
//!   intentionally bypass the queue.
//!
//! Prefer queued endpoints for runtime command producers unless the caller owns the ordering and
//! re-entrancy implications of direct dispatch. The risk and execution queued endpoints can fall
//! back to direct dispatch when no trading command sender is installed.

use std::{fmt::Write, num::NonZeroUsize, sync::OnceLock};

use ahash::AHashMap;
use nautilus_model::{
    data::{BarType, DataType, data_type::IDENTIFIER_TOPIC_SUFFIX},
    identifiers::{
        ClientId, ClientOrderId, InstrumentId, OptionSeriesId, PositionId, StrategyId, Venue,
    },
};

use super::mstr::{Endpoint, MStr, Pattern, Topic};
use crate::{msgbus::get_message_bus, runner::SystemChannel};

pub const CLOSE_TOPIC: &str = "CLOSE";
pub const TIME_EVENT_TOPIC: &str = "clock.time_event";

static DATA_QUEUE_COMMAND_ENDPOINT: OnceLock<MStr<Endpoint>> = OnceLock::new();
static DATA_EXECUTE_ENDPOINT: OnceLock<MStr<Endpoint>> = OnceLock::new();
static DATA_PROCESS_ANY_ENDPOINT: OnceLock<MStr<Endpoint>> = OnceLock::new();
static DATA_PROCESS_DATA_ENDPOINT: OnceLock<MStr<Endpoint>> = OnceLock::new();
static DATA_RESPONSE_ENDPOINT: OnceLock<MStr<Endpoint>> = OnceLock::new();
static DATA_RESPONSE_TOPIC: OnceLock<MStr<Topic>> = OnceLock::new();
static TIME_EVENT_TOPIC_MSTR: OnceLock<MStr<Topic>> = OnceLock::new();
static EXEC_QUEUE_COMMAND_ENDPOINT: OnceLock<MStr<Endpoint>> = OnceLock::new();
static EXEC_EXECUTE_ENDPOINT: OnceLock<MStr<Endpoint>> = OnceLock::new();
static EXEC_PROCESS_ENDPOINT: OnceLock<MStr<Endpoint>> = OnceLock::new();
static EXEC_RECONCILE_REPORT_ENDPOINT: OnceLock<MStr<Endpoint>> = OnceLock::new();
static RISK_EXECUTE_ENDPOINT: OnceLock<MStr<Endpoint>> = OnceLock::new();
static RISK_QUEUE_EXECUTE_ENDPOINT: OnceLock<MStr<Endpoint>> = OnceLock::new();
static RISK_PROCESS_ENDPOINT: OnceLock<MStr<Endpoint>> = OnceLock::new();
static RISK_EVENTS_TOPIC: OnceLock<MStr<Topic>> = OnceLock::new();
static ORDER_EMULATOR_ENDPOINT: OnceLock<MStr<Endpoint>> = OnceLock::new();
static PORTFOLIO_ACCOUNT_ENDPOINT: OnceLock<MStr<Endpoint>> = OnceLock::new();
static PORTFOLIO_ORDER_ENDPOINT: OnceLock<MStr<Endpoint>> = OnceLock::new();
static SYSTEM_SHUTDOWN_TOPIC: OnceLock<MStr<Topic>> = OnceLock::new();
static SUBMISSION_RECOVERY_EXHAUSTED_TOPIC: OnceLock<MStr<Topic>> = OnceLock::new();
static RECONCILIATION_RAW_ORDER_REPORT_TOPIC: OnceLock<MStr<Topic>> = OnceLock::new();
static RECONCILIATION_RAW_FILL_REPORT_TOPIC: OnceLock<MStr<Topic>> = OnceLock::new();
static RECONCILIATION_RAW_POSITION_REPORT_TOPIC: OnceLock<MStr<Topic>> = OnceLock::new();

#[cfg(feature = "defi")]
static DATA_PROCESS_DEFI_DATA_ENDPOINT: OnceLock<MStr<Endpoint>> = OnceLock::new();

macro_rules! define_switchboard {
    ($(
        $field:ident: $key_ty:ty,
        $method:ident($($arg_name:ident: $arg_ty:ty),*) -> $key_expr:expr,
        $val_fmt:expr,
        $($val_args:expr),*
    );* $(;)?) => {
        /// Represents a switchboard of built-in messaging endpoint names.
        #[derive(Clone, Debug)]
        pub struct MessagingSwitchboard {
            $(
                $field: AHashMap<$key_ty, MStr<Topic>>,
            )*
            pipeline_topics: AHashMap<MStr<Topic>, MStr<Topic>>,
            instruments_patterns: AHashMap<Venue, MStr<Pattern>>,
            book_deltas_patterns: AHashMap<InstrumentId, MStr<Pattern>>,
            book_depth_patterns: AHashMap<InstrumentId, MStr<Pattern>>,
            book_snapshots_patterns: AHashMap<(InstrumentId, NonZeroUsize), MStr<Pattern>>,
            signal_topics: AHashMap<String, MStr<Topic>>,
            signal_patterns: AHashMap<String, MStr<Pattern>>,
            #[cfg(feature = "defi")]
            pub(crate) defi: crate::defi::switchboard::DefiSwitchboard,
        }

        impl Default for MessagingSwitchboard {
            /// Creates a new default [`MessagingSwitchboard`] instance.
            fn default() -> Self {
                Self {
                    $(
                        $field: AHashMap::new(),
                    )*
                    pipeline_topics: AHashMap::new(),
                    instruments_patterns: AHashMap::new(),
                    book_deltas_patterns: AHashMap::new(),
                    book_depth_patterns: AHashMap::new(),
                    book_snapshots_patterns: AHashMap::new(),
                    signal_topics: AHashMap::new(),
                    signal_patterns: AHashMap::new(),
                    #[cfg(feature = "defi")]
                    defi: crate::defi::switchboard::DefiSwitchboard::default(),
                }
            }
        }

        impl MessagingSwitchboard {
            // Static endpoints

            /// Queued entry point for `DataEngine` commands.
            #[inline]
            #[must_use]
            pub fn data_engine_queue_execute() -> MStr<Endpoint> {
                *DATA_QUEUE_COMMAND_ENDPOINT.get_or_init(|| "DataEngine.queue_execute".into())
            }

            /// Direct dispatch endpoint for `DataEngine` commands.
            #[inline]
            #[must_use]
            pub fn data_engine_execute() -> MStr<Endpoint> {
                *DATA_EXECUTE_ENDPOINT.get_or_init(|| "DataEngine.execute".into())
            }

            #[inline]
            #[must_use]
            pub fn data_engine_process() -> MStr<Endpoint> {
                *DATA_PROCESS_ANY_ENDPOINT.get_or_init(|| "DataEngine.process".into())
            }

            #[inline]
            #[must_use]
            pub fn data_engine_process_data() -> MStr<Endpoint> {
                *DATA_PROCESS_DATA_ENDPOINT.get_or_init(|| "DataEngine.process_data".into())
            }

            #[cfg(feature = "defi")]
            #[inline]
            #[must_use]
            pub fn data_engine_process_defi_data() -> MStr<Endpoint> {
                *DATA_PROCESS_DEFI_DATA_ENDPOINT
                    .get_or_init(|| "DataEngine.process_defi_data".into())
            }

            #[inline]
            #[must_use]
            pub fn data_engine_response() -> MStr<Endpoint> {
                *DATA_RESPONSE_ENDPOINT.get_or_init(|| "DataEngine.response".into())
            }

            #[inline]
            #[must_use]
            pub fn data_response_topic() -> MStr<Topic> {
                *DATA_RESPONSE_TOPIC.get_or_init(|| "data.response".into())
            }

            /// Pub/sub topic used by the event-store tap for fired clock events.
            #[inline]
            #[must_use]
            pub fn time_event_topic() -> MStr<Topic> {
                *TIME_EVENT_TOPIC_MSTR.get_or_init(|| TIME_EVENT_TOPIC.into())
            }

            /// Direct dispatch endpoint for `ExecEngine` commands.
            #[inline]
            #[must_use]
            pub fn exec_engine_execute() -> MStr<Endpoint> {
                *EXEC_EXECUTE_ENDPOINT.get_or_init(|| "ExecEngine.execute".into())
            }

            /// Queued entry point for `ExecEngine` commands.
            #[inline]
            #[must_use]
            pub fn exec_engine_queue_execute() -> MStr<Endpoint> {
                *EXEC_QUEUE_COMMAND_ENDPOINT.get_or_init(|| "ExecEngine.queue_execute".into())
            }

            #[inline]
            #[must_use]
            pub fn exec_engine_process() -> MStr<Endpoint> {
                *EXEC_PROCESS_ENDPOINT.get_or_init(|| "ExecEngine.process".into())
            }

            #[inline]
            #[must_use]
            pub fn exec_engine_reconcile_execution_report() -> MStr<Endpoint> {
                *EXEC_RECONCILE_REPORT_ENDPOINT.get_or_init(|| "ExecEngine.reconcile_execution_report".into())
            }

            /// Direct dispatch endpoint for `RiskEngine` commands.
            #[inline]
            #[must_use]
            pub fn risk_engine_execute() -> MStr<Endpoint> {
                *RISK_EXECUTE_ENDPOINT.get_or_init(|| "RiskEngine.execute".into())
            }

            /// Queued entry point for `RiskEngine` commands.
            #[inline]
            #[must_use]
            pub fn risk_engine_queue_execute() -> MStr<Endpoint> {
                *RISK_QUEUE_EXECUTE_ENDPOINT.get_or_init(|| "RiskEngine.queue_execute".into())
            }

            #[inline]
            #[must_use]
            pub fn risk_engine_process() -> MStr<Endpoint> {
                *RISK_PROCESS_ENDPOINT.get_or_init(|| "RiskEngine.process".into())
            }

            /// Pub/sub topic carrying risk engine state events.
            #[inline]
            #[must_use]
            pub fn risk_events_topic() -> MStr<Topic> {
                *RISK_EVENTS_TOPIC.get_or_init(|| "events.risk".into())
            }

            #[inline]
            #[must_use]
            pub fn order_emulator_execute() -> MStr<Endpoint> {
                *ORDER_EMULATOR_ENDPOINT.get_or_init(|| "OrderEmulator.execute".into())
            }

            #[inline]
            #[must_use]
            pub fn portfolio_update_account() -> MStr<Endpoint> {
                *PORTFOLIO_ACCOUNT_ENDPOINT.get_or_init(|| "Portfolio.update_account".into())
            }

            #[inline]
            #[must_use]
            pub fn portfolio_update_order() -> MStr<Endpoint> {
                *PORTFOLIO_ORDER_ENDPOINT.get_or_init(|| "Portfolio.update_order".into())
            }

            /// Pub/sub topic carrying queue state changes for one runner channel.
            #[must_use]
            pub fn queue_state_changed_topic(channel: SystemChannel) -> MStr<Topic> {
                Self::queue_state_changed_pattern(Some(channel)).as_ref().into()
            }

            /// Subscription pattern for queue state changes. `None` matches every channel.
            #[must_use]
            pub fn queue_state_changed_pattern(channel: Option<SystemChannel>) -> MStr<Pattern> {
                let channel = channel.map_or_else(|| "*".to_string(), |value| format!("{value:?}"));
                format!("events.system.QueueStateChanged.{channel}").into()
            }

            /// Pub/sub topic carrying socket state changes for one client endpoint.
            #[must_use]
            pub fn socket_state_changed_topic(client_id: ClientId, endpoint: &str) -> MStr<Topic> {
                Self::socket_state_changed_pattern(Some(client_id), Some(endpoint)).as_ref().into()
            }

            /// Subscription pattern for socket state changes.
            ///
            /// Each `None` matches every value of that field. Supplied values match literally;
            /// topic components percent-encode bytes other than ASCII letters, digits, `-`, and `_`.
            #[must_use]
            pub fn socket_state_changed_pattern(
                client_id: Option<ClientId>,
                endpoint: Option<&str>,
            ) -> MStr<Pattern> {
                let client_id = client_id.map_or_else(
                    || "*".to_string(),
                    |value| state_topic_component(value.as_str()),
                );
                let endpoint = endpoint.map_or_else(|| "*".to_string(), state_topic_component);
                format!("events.system.SocketStateChanged.{client_id}.{endpoint}").into()
            }

            /// Pub/sub topic carrying `ShutdownSystem` commands published by
            /// actors, engines, and strategies.
            ///
            /// Matches the Python topic. The kernel subscribes to validate the
            /// command and signal graceful shutdown; additional components may
            /// subscribe to react to the same signal.
            #[inline]
            #[must_use]
            pub fn shutdown_system_topic() -> MStr<Topic> {
                *SYSTEM_SHUTDOWN_TOPIC.get_or_init(|| "commands.system.shutdown".into())
            }

            /// Pub/sub topic carrying native submission recovery exhaustion diagnostics.
            #[inline]
            #[must_use]
            pub fn submission_recovery_exhausted_topic() -> MStr<Topic> {
                *SUBMISSION_RECOVERY_EXHAUSTED_TOPIC
                    .get_or_init(|| "reconciliation.SubmissionRecoveryExhausted".into())
            }

            /// Pub/sub topic carrying raw `OrderStatusReport`s that arrived from
            /// a venue client, published by the execution engine at the top of
            /// reconciliation before any state mutation.
            ///
            /// The event store bus tap captures publications on this topic so
            /// forensic replay can re-run reconciliation against the same raw
            /// inputs the live engine saw. Subscribers are not expected in
            /// production; the capture surface is the sole consumer today.
            #[inline]
            #[must_use]
            pub fn reconciliation_raw_order_status_report_topic() -> MStr<Topic> {
                *RECONCILIATION_RAW_ORDER_REPORT_TOPIC
                    .get_or_init(|| "reconciliation.raw.OrderStatusReport".into())
            }

            /// Pub/sub topic carrying raw `FillReport`s that arrived from a
            /// venue client, published by the execution engine at the top of
            /// reconciliation before any state mutation.
            ///
            /// See [`Self::reconciliation_raw_order_status_report_topic`] for the
            /// capture contract.
            #[inline]
            #[must_use]
            pub fn reconciliation_raw_fill_report_topic() -> MStr<Topic> {
                *RECONCILIATION_RAW_FILL_REPORT_TOPIC
                    .get_or_init(|| "reconciliation.raw.FillReport".into())
            }

            /// Pub/sub topic carrying raw `PositionStatusReport`s that arrived
            /// from a venue client, published by the execution engine at the
            /// top of reconciliation before any state mutation.
            ///
            /// See [`Self::reconciliation_raw_order_status_report_topic`] for the
            /// capture contract.
            #[inline]
            #[must_use]
            pub fn reconciliation_raw_position_status_report_topic() -> MStr<Topic> {
                *RECONCILIATION_RAW_POSITION_REPORT_TOPIC
                    .get_or_init(|| "reconciliation.raw.PositionStatusReport".into())
            }

            /// Returns a wildcard pattern for matching all instrument topics for a venue.
            #[must_use]
            pub fn instruments_pattern(&mut self, venue: Venue) -> MStr<Pattern> {
                *self.instruments_patterns
                    .entry(venue)
                    .or_insert_with(|| format!("data.instrument.{venue}.*").into())
            }

            /// Returns the exact signal publish topic for `name`
            /// (`data.Signal<TitleName>`).
            ///
            /// The title-cased encoding mirrors the v1 Python convention so
            /// subscribers keyed on either a specific name or the global
            /// `data.Signal*` wildcard receive published signals.
            #[must_use]
            pub fn signal_topic(&mut self, name: &str) -> MStr<Topic> {
                *self
                    .signal_topics
                    .entry(name.to_string())
                    .or_insert_with(|| {
                        format!(
                            "data.Signal{}",
                            nautilus_core::string::conversions::title_case(name)
                        )
                        .into()
                    })
            }

            /// Returns the subscription pattern for `name`
            /// (`data.Signal<TitleName>*`).
            ///
            /// An empty `name` yields the wildcard `data.Signal*` that matches
            /// every signal topic.
            #[must_use]
            pub fn signal_pattern(&mut self, name: &str) -> MStr<Pattern> {
                *self
                    .signal_patterns
                    .entry(name.to_string())
                    .or_insert_with(|| {
                        format!(
                            "data.Signal{}*",
                            nautilus_core::string::conversions::title_case(name)
                        )
                        .into()
                    })
            }

            // Dynamic topics
            $(
                #[must_use]
                pub fn $method(&mut self, $($arg_name: $arg_ty),*) -> MStr<Topic> {
                    let key = $key_expr;
                    *self.$field
                        .entry(key)
                        .or_insert_with(|| format!($val_fmt, $($val_args),*).into())
                }
            )*
        }
    };
}

define_switchboard! {
    custom_topics: DataType,
    get_custom_topic(data_type: &DataType) -> data_type.clone(),
    "data.{}", data_type.topic();

    instruments_topics: Venue,
    get_instruments_topic(venue: Venue) -> venue,
    "data.instrument.{}", venue;

    instrument_topics: InstrumentId,
    get_instrument_topic(instrument_id: InstrumentId) -> instrument_id,
    "data.instrument.{}.{}", instrument_id.venue, instrument_id.symbol;

    book_deltas_topics: InstrumentId,
    get_book_deltas_topic(instrument_id: InstrumentId) -> instrument_id,
    "data.book.deltas.{}.{}", instrument_id.venue, instrument_id.symbol;

    book_depth_topics: InstrumentId,
    get_book_depth_topic(instrument_id: InstrumentId) -> instrument_id,
    "data.book.depth.{}.{}", instrument_id.venue, instrument_id.symbol;

    book_snapshots_topics: (InstrumentId, NonZeroUsize),
    get_book_snapshots_topic(instrument_id: InstrumentId, interval_ms: NonZeroUsize) -> (instrument_id, interval_ms),
    "data.book.snapshots.{}.{}.{}", instrument_id.venue, instrument_id.symbol, interval_ms;

    quote_topics: InstrumentId,
    get_quotes_topic(instrument_id: InstrumentId) -> instrument_id,
    "data.quotes.{}.{}", instrument_id.venue, instrument_id.symbol;

    trade_topics: InstrumentId,
    get_trades_topic(instrument_id: InstrumentId) -> instrument_id,
    "data.trades.{}.{}", instrument_id.venue, instrument_id.symbol;

    bar_topics: BarType,
    get_bars_topic(bar_type: BarType) -> bar_type,
    "data.bars.{}", bar_type;

    mark_price_topics: InstrumentId,
    get_mark_price_topic(instrument_id: InstrumentId) -> instrument_id,
    "data.mark_prices.{}.{}", instrument_id.venue, instrument_id.symbol;

    index_price_topics: InstrumentId,
    get_index_price_topic(instrument_id: InstrumentId) -> instrument_id,
    "data.index_prices.{}.{}", instrument_id.venue, instrument_id.symbol;

    funding_rate_topics: InstrumentId,
    get_funding_rate_topic(instrument_id: InstrumentId) -> instrument_id,
    "data.funding_rates.{}.{}", instrument_id.venue, instrument_id.symbol;

    funding_settlement_topics: InstrumentId,
    get_funding_settlement_topic(instrument_id: InstrumentId) -> instrument_id,
    "events.funding_settlements.{}.{}", instrument_id.venue, instrument_id.symbol;

    instrument_status_topics: InstrumentId,
    get_instrument_status_topic(instrument_id: InstrumentId) -> instrument_id,
    "data.status.{}.{}", instrument_id.venue, instrument_id.symbol;

    instrument_close_topics: InstrumentId,
    get_instrument_close_topic(instrument_id: InstrumentId) -> instrument_id,
    "data.close.{}.{}", instrument_id.venue, instrument_id.symbol;

    option_greeks_topics: InstrumentId,
    get_option_greeks_topic(instrument_id: InstrumentId) -> instrument_id,
    "data.option_greeks.{}.{}", instrument_id.venue, instrument_id.symbol;

    option_chain_topics: OptionSeriesId,
    get_option_chain_topic(series_id: OptionSeriesId) -> series_id,
    "data.option_chain.{}", series_id;

    order_submitted_topics: InstrumentId,
    get_order_submitted_topic(instrument_id: InstrumentId) -> instrument_id,
    "events.order_submitted.{}", instrument_id;

    order_rejected_topics: InstrumentId,
    get_order_rejected_topic(instrument_id: InstrumentId) -> instrument_id,
    "events.order_rejected.{}", instrument_id;

    order_pending_update_topics: InstrumentId,
    get_order_pending_update_topic(instrument_id: InstrumentId) -> instrument_id,
    "events.order_pending_update.{}", instrument_id;

    order_pending_cancel_topics: InstrumentId,
    get_order_pending_cancel_topic(instrument_id: InstrumentId) -> instrument_id,
    "events.order_pending_cancel.{}", instrument_id;

    order_modify_rejected_topics: InstrumentId,
    get_order_modify_rejected_topic(instrument_id: InstrumentId) -> instrument_id,
    "events.order_modify_rejected.{}", instrument_id;

    order_cancel_rejected_topics: InstrumentId,
    get_order_cancel_rejected_topic(instrument_id: InstrumentId) -> instrument_id,
    "events.order_cancel_rejected.{}", instrument_id;

    order_canceled_topics: InstrumentId,
    get_order_canceled_topic(instrument_id: InstrumentId) -> instrument_id,
    "events.order_canceled.{}", instrument_id;

    order_filled_topics: InstrumentId,
    get_order_filled_topic(instrument_id: InstrumentId) -> instrument_id,
    "events.order_filled.{}", instrument_id;

    order_fill_voided_topics: InstrumentId,
    get_order_fill_voided_topic(instrument_id: InstrumentId) -> instrument_id,
    "events.order_fill_voided.{}", instrument_id;

    order_fill_declined_topics: InstrumentId,
    get_order_fill_declined_topic(instrument_id: InstrumentId) -> instrument_id,
    "events.order_fill_declined.{}", instrument_id;

    event_order_topics: StrategyId,
    get_event_order_topic(strategy_id: StrategyId) -> strategy_id,
    "events.order.{}", strategy_id;

    event_position_topics: StrategyId,
    get_event_position_topic(strategy_id: StrategyId) -> strategy_id,
    "events.position.{}", strategy_id;

    snapshot_order_topics: ClientOrderId,
    get_snapshot_order_topic(client_order_id: ClientOrderId) -> client_order_id,
    "snapshots.order.{}", client_order_id;

    snapshot_position_topics: PositionId,
    get_snapshot_position_topic(position_id: PositionId) -> position_id,
    "snapshots.position.{}", position_id;

}

impl MessagingSwitchboard {
    #[inline]
    fn pipeline_topic(&mut self, live: MStr<Topic>) -> MStr<Topic> {
        *self.pipeline_topics.entry(live).or_insert_with(|| {
            let live = live.as_ref();
            let suffix = live
                .strip_prefix("data.")
                .expect("live data topic must start with data.");
            MStr::<Topic>::from(format!("data.pipeline.{suffix}"))
        })
    }

    #[must_use]
    pub fn get_pipeline_custom_topic(&mut self, data_type: &DataType) -> MStr<Topic> {
        let live = self.get_custom_topic(data_type);
        self.pipeline_topic(live)
    }

    #[must_use]
    pub fn get_pipeline_book_deltas_topic(&mut self, instrument_id: InstrumentId) -> MStr<Topic> {
        let live = self.get_book_deltas_topic(instrument_id);
        self.pipeline_topic(live)
    }

    #[must_use]
    pub fn get_pipeline_book_depth_topic(&mut self, instrument_id: InstrumentId) -> MStr<Topic> {
        let live = self.get_book_depth_topic(instrument_id);
        self.pipeline_topic(live)
    }

    #[must_use]
    pub fn get_pipeline_quotes_topic(&mut self, instrument_id: InstrumentId) -> MStr<Topic> {
        let live = self.get_quotes_topic(instrument_id);
        self.pipeline_topic(live)
    }

    #[must_use]
    pub fn get_pipeline_trades_topic(&mut self, instrument_id: InstrumentId) -> MStr<Topic> {
        let live = self.get_trades_topic(instrument_id);
        self.pipeline_topic(live)
    }

    #[must_use]
    pub fn get_pipeline_bars_topic(&mut self, bar_type: BarType) -> MStr<Topic> {
        let live = self.get_bars_topic(bar_type);
        self.pipeline_topic(live)
    }

    #[must_use]
    pub fn get_pipeline_mark_price_topic(&mut self, instrument_id: InstrumentId) -> MStr<Topic> {
        let live = self.get_mark_price_topic(instrument_id);
        self.pipeline_topic(live)
    }

    #[must_use]
    pub fn get_pipeline_index_price_topic(&mut self, instrument_id: InstrumentId) -> MStr<Topic> {
        let live = self.get_index_price_topic(instrument_id);
        self.pipeline_topic(live)
    }

    #[must_use]
    pub fn get_pipeline_funding_rate_topic(&mut self, instrument_id: InstrumentId) -> MStr<Topic> {
        let live = self.get_funding_rate_topic(instrument_id);
        self.pipeline_topic(live)
    }

    #[must_use]
    pub fn get_pipeline_instrument_status_topic(
        &mut self,
        instrument_id: InstrumentId,
    ) -> MStr<Topic> {
        let live = self.get_instrument_status_topic(instrument_id);
        self.pipeline_topic(live)
    }

    #[must_use]
    pub fn get_pipeline_option_greeks_topic(&mut self, instrument_id: InstrumentId) -> MStr<Topic> {
        let live = self.get_option_greeks_topic(instrument_id);
        self.pipeline_topic(live)
    }

    #[must_use]
    pub fn get_pipeline_instrument_close_topic(
        &mut self,
        instrument_id: InstrumentId,
    ) -> MStr<Topic> {
        let live = self.get_instrument_close_topic(instrument_id);
        self.pipeline_topic(live)
    }

    /// Returns the subscription pattern for order book deltas on `instrument_id`.
    #[must_use]
    pub fn get_book_deltas_pattern(&mut self, instrument_id: InstrumentId) -> MStr<Pattern> {
        *self
            .book_deltas_patterns
            .entry(instrument_id)
            .or_insert_with(|| {
                format!(
                    "data.book.deltas.{}.{}",
                    instrument_id.venue,
                    instrument_id.symbol.topic(),
                )
                .into()
            })
    }

    /// Returns the subscription pattern for order book depth snapshots on `instrument_id`.
    #[must_use]
    pub fn get_book_depth_pattern(&mut self, instrument_id: InstrumentId) -> MStr<Pattern> {
        *self
            .book_depth_patterns
            .entry(instrument_id)
            .or_insert_with(|| {
                format!(
                    "data.book.depth.{}.{}",
                    instrument_id.venue,
                    instrument_id.symbol.topic(),
                )
                .into()
            })
    }

    /// Returns the subscription pattern for periodic order book snapshots on `instrument_id`.
    #[must_use]
    pub fn get_book_snapshots_pattern(
        &mut self,
        instrument_id: InstrumentId,
        interval_ms: NonZeroUsize,
    ) -> MStr<Pattern> {
        *self
            .book_snapshots_patterns
            .entry((instrument_id, interval_ms))
            .or_insert_with(|| {
                format!(
                    "data.book.snapshots.{}.{}.{}",
                    instrument_id.venue,
                    instrument_id.symbol.topic(),
                    interval_ms,
                )
                .into()
            })
    }
}

macro_rules! define_wrappers {
    ($($method:ident($($arg_name:ident: $arg_ty:ty),*) -> $ret:ty),* $(,)?) => {
        $(
            #[must_use]
            pub fn $method($($arg_name: $arg_ty),*) -> $ret {
                get_message_bus()
                    .borrow_mut()
                    .switchboard
                    .$method($($arg_name),*)
            }
        )*
    }
}

define_wrappers! {
    get_custom_topic(data_type: &DataType) -> MStr<Topic>,
    get_instruments_topic(venue: Venue) -> MStr<Topic>,
    get_instrument_topic(instrument_id: InstrumentId) -> MStr<Topic>,
    get_book_deltas_topic(instrument_id: InstrumentId) -> MStr<Topic>,
    get_book_depth_topic(instrument_id: InstrumentId) -> MStr<Topic>,
    get_book_snapshots_topic(instrument_id: InstrumentId, interval_ms: NonZeroUsize) -> MStr<Topic>,
    get_quotes_topic(instrument_id: InstrumentId) -> MStr<Topic>,
    get_trades_topic(instrument_id: InstrumentId) -> MStr<Topic>,
    get_bars_topic(bar_type: BarType) -> MStr<Topic>,
    get_mark_price_topic(instrument_id: InstrumentId) -> MStr<Topic>,
    get_index_price_topic(instrument_id: InstrumentId) -> MStr<Topic>,
    get_funding_rate_topic(instrument_id: InstrumentId) -> MStr<Topic>,
    get_funding_settlement_topic(instrument_id: InstrumentId) -> MStr<Topic>,
    get_instrument_status_topic(instrument_id: InstrumentId) -> MStr<Topic>,
    get_instrument_close_topic(instrument_id: InstrumentId) -> MStr<Topic>,
    get_option_greeks_topic(instrument_id: InstrumentId) -> MStr<Topic>,
    get_option_chain_topic(series_id: OptionSeriesId) -> MStr<Topic>,
    get_pipeline_custom_topic(data_type: &DataType) -> MStr<Topic>,
    get_pipeline_book_deltas_topic(instrument_id: InstrumentId) -> MStr<Topic>,
    get_pipeline_book_depth_topic(instrument_id: InstrumentId) -> MStr<Topic>,
    get_pipeline_quotes_topic(instrument_id: InstrumentId) -> MStr<Topic>,
    get_pipeline_trades_topic(instrument_id: InstrumentId) -> MStr<Topic>,
    get_pipeline_bars_topic(bar_type: BarType) -> MStr<Topic>,
    get_pipeline_mark_price_topic(instrument_id: InstrumentId) -> MStr<Topic>,
    get_pipeline_index_price_topic(instrument_id: InstrumentId) -> MStr<Topic>,
    get_pipeline_funding_rate_topic(instrument_id: InstrumentId) -> MStr<Topic>,
    get_pipeline_instrument_status_topic(instrument_id: InstrumentId) -> MStr<Topic>,
    get_pipeline_option_greeks_topic(instrument_id: InstrumentId) -> MStr<Topic>,
    get_pipeline_instrument_close_topic(instrument_id: InstrumentId) -> MStr<Topic>,
    get_order_submitted_topic(instrument_id: InstrumentId) -> MStr<Topic>,
    get_order_rejected_topic(instrument_id: InstrumentId) -> MStr<Topic>,
    get_order_pending_update_topic(instrument_id: InstrumentId) -> MStr<Topic>,
    get_order_pending_cancel_topic(instrument_id: InstrumentId) -> MStr<Topic>,
    get_order_modify_rejected_topic(instrument_id: InstrumentId) -> MStr<Topic>,
    get_order_cancel_rejected_topic(instrument_id: InstrumentId) -> MStr<Topic>,
    get_order_canceled_topic(instrument_id: InstrumentId) -> MStr<Topic>,
    get_order_filled_topic(instrument_id: InstrumentId) -> MStr<Topic>,
    get_order_fill_voided_topic(instrument_id: InstrumentId) -> MStr<Topic>,
    get_order_fill_declined_topic(instrument_id: InstrumentId) -> MStr<Topic>,
    get_snapshot_order_topic(client_order_id: ClientOrderId) -> MStr<Topic>,
    get_snapshot_position_topic(position_id: PositionId) -> MStr<Topic>,
    get_event_order_topic(strategy_id: StrategyId) -> MStr<Topic>,
    get_event_position_topic(strategy_id: StrategyId) -> MStr<Topic>,
}

/// Returns a wildcard subscription pattern that matches all instrument topics
/// for the given `venue`.
///
/// For example, venue `BINANCE` produces pattern `data.instrument.BINANCE.*`,
/// which matches per-instrument topics like `data.instrument.BINANCE.BTCUSDT`.
#[must_use]
pub fn get_instruments_pattern(venue: Venue) -> MStr<Pattern> {
    get_message_bus()
        .borrow_mut()
        .switchboard
        .instruments_pattern(venue)
}

/// Returns the subscription pattern for order book deltas on `instrument_id`.
#[must_use]
pub fn get_book_deltas_pattern(instrument_id: InstrumentId) -> MStr<Pattern> {
    get_message_bus()
        .borrow_mut()
        .switchboard
        .get_book_deltas_pattern(instrument_id)
}

/// Returns the subscription pattern for order book depth snapshots on `instrument_id`.
#[must_use]
pub fn get_book_depth_pattern(instrument_id: InstrumentId) -> MStr<Pattern> {
    get_message_bus()
        .borrow_mut()
        .switchboard
        .get_book_depth_pattern(instrument_id)
}

/// Returns the subscription pattern for periodic order book snapshots on `instrument_id`.
#[must_use]
pub fn get_book_snapshots_pattern(
    instrument_id: InstrumentId,
    interval_ms: NonZeroUsize,
) -> MStr<Pattern> {
    get_message_bus()
        .borrow_mut()
        .switchboard
        .get_book_snapshots_pattern(instrument_id, interval_ms)
}

/// Returns the exact signal publish topic for `name` (`data.Signal<TitleName>`).
#[must_use]
pub fn get_signal_topic(name: &str) -> MStr<Topic> {
    get_message_bus()
        .borrow_mut()
        .switchboard
        .signal_topic(name)
}

/// Returns the signal subscription pattern for `name` (`data.Signal<TitleName>*`).
///
/// An empty `name` yields the wildcard `data.Signal*` matching every signal topic.
#[must_use]
pub fn get_signal_pattern(name: &str) -> MStr<Pattern> {
    get_message_bus()
        .borrow_mut()
        .switchboard
        .signal_pattern(name)
}

/// Returns subscriptions for a custom data type and its optional identifier scope.
#[must_use]
pub fn get_custom_subscription_topics(data_type: &DataType) -> Vec<MStr<Pattern>> {
    let topic = get_custom_topic(data_type);
    let mut topics = vec![topic.into()];
    if data_type.identifier().is_none() {
        topics.push(MStr::pattern(format!("{topic}{IDENTIFIER_TOPIC_SUFFIX}*")));
    }
    topics
}

fn state_topic_component(value: &str) -> String {
    let mut encoded = String::with_capacity(value.len());
    for byte in value.bytes() {
        if byte.is_ascii_alphanumeric() || matches!(byte, b'-' | b'_') {
            encoded.push(char::from(byte));
        } else {
            write!(encoded, "%{byte:02X}").expect("String formatting cannot fail");
        }
    }

    encoded
}

#[cfg(test)]
mod tests {
    use nautilus_model::{
        data::{BarType, DataType},
        identifiers::{InstrumentId, Venue},
    };
    use rstest::*;

    use super::*;
    use crate::msgbus::matching::is_matching_backtracking;

    #[fixture]
    fn switchboard() -> MessagingSwitchboard {
        MessagingSwitchboard::default()
    }

    #[fixture]
    fn instrument_id() -> InstrumentId {
        InstrumentId::from("ESZ24.XCME")
    }

    #[rstest]
    fn test_data_response_topic() {
        let expected_topic = "data.response".into();
        let result = MessagingSwitchboard::data_response_topic();
        assert_eq!(result, expected_topic);
    }

    #[rstest]
    fn test_time_event_topic() {
        let expected_topic = "clock.time_event".into();
        let result = MessagingSwitchboard::time_event_topic();
        assert_eq!(result, expected_topic);
    }

    #[rstest]
    fn test_reconciliation_raw_order_status_report_topic() {
        let expected_topic = "reconciliation.raw.OrderStatusReport".into();
        let result = MessagingSwitchboard::reconciliation_raw_order_status_report_topic();
        assert_eq!(result, expected_topic);
    }

    #[rstest]
    fn test_reconciliation_raw_fill_report_topic() {
        let expected_topic = "reconciliation.raw.FillReport".into();
        let result = MessagingSwitchboard::reconciliation_raw_fill_report_topic();
        assert_eq!(result, expected_topic);
    }

    #[rstest]
    fn test_reconciliation_raw_position_status_report_topic() {
        let expected_topic = "reconciliation.raw.PositionStatusReport".into();
        let result = MessagingSwitchboard::reconciliation_raw_position_status_report_topic();
        assert_eq!(result, expected_topic);
    }

    #[rstest]
    fn test_get_custom_topic(mut switchboard: MessagingSwitchboard) {
        let data_type = DataType::new("ExampleDataType", None, None);
        let expected_topic = "data.ExampleDataType".into();
        let result = switchboard.get_custom_topic(&data_type);
        assert_eq!(result, expected_topic);
        assert!(switchboard.custom_topics.contains_key(&data_type));
    }

    #[rstest]
    fn test_get_instrument_topic(
        mut switchboard: MessagingSwitchboard,
        instrument_id: InstrumentId,
    ) {
        let expected_topic = "data.instrument.XCME.ESZ24".into();
        let result = switchboard.get_instrument_topic(instrument_id);
        assert_eq!(result, expected_topic);
        assert!(switchboard.instrument_topics.contains_key(&instrument_id));
    }

    #[rstest]
    fn test_get_book_deltas_topic(
        mut switchboard: MessagingSwitchboard,
        instrument_id: InstrumentId,
    ) {
        let expected_topic = "data.book.deltas.XCME.ESZ24".into();
        let result = switchboard.get_book_deltas_topic(instrument_id);
        assert_eq!(result, expected_topic);
        assert!(switchboard.book_deltas_topics.contains_key(&instrument_id));
    }

    #[rstest]
    fn test_get_book_depth_topic(
        mut switchboard: MessagingSwitchboard,
        instrument_id: InstrumentId,
    ) {
        let expected_topic = "data.book.depth.XCME.ESZ24".into();
        let result = switchboard.get_book_depth_topic(instrument_id);
        assert_eq!(result, expected_topic);
        assert!(switchboard.book_depth_topics.contains_key(&instrument_id));
    }

    #[rstest]
    fn test_get_book_snapshots_topic(
        mut switchboard: MessagingSwitchboard,
        instrument_id: InstrumentId,
    ) {
        let expected_topic = "data.book.snapshots.XCME.ESZ24.1000".into();
        let interval_ms = NonZeroUsize::new(1000).unwrap();
        let result = switchboard.get_book_snapshots_topic(instrument_id, interval_ms);
        assert_eq!(result, expected_topic);

        assert!(
            switchboard
                .book_snapshots_topics
                .contains_key(&(instrument_id, interval_ms))
        );
    }

    #[rstest]
    fn test_get_quotes_topic(mut switchboard: MessagingSwitchboard, instrument_id: InstrumentId) {
        let expected_topic = "data.quotes.XCME.ESZ24".into();
        let result = switchboard.get_quotes_topic(instrument_id);
        assert_eq!(result, expected_topic);
        assert!(switchboard.quote_topics.contains_key(&instrument_id));
    }

    #[rstest]
    fn test_get_trades_topic(mut switchboard: MessagingSwitchboard, instrument_id: InstrumentId) {
        let expected_topic = "data.trades.XCME.ESZ24".into();
        let result = switchboard.get_trades_topic(instrument_id);
        assert_eq!(result, expected_topic);
        assert!(switchboard.trade_topics.contains_key(&instrument_id));
    }

    #[rstest]
    fn test_get_bars_topic(mut switchboard: MessagingSwitchboard) {
        let bar_type = BarType::from("ESZ24.XCME-1-MINUTE-LAST-INTERNAL");
        let expected_topic = format!("data.bars.{bar_type}").into();
        let result = switchboard.get_bars_topic(bar_type);
        assert_eq!(result, expected_topic);
        assert!(switchboard.bar_topics.contains_key(&bar_type));
    }

    #[rstest]
    fn test_get_pipeline_custom_topic(mut switchboard: MessagingSwitchboard) {
        let data_type = DataType::new("ExampleDataType", None, None);
        let expected_topic = "data.pipeline.ExampleDataType".into();
        let result = switchboard.get_pipeline_custom_topic(&data_type);
        assert_eq!(result, expected_topic);
        assert!(switchboard.custom_topics.contains_key(&data_type));
        assert_eq!(switchboard.pipeline_topics.len(), 1);
    }

    type PipelineInstrumentIdTopicFn = fn(&mut MessagingSwitchboard, InstrumentId) -> MStr<Topic>;

    #[rstest]
    #[case::book_deltas(
        MessagingSwitchboard::get_pipeline_book_deltas_topic as PipelineInstrumentIdTopicFn,
        "data.pipeline.book.deltas.XCME.ESZ24",
    )]
    #[case::book_depth(
        MessagingSwitchboard::get_pipeline_book_depth_topic as PipelineInstrumentIdTopicFn,
        "data.pipeline.book.depth.XCME.ESZ24",
    )]
    #[case::quotes(
        MessagingSwitchboard::get_pipeline_quotes_topic as PipelineInstrumentIdTopicFn,
        "data.pipeline.quotes.XCME.ESZ24",
    )]
    #[case::trades(
        MessagingSwitchboard::get_pipeline_trades_topic as PipelineInstrumentIdTopicFn,
        "data.pipeline.trades.XCME.ESZ24",
    )]
    #[case::mark_prices(
        MessagingSwitchboard::get_pipeline_mark_price_topic as PipelineInstrumentIdTopicFn,
        "data.pipeline.mark_prices.XCME.ESZ24",
    )]
    #[case::index_prices(
        MessagingSwitchboard::get_pipeline_index_price_topic as PipelineInstrumentIdTopicFn,
        "data.pipeline.index_prices.XCME.ESZ24",
    )]
    #[case::funding_rates(
        MessagingSwitchboard::get_pipeline_funding_rate_topic as PipelineInstrumentIdTopicFn,
        "data.pipeline.funding_rates.XCME.ESZ24",
    )]
    #[case::status(
        MessagingSwitchboard::get_pipeline_instrument_status_topic as PipelineInstrumentIdTopicFn,
        "data.pipeline.status.XCME.ESZ24",
    )]
    #[case::close(
        MessagingSwitchboard::get_pipeline_instrument_close_topic as PipelineInstrumentIdTopicFn,
        "data.pipeline.close.XCME.ESZ24",
    )]
    fn test_get_pipeline_instrument_id_topic(
        mut switchboard: MessagingSwitchboard,
        instrument_id: InstrumentId,
        #[case] topic_fn: PipelineInstrumentIdTopicFn,
        #[case] expected: &str,
    ) {
        let result = topic_fn(&mut switchboard, instrument_id);
        assert_eq!(result.as_ref(), expected);
        assert_eq!(switchboard.pipeline_topics.len(), 1);
    }

    #[rstest]
    fn test_get_pipeline_bars_topic(mut switchboard: MessagingSwitchboard) {
        let bar_type = BarType::from("ESZ24.XCME-1-MINUTE-LAST-INTERNAL");
        let expected_topic = format!("data.pipeline.bars.{bar_type}").into();
        let result = switchboard.get_pipeline_bars_topic(bar_type);
        assert_eq!(result, expected_topic);
        assert!(switchboard.bar_topics.contains_key(&bar_type));
        assert_eq!(switchboard.pipeline_topics.len(), 1);
    }

    type OrderEventTopicFn = fn(&mut MessagingSwitchboard, InstrumentId) -> MStr<Topic>;

    #[rstest]
    #[case::submitted(
        MessagingSwitchboard::get_order_submitted_topic as OrderEventTopicFn,
        "events.order_submitted.ESZ24.XCME",
    )]
    #[case::rejected(
        MessagingSwitchboard::get_order_rejected_topic as OrderEventTopicFn,
        "events.order_rejected.ESZ24.XCME",
    )]
    #[case::pending_update(
        MessagingSwitchboard::get_order_pending_update_topic as OrderEventTopicFn,
        "events.order_pending_update.ESZ24.XCME",
    )]
    #[case::pending_cancel(
        MessagingSwitchboard::get_order_pending_cancel_topic as OrderEventTopicFn,
        "events.order_pending_cancel.ESZ24.XCME",
    )]
    #[case::modify_rejected(
        MessagingSwitchboard::get_order_modify_rejected_topic as OrderEventTopicFn,
        "events.order_modify_rejected.ESZ24.XCME",
    )]
    #[case::cancel_rejected(
        MessagingSwitchboard::get_order_cancel_rejected_topic as OrderEventTopicFn,
        "events.order_cancel_rejected.ESZ24.XCME",
    )]
    #[case::canceled(
        MessagingSwitchboard::get_order_canceled_topic as OrderEventTopicFn,
        "events.order_canceled.ESZ24.XCME",
    )]
    #[case::filled(
        MessagingSwitchboard::get_order_filled_topic as OrderEventTopicFn,
        "events.order_filled.ESZ24.XCME",
    )]
    #[case::fill_voided(
        MessagingSwitchboard::get_order_fill_voided_topic as OrderEventTopicFn,
        "events.order_fill_voided.ESZ24.XCME",
    )]
    fn test_get_order_event_topic(
        mut switchboard: MessagingSwitchboard,
        instrument_id: InstrumentId,
        #[case] topic_fn: OrderEventTopicFn,
        #[case] expected: &str,
    ) {
        let result = topic_fn(&mut switchboard, instrument_id);
        assert_eq!(result.as_ref(), expected);
    }

    #[rstest]
    #[case::submitted(MessagingSwitchboard::get_order_submitted_topic as OrderEventTopicFn)]
    #[case::rejected(MessagingSwitchboard::get_order_rejected_topic as OrderEventTopicFn)]
    #[case::pending_update(MessagingSwitchboard::get_order_pending_update_topic as OrderEventTopicFn)]
    #[case::pending_cancel(MessagingSwitchboard::get_order_pending_cancel_topic as OrderEventTopicFn)]
    #[case::modify_rejected(MessagingSwitchboard::get_order_modify_rejected_topic as OrderEventTopicFn)]
    #[case::cancel_rejected(MessagingSwitchboard::get_order_cancel_rejected_topic as OrderEventTopicFn)]
    #[case::canceled(MessagingSwitchboard::get_order_canceled_topic as OrderEventTopicFn)]
    #[case::filled(MessagingSwitchboard::get_order_filled_topic as OrderEventTopicFn)]
    #[case::fill_voided(MessagingSwitchboard::get_order_fill_voided_topic as OrderEventTopicFn)]
    fn test_order_event_topic_does_not_match_strategy_order_pattern(
        mut switchboard: MessagingSwitchboard,
        instrument_id: InstrumentId,
        #[case] topic_fn: OrderEventTopicFn,
    ) {
        let topic = topic_fn(&mut switchboard, instrument_id);
        assert!(!is_matching_backtracking(topic, "events.order.*".into()));
    }

    #[rstest]
    fn test_get_snapshot_order_topic(mut switchboard: MessagingSwitchboard) {
        let client_order_id = ClientOrderId::from("O-123456789");
        let expected_topic = format!("snapshots.order.{client_order_id}").into();
        let result = switchboard.get_snapshot_order_topic(client_order_id);
        assert_eq!(result, expected_topic);
        assert!(
            switchboard
                .snapshot_order_topics
                .contains_key(&client_order_id)
        );
    }

    #[rstest]
    fn test_get_snapshot_position_topic(mut switchboard: MessagingSwitchboard) {
        let position_id = PositionId::from("P-123456789");
        let expected_topic = format!("snapshots.position.{position_id}").into();
        let result = switchboard.get_snapshot_position_topic(position_id);
        assert_eq!(result, expected_topic);
        assert!(
            switchboard
                .snapshot_position_topics
                .contains_key(&position_id)
        );
    }

    #[rstest]
    #[case(
        "CLIENT.A",
        "orders",
        "events.system.SocketStateChanged.CLIENT%2EA.orders"
    )]
    #[case(
        "CLIENT*?",
        "public.market",
        "events.system.SocketStateChanged.CLIENT%2A%3F.public%2Emarket"
    )]
    #[case(
        "CLIENT%2E",
        "market",
        "events.system.SocketStateChanged.CLIENT%252E.market"
    )]
    fn test_socket_state_topic_encodes_literal_components(
        #[case] client_id: &str,
        #[case] endpoint: &str,
        #[case] expected: &str,
    ) {
        let topic =
            MessagingSwitchboard::socket_state_changed_topic(ClientId::from(client_id), endpoint);
        assert_eq!(topic.as_ref(), expected);
    }

    #[rstest]
    #[case(Some("CLIENT"), None, "CLIENT.A", "market", false)]
    #[case(None, Some("market"), "CLIENT", "public.market", false)]
    #[case(Some("CLIENT*"), None, "CLIENT1", "market", false)]
    #[case(Some("CLIENT?"), None, "CLIENT1", "market", false)]
    #[case(
        Some("CLIENT.A"),
        Some("public.market"),
        "CLIENT.A",
        "public.market",
        true
    )]
    #[case(None, Some("public.market"), "CLIENT.A", "public.market", true)]
    fn test_socket_state_pattern_matches_literal_fields(
        #[case] client_filter: Option<&str>,
        #[case] endpoint_filter: Option<&str>,
        #[case] client_id: &str,
        #[case] endpoint: &str,
        #[case] expected: bool,
    ) {
        let topic =
            MessagingSwitchboard::socket_state_changed_topic(ClientId::from(client_id), endpoint);
        let pattern = MessagingSwitchboard::socket_state_changed_pattern(
            client_filter.map(ClientId::from),
            endpoint_filter,
        );
        assert_eq!(is_matching_backtracking(topic, pattern), expected);
    }

    #[rstest]
    fn test_queue_state_changed_topic_identity() {
        assert_eq!(
            MessagingSwitchboard::queue_state_changed_topic(SystemChannel::ExecCommands).as_ref(),
            "events.system.QueueStateChanged.ExecCommands"
        );
    }

    #[rstest]
    fn test_socket_state_changed_topic_identity() {
        assert_eq!(
            MessagingSwitchboard::socket_state_changed_topic(ClientId::from("BINANCE"), "market")
                .as_ref(),
            "events.system.SocketStateChanged.BINANCE.market"
        );
    }

    #[rstest]
    fn test_instruments_pattern_matches_instrument_topic(
        mut switchboard: MessagingSwitchboard,
        instrument_id: InstrumentId,
    ) {
        let venue = instrument_id.venue;
        let pattern = switchboard.instruments_pattern(venue);
        let topic = switchboard.get_instrument_topic(instrument_id);

        assert_eq!(pattern.as_ref(), "data.instrument.XCME.*");
        assert!(is_matching_backtracking(topic, pattern));
    }

    #[rstest]
    fn test_instruments_pattern_does_not_match_other_venue(mut switchboard: MessagingSwitchboard) {
        let pattern = switchboard.instruments_pattern(Venue::from("BINANCE"));
        let topic = switchboard.get_instrument_topic(InstrumentId::from("ESZ24.XCME"));

        assert!(!is_matching_backtracking(topic, pattern));
    }

    #[rstest]
    fn test_composite_book_deltas_pattern_uses_wildcard(mut switchboard: MessagingSwitchboard) {
        let composite_id = InstrumentId::from("ES.FUT.XCME");
        let underlying_id = InstrumentId::from("ESZ24.XCME");

        let composite_pattern = switchboard.get_book_deltas_pattern(composite_id);
        let underlying_topic = switchboard.get_book_deltas_topic(underlying_id);

        assert_eq!(composite_pattern.as_ref(), "data.book.deltas.XCME.ES*");
        assert_eq!(underlying_topic.as_ref(), "data.book.deltas.XCME.ESZ24");
        assert!(is_matching_backtracking(
            underlying_topic,
            composite_pattern
        ));
    }

    #[rstest]
    fn test_book_deltas_pattern_for_non_composite_is_literal(
        mut switchboard: MessagingSwitchboard,
        instrument_id: InstrumentId,
    ) {
        let pattern = switchboard.get_book_deltas_pattern(instrument_id);
        assert_eq!(pattern.as_ref(), "data.book.deltas.XCME.ESZ24");
    }

    type PatternFn = fn(&mut MessagingSwitchboard, InstrumentId) -> MStr<Pattern>;

    #[rstest]
    #[case::book_depth(
        MessagingSwitchboard::get_book_depth_pattern as PatternFn,
        "data.book.depth.XCME.ESZ24",
    )]
    fn test_pattern_for_non_composite_is_literal(
        mut switchboard: MessagingSwitchboard,
        instrument_id: InstrumentId,
        #[case] pattern_fn: PatternFn,
        #[case] expected: &str,
    ) {
        let pattern = pattern_fn(&mut switchboard, instrument_id);
        assert_eq!(pattern.as_ref(), expected);
    }

    #[rstest]
    fn test_book_snapshots_pattern_for_non_composite_is_literal(
        mut switchboard: MessagingSwitchboard,
        instrument_id: InstrumentId,
    ) {
        let interval_ms = NonZeroUsize::new(1000).unwrap();
        let pattern = switchboard.get_book_snapshots_pattern(instrument_id, interval_ms);
        assert_eq!(pattern.as_ref(), "data.book.snapshots.XCME.ESZ24.1000");
    }

    #[rstest]
    #[case::book_deltas(MessagingSwitchboard::get_book_deltas_pattern as PatternFn)]
    #[case::book_depth(MessagingSwitchboard::get_book_depth_pattern as PatternFn)]
    fn test_pattern_function_is_idempotent(
        mut switchboard: MessagingSwitchboard,
        instrument_id: InstrumentId,
        #[case] pattern_fn: PatternFn,
    ) {
        let first = pattern_fn(&mut switchboard, instrument_id);
        let second = pattern_fn(&mut switchboard, instrument_id);
        assert_eq!(first, second);
    }

    #[rstest]
    fn test_book_snapshots_pattern_is_idempotent(
        mut switchboard: MessagingSwitchboard,
        instrument_id: InstrumentId,
    ) {
        let interval_ms = NonZeroUsize::new(1000).unwrap();
        let first = switchboard.get_book_snapshots_pattern(instrument_id, interval_ms);
        let second = switchboard.get_book_snapshots_pattern(instrument_id, interval_ms);
        assert_eq!(first, second);
    }

    #[rstest]
    fn test_composite_book_depth_pattern_uses_wildcard(mut switchboard: MessagingSwitchboard) {
        let composite_id = InstrumentId::from("ES.FUT.XCME");
        let underlying_id = InstrumentId::from("ESZ24.XCME");

        let composite_pattern = switchboard.get_book_depth_pattern(composite_id);
        let underlying_topic = switchboard.get_book_depth_topic(underlying_id);

        assert_eq!(composite_pattern.as_ref(), "data.book.depth.XCME.ES*");
        assert!(is_matching_backtracking(
            underlying_topic,
            composite_pattern
        ));
    }

    #[rstest]
    fn test_composite_book_snapshots_pattern_uses_wildcard(mut switchboard: MessagingSwitchboard) {
        let composite_id = InstrumentId::from("ES.FUT.XCME");
        let underlying_id = InstrumentId::from("ESZ24.XCME");
        let interval_ms = NonZeroUsize::new(1000).unwrap();

        let composite_pattern = switchboard.get_book_snapshots_pattern(composite_id, interval_ms);
        let underlying_topic = switchboard.get_book_snapshots_topic(underlying_id, interval_ms);

        assert_eq!(
            composite_pattern.as_ref(),
            "data.book.snapshots.XCME.ES*.1000"
        );
        assert_eq!(
            underlying_topic.as_ref(),
            "data.book.snapshots.XCME.ESZ24.1000"
        );
        assert!(is_matching_backtracking(
            underlying_topic,
            composite_pattern
        ));
    }
}