/*
 * Binance Derivatives Trading COIN Futures WebSocket Market Streams
 *
 * OpenAPI Specification for the Binance Derivatives Trading COIN Futures WebSocket Market Streams
 *
 * The version of the OpenAPI document: 1.0.0
 *
 *
 * NOTE: This class is auto generated by OpenAPI Generator (https://openapi-generator.tech).
 * https://openapi-generator.tech
 * Do not edit the class manually.
 */

#![allow(unused_imports)]
use async_trait::async_trait;
use derive_builder::Builder;
use serde::{Deserialize, Serialize};
use serde_json::Value;
use std::{collections::HashMap, sync::Arc};

use crate::common::{
    models::ParamBuildError,
    utils::replace_websocket_streams_placeholders,
    websocket::{WebsocketBase, WebsocketStream, WebsocketStreams, create_stream_handler},
};
use crate::derivatives_trading_coin_futures::websocket_streams::models;
use crate::models::StreamId;

#[async_trait]
pub trait WebsocketMarketStreamsApi: Send + Sync {
    async fn aggregate_trade_streams(
        &self,
        params: AggregateTradeStreamsParams,
    ) -> anyhow::Result<Arc<WebsocketStream<models::AggregateTradeStreamsResponse>>>;
    async fn all_book_tickers_stream(
        &self,
        params: AllBookTickersStreamParams,
    ) -> anyhow::Result<Arc<WebsocketStream<models::AllBookTickersStreamResponse>>>;
    async fn all_market_liquidation_order_streams(
        &self,
        params: AllMarketLiquidationOrderStreamsParams,
    ) -> anyhow::Result<Arc<WebsocketStream<models::AllMarketLiquidationOrderStreamsResponse>>>;
    async fn all_market_mini_tickers_stream(
        &self,
        params: AllMarketMiniTickersStreamParams,
    ) -> anyhow::Result<Arc<WebsocketStream<Vec<models::AllMarketMiniTickersStreamResponseInner>>>>;
    async fn all_market_tickers_streams(
        &self,
        params: AllMarketTickersStreamsParams,
    ) -> anyhow::Result<Arc<WebsocketStream<Vec<models::AllMarketTickersStreamsResponseInner>>>>;
    async fn continuous_contract_kline_candlestick_streams(
        &self,
        params: ContinuousContractKlineCandlestickStreamsParams,
    ) -> anyhow::Result<
        Arc<WebsocketStream<models::ContinuousContractKlineCandlestickStreamsResponse>>,
    >;
    async fn contract_info_stream(
        &self,
        params: ContractInfoStreamParams,
    ) -> anyhow::Result<Arc<WebsocketStream<models::ContractInfoStreamResponse>>>;
    async fn diff_book_depth_streams(
        &self,
        params: DiffBookDepthStreamsParams,
    ) -> anyhow::Result<Arc<WebsocketStream<models::DiffBookDepthStreamsResponse>>>;
    async fn index_kline_candlestick_streams(
        &self,
        params: IndexKlineCandlestickStreamsParams,
    ) -> anyhow::Result<Arc<WebsocketStream<models::IndexKlineCandlestickStreamsResponse>>>;
    async fn index_price_stream(
        &self,
        params: IndexPriceStreamParams,
    ) -> anyhow::Result<Arc<WebsocketStream<models::IndexPriceStreamResponse>>>;
    async fn individual_symbol_book_ticker_streams(
        &self,
        params: IndividualSymbolBookTickerStreamsParams,
    ) -> anyhow::Result<Arc<WebsocketStream<models::IndividualSymbolBookTickerStreamsResponse>>>;
    async fn individual_symbol_mini_ticker_stream(
        &self,
        params: IndividualSymbolMiniTickerStreamParams,
    ) -> anyhow::Result<Arc<WebsocketStream<models::IndividualSymbolMiniTickerStreamResponse>>>;
    async fn individual_symbol_ticker_streams(
        &self,
        params: IndividualSymbolTickerStreamsParams,
    ) -> anyhow::Result<Arc<WebsocketStream<models::IndividualSymbolTickerStreamsResponse>>>;
    async fn kline_candlestick_streams(
        &self,
        params: KlineCandlestickStreamsParams,
    ) -> anyhow::Result<Arc<WebsocketStream<models::KlineCandlestickStreamsResponse>>>;
    async fn liquidation_order_streams(
        &self,
        params: LiquidationOrderStreamsParams,
    ) -> anyhow::Result<Arc<WebsocketStream<models::LiquidationOrderStreamsResponse>>>;
    async fn mark_price_kline_candlestick_streams(
        &self,
        params: MarkPriceKlineCandlestickStreamsParams,
    ) -> anyhow::Result<Arc<WebsocketStream<models::MarkPriceKlineCandlestickStreamsResponse>>>;
    async fn mark_price_of_all_symbols_of_a_pair(
        &self,
        params: MarkPriceOfAllSymbolsOfAPairParams,
    ) -> anyhow::Result<Arc<WebsocketStream<Vec<models::MarkPriceOfAllSymbolsOfAPairResponseInner>>>>;
    async fn mark_price_stream(
        &self,
        params: MarkPriceStreamParams,
    ) -> anyhow::Result<Arc<WebsocketStream<models::MarkPriceStreamResponse>>>;
    async fn partial_book_depth_streams(
        &self,
        params: PartialBookDepthStreamsParams,
    ) -> anyhow::Result<Arc<WebsocketStream<models::PartialBookDepthStreamsResponse>>>;
}

pub struct WebsocketMarketStreamsApiClient {
    websocket_streams_base: Arc<WebsocketStreams>,
}

impl WebsocketMarketStreamsApiClient {
    pub fn new(websocket_streams_base: Arc<WebsocketStreams>) -> Self {
        Self {
            websocket_streams_base,
        }
    }
}

/// Request parameters for the [`aggregate_trade_streams`] operation.
///
/// This struct holds all of the inputs you can pass when calling
/// [`aggregate_trade_streams`](#method.aggregate_trade_streams).
#[derive(Clone, Debug, Builder)]
#[builder(pattern = "owned", build_fn(error = "ParamBuildError"))]
pub struct AggregateTradeStreamsParams {
    /// The symbol parameter
    ///
    /// This field is **required.
    #[builder(setter(into))]
    pub symbol: String,
    /// Unique WebSocket request ID.
    ///
    /// This field is **optional.
    #[builder(setter(into), default)]
    pub id: Option<String>,
}

impl AggregateTradeStreamsParams {
    /// Create a builder for [`aggregate_trade_streams`].
    ///
    /// Required parameters:
    ///
    /// * `symbol` — The symbol parameter
    ///
    #[must_use]
    pub fn builder(symbol: String) -> AggregateTradeStreamsParamsBuilder {
        AggregateTradeStreamsParamsBuilder::default().symbol(symbol)
    }
}
/// Request parameters for the [`all_book_tickers_stream`] operation.
///
/// This struct holds all of the inputs you can pass when calling
/// [`all_book_tickers_stream`](#method.all_book_tickers_stream).
#[derive(Clone, Debug, Builder, Default)]
#[builder(pattern = "owned", build_fn(error = "ParamBuildError"))]
pub struct AllBookTickersStreamParams {
    /// Unique WebSocket request ID.
    ///
    /// This field is **optional.
    #[builder(setter(into), default)]
    pub id: Option<String>,
}

impl AllBookTickersStreamParams {
    /// Create a builder for [`all_book_tickers_stream`].
    ///
    #[must_use]
    pub fn builder() -> AllBookTickersStreamParamsBuilder {
        AllBookTickersStreamParamsBuilder::default()
    }
}
/// Request parameters for the [`all_market_liquidation_order_streams`] operation.
///
/// This struct holds all of the inputs you can pass when calling
/// [`all_market_liquidation_order_streams`](#method.all_market_liquidation_order_streams).
#[derive(Clone, Debug, Builder, Default)]
#[builder(pattern = "owned", build_fn(error = "ParamBuildError"))]
pub struct AllMarketLiquidationOrderStreamsParams {
    /// Unique WebSocket request ID.
    ///
    /// This field is **optional.
    #[builder(setter(into), default)]
    pub id: Option<String>,
}

impl AllMarketLiquidationOrderStreamsParams {
    /// Create a builder for [`all_market_liquidation_order_streams`].
    ///
    #[must_use]
    pub fn builder() -> AllMarketLiquidationOrderStreamsParamsBuilder {
        AllMarketLiquidationOrderStreamsParamsBuilder::default()
    }
}
/// Request parameters for the [`all_market_mini_tickers_stream`] operation.
///
/// This struct holds all of the inputs you can pass when calling
/// [`all_market_mini_tickers_stream`](#method.all_market_mini_tickers_stream).
#[derive(Clone, Debug, Builder, Default)]
#[builder(pattern = "owned", build_fn(error = "ParamBuildError"))]
pub struct AllMarketMiniTickersStreamParams {
    /// Unique WebSocket request ID.
    ///
    /// This field is **optional.
    #[builder(setter(into), default)]
    pub id: Option<String>,
}

impl AllMarketMiniTickersStreamParams {
    /// Create a builder for [`all_market_mini_tickers_stream`].
    ///
    #[must_use]
    pub fn builder() -> AllMarketMiniTickersStreamParamsBuilder {
        AllMarketMiniTickersStreamParamsBuilder::default()
    }
}
/// Request parameters for the [`all_market_tickers_streams`] operation.
///
/// This struct holds all of the inputs you can pass when calling
/// [`all_market_tickers_streams`](#method.all_market_tickers_streams).
#[derive(Clone, Debug, Builder, Default)]
#[builder(pattern = "owned", build_fn(error = "ParamBuildError"))]
pub struct AllMarketTickersStreamsParams {
    /// Unique WebSocket request ID.
    ///
    /// This field is **optional.
    #[builder(setter(into), default)]
    pub id: Option<String>,
}

impl AllMarketTickersStreamsParams {
    /// Create a builder for [`all_market_tickers_streams`].
    ///
    #[must_use]
    pub fn builder() -> AllMarketTickersStreamsParamsBuilder {
        AllMarketTickersStreamsParamsBuilder::default()
    }
}
/// Request parameters for the [`continuous_contract_kline_candlestick_streams`] operation.
///
/// This struct holds all of the inputs you can pass when calling
/// [`continuous_contract_kline_candlestick_streams`](#method.continuous_contract_kline_candlestick_streams).
#[derive(Clone, Debug, Builder)]
#[builder(pattern = "owned", build_fn(error = "ParamBuildError"))]
pub struct ContinuousContractKlineCandlestickStreamsParams {
    /// The pair parameter
    ///
    /// This field is **required.
    #[builder(setter(into))]
    pub pair: String,
    /// The contractType parameter
    ///
    /// This field is **required.
    #[builder(setter(into))]
    pub contract_type: String,
    /// The interval parameter
    ///
    /// This field is **required.
    #[builder(setter(into))]
    pub interval: String,
    /// Unique WebSocket request ID.
    ///
    /// This field is **optional.
    #[builder(setter(into), default)]
    pub id: Option<String>,
}

impl ContinuousContractKlineCandlestickStreamsParams {
    /// Create a builder for [`continuous_contract_kline_candlestick_streams`].
    ///
    /// Required parameters:
    ///
    /// * `pair` — The pair parameter
    /// * `contract_type` — The contractType parameter
    /// * `interval` — The interval parameter
    ///
    #[must_use]
    pub fn builder(
        pair: String,
        contract_type: String,
        interval: String,
    ) -> ContinuousContractKlineCandlestickStreamsParamsBuilder {
        ContinuousContractKlineCandlestickStreamsParamsBuilder::default()
            .pair(pair)
            .contract_type(contract_type)
            .interval(interval)
    }
}
/// Request parameters for the [`contract_info_stream`] operation.
///
/// This struct holds all of the inputs you can pass when calling
/// [`contract_info_stream`](#method.contract_info_stream).
#[derive(Clone, Debug, Builder, Default)]
#[builder(pattern = "owned", build_fn(error = "ParamBuildError"))]
pub struct ContractInfoStreamParams {
    /// Unique WebSocket request ID.
    ///
    /// This field is **optional.
    #[builder(setter(into), default)]
    pub id: Option<String>,
}

impl ContractInfoStreamParams {
    /// Create a builder for [`contract_info_stream`].
    ///
    #[must_use]
    pub fn builder() -> ContractInfoStreamParamsBuilder {
        ContractInfoStreamParamsBuilder::default()
    }
}
/// Request parameters for the [`diff_book_depth_streams`] operation.
///
/// This struct holds all of the inputs you can pass when calling
/// [`diff_book_depth_streams`](#method.diff_book_depth_streams).
#[derive(Clone, Debug, Builder)]
#[builder(pattern = "owned", build_fn(error = "ParamBuildError"))]
pub struct DiffBookDepthStreamsParams {
    /// The symbol parameter
    ///
    /// This field is **required.
    #[builder(setter(into))]
    pub symbol: String,
    /// Unique WebSocket request ID.
    ///
    /// This field is **optional.
    #[builder(setter(into), default)]
    pub id: Option<String>,
    /// WebSocket stream update speed
    ///
    /// This field is **optional.
    #[builder(setter(into), default)]
    pub update_speed: Option<String>,
}

impl DiffBookDepthStreamsParams {
    /// Create a builder for [`diff_book_depth_streams`].
    ///
    /// Required parameters:
    ///
    /// * `symbol` — The symbol parameter
    ///
    #[must_use]
    pub fn builder(symbol: String) -> DiffBookDepthStreamsParamsBuilder {
        DiffBookDepthStreamsParamsBuilder::default().symbol(symbol)
    }
}
/// Request parameters for the [`index_kline_candlestick_streams`] operation.
///
/// This struct holds all of the inputs you can pass when calling
/// [`index_kline_candlestick_streams`](#method.index_kline_candlestick_streams).
#[derive(Clone, Debug, Builder)]
#[builder(pattern = "owned", build_fn(error = "ParamBuildError"))]
pub struct IndexKlineCandlestickStreamsParams {
    /// The pair parameter
    ///
    /// This field is **required.
    #[builder(setter(into))]
    pub pair: String,
    /// The interval parameter
    ///
    /// This field is **required.
    #[builder(setter(into))]
    pub interval: String,
    /// Unique WebSocket request ID.
    ///
    /// This field is **optional.
    #[builder(setter(into), default)]
    pub id: Option<String>,
}

impl IndexKlineCandlestickStreamsParams {
    /// Create a builder for [`index_kline_candlestick_streams`].
    ///
    /// Required parameters:
    ///
    /// * `pair` — The pair parameter
    /// * `interval` — The interval parameter
    ///
    #[must_use]
    pub fn builder(pair: String, interval: String) -> IndexKlineCandlestickStreamsParamsBuilder {
        IndexKlineCandlestickStreamsParamsBuilder::default()
            .pair(pair)
            .interval(interval)
    }
}
/// Request parameters for the [`index_price_stream`] operation.
///
/// This struct holds all of the inputs you can pass when calling
/// [`index_price_stream`](#method.index_price_stream).
#[derive(Clone, Debug, Builder)]
#[builder(pattern = "owned", build_fn(error = "ParamBuildError"))]
pub struct IndexPriceStreamParams {
    /// The pair parameter
    ///
    /// This field is **required.
    #[builder(setter(into))]
    pub pair: String,
    /// Unique WebSocket request ID.
    ///
    /// This field is **optional.
    #[builder(setter(into), default)]
    pub id: Option<String>,
    /// WebSocket stream update speed
    ///
    /// This field is **optional.
    #[builder(setter(into), default)]
    pub update_speed: Option<String>,
}

impl IndexPriceStreamParams {
    /// Create a builder for [`index_price_stream`].
    ///
    /// Required parameters:
    ///
    /// * `pair` — The pair parameter
    ///
    #[must_use]
    pub fn builder(pair: String) -> IndexPriceStreamParamsBuilder {
        IndexPriceStreamParamsBuilder::default().pair(pair)
    }
}
/// Request parameters for the [`individual_symbol_book_ticker_streams`] operation.
///
/// This struct holds all of the inputs you can pass when calling
/// [`individual_symbol_book_ticker_streams`](#method.individual_symbol_book_ticker_streams).
#[derive(Clone, Debug, Builder)]
#[builder(pattern = "owned", build_fn(error = "ParamBuildError"))]
pub struct IndividualSymbolBookTickerStreamsParams {
    /// The symbol parameter
    ///
    /// This field is **required.
    #[builder(setter(into))]
    pub symbol: String,
    /// Unique WebSocket request ID.
    ///
    /// This field is **optional.
    #[builder(setter(into), default)]
    pub id: Option<String>,
}

impl IndividualSymbolBookTickerStreamsParams {
    /// Create a builder for [`individual_symbol_book_ticker_streams`].
    ///
    /// Required parameters:
    ///
    /// * `symbol` — The symbol parameter
    ///
    #[must_use]
    pub fn builder(symbol: String) -> IndividualSymbolBookTickerStreamsParamsBuilder {
        IndividualSymbolBookTickerStreamsParamsBuilder::default().symbol(symbol)
    }
}
/// Request parameters for the [`individual_symbol_mini_ticker_stream`] operation.
///
/// This struct holds all of the inputs you can pass when calling
/// [`individual_symbol_mini_ticker_stream`](#method.individual_symbol_mini_ticker_stream).
#[derive(Clone, Debug, Builder)]
#[builder(pattern = "owned", build_fn(error = "ParamBuildError"))]
pub struct IndividualSymbolMiniTickerStreamParams {
    /// The symbol parameter
    ///
    /// This field is **required.
    #[builder(setter(into))]
    pub symbol: String,
    /// Unique WebSocket request ID.
    ///
    /// This field is **optional.
    #[builder(setter(into), default)]
    pub id: Option<String>,
}

impl IndividualSymbolMiniTickerStreamParams {
    /// Create a builder for [`individual_symbol_mini_ticker_stream`].
    ///
    /// Required parameters:
    ///
    /// * `symbol` — The symbol parameter
    ///
    #[must_use]
    pub fn builder(symbol: String) -> IndividualSymbolMiniTickerStreamParamsBuilder {
        IndividualSymbolMiniTickerStreamParamsBuilder::default().symbol(symbol)
    }
}
/// Request parameters for the [`individual_symbol_ticker_streams`] operation.
///
/// This struct holds all of the inputs you can pass when calling
/// [`individual_symbol_ticker_streams`](#method.individual_symbol_ticker_streams).
#[derive(Clone, Debug, Builder)]
#[builder(pattern = "owned", build_fn(error = "ParamBuildError"))]
pub struct IndividualSymbolTickerStreamsParams {
    /// The symbol parameter
    ///
    /// This field is **required.
    #[builder(setter(into))]
    pub symbol: String,
    /// Unique WebSocket request ID.
    ///
    /// This field is **optional.
    #[builder(setter(into), default)]
    pub id: Option<String>,
}

impl IndividualSymbolTickerStreamsParams {
    /// Create a builder for [`individual_symbol_ticker_streams`].
    ///
    /// Required parameters:
    ///
    /// * `symbol` — The symbol parameter
    ///
    #[must_use]
    pub fn builder(symbol: String) -> IndividualSymbolTickerStreamsParamsBuilder {
        IndividualSymbolTickerStreamsParamsBuilder::default().symbol(symbol)
    }
}
/// Request parameters for the [`kline_candlestick_streams`] operation.
///
/// This struct holds all of the inputs you can pass when calling
/// [`kline_candlestick_streams`](#method.kline_candlestick_streams).
#[derive(Clone, Debug, Builder)]
#[builder(pattern = "owned", build_fn(error = "ParamBuildError"))]
pub struct KlineCandlestickStreamsParams {
    /// The symbol parameter
    ///
    /// This field is **required.
    #[builder(setter(into))]
    pub symbol: String,
    /// The interval parameter
    ///
    /// This field is **required.
    #[builder(setter(into))]
    pub interval: String,
    /// Unique WebSocket request ID.
    ///
    /// This field is **optional.
    #[builder(setter(into), default)]
    pub id: Option<String>,
}

impl KlineCandlestickStreamsParams {
    /// Create a builder for [`kline_candlestick_streams`].
    ///
    /// Required parameters:
    ///
    /// * `symbol` — The symbol parameter
    /// * `interval` — The interval parameter
    ///
    #[must_use]
    pub fn builder(symbol: String, interval: String) -> KlineCandlestickStreamsParamsBuilder {
        KlineCandlestickStreamsParamsBuilder::default()
            .symbol(symbol)
            .interval(interval)
    }
}
/// Request parameters for the [`liquidation_order_streams`] operation.
///
/// This struct holds all of the inputs you can pass when calling
/// [`liquidation_order_streams`](#method.liquidation_order_streams).
#[derive(Clone, Debug, Builder)]
#[builder(pattern = "owned", build_fn(error = "ParamBuildError"))]
pub struct LiquidationOrderStreamsParams {
    /// The symbol parameter
    ///
    /// This field is **required.
    #[builder(setter(into))]
    pub symbol: String,
    /// Unique WebSocket request ID.
    ///
    /// This field is **optional.
    #[builder(setter(into), default)]
    pub id: Option<String>,
}

impl LiquidationOrderStreamsParams {
    /// Create a builder for [`liquidation_order_streams`].
    ///
    /// Required parameters:
    ///
    /// * `symbol` — The symbol parameter
    ///
    #[must_use]
    pub fn builder(symbol: String) -> LiquidationOrderStreamsParamsBuilder {
        LiquidationOrderStreamsParamsBuilder::default().symbol(symbol)
    }
}
/// Request parameters for the [`mark_price_kline_candlestick_streams`] operation.
///
/// This struct holds all of the inputs you can pass when calling
/// [`mark_price_kline_candlestick_streams`](#method.mark_price_kline_candlestick_streams).
#[derive(Clone, Debug, Builder)]
#[builder(pattern = "owned", build_fn(error = "ParamBuildError"))]
pub struct MarkPriceKlineCandlestickStreamsParams {
    /// The symbol parameter
    ///
    /// This field is **required.
    #[builder(setter(into))]
    pub symbol: String,
    /// The interval parameter
    ///
    /// This field is **required.
    #[builder(setter(into))]
    pub interval: String,
    /// Unique WebSocket request ID.
    ///
    /// This field is **optional.
    #[builder(setter(into), default)]
    pub id: Option<String>,
}

impl MarkPriceKlineCandlestickStreamsParams {
    /// Create a builder for [`mark_price_kline_candlestick_streams`].
    ///
    /// Required parameters:
    ///
    /// * `symbol` — The symbol parameter
    /// * `interval` — The interval parameter
    ///
    #[must_use]
    pub fn builder(
        symbol: String,
        interval: String,
    ) -> MarkPriceKlineCandlestickStreamsParamsBuilder {
        MarkPriceKlineCandlestickStreamsParamsBuilder::default()
            .symbol(symbol)
            .interval(interval)
    }
}
/// Request parameters for the [`mark_price_of_all_symbols_of_a_pair`] operation.
///
/// This struct holds all of the inputs you can pass when calling
/// [`mark_price_of_all_symbols_of_a_pair`](#method.mark_price_of_all_symbols_of_a_pair).
#[derive(Clone, Debug, Builder)]
#[builder(pattern = "owned", build_fn(error = "ParamBuildError"))]
pub struct MarkPriceOfAllSymbolsOfAPairParams {
    /// The pair parameter
    ///
    /// This field is **required.
    #[builder(setter(into))]
    pub pair: String,
    /// Unique WebSocket request ID.
    ///
    /// This field is **optional.
    #[builder(setter(into), default)]
    pub id: Option<String>,
    /// WebSocket stream update speed
    ///
    /// This field is **optional.
    #[builder(setter(into), default)]
    pub update_speed: Option<String>,
}

impl MarkPriceOfAllSymbolsOfAPairParams {
    /// Create a builder for [`mark_price_of_all_symbols_of_a_pair`].
    ///
    /// Required parameters:
    ///
    /// * `pair` — The pair parameter
    ///
    #[must_use]
    pub fn builder(pair: String) -> MarkPriceOfAllSymbolsOfAPairParamsBuilder {
        MarkPriceOfAllSymbolsOfAPairParamsBuilder::default().pair(pair)
    }
}
/// Request parameters for the [`mark_price_stream`] operation.
///
/// This struct holds all of the inputs you can pass when calling
/// [`mark_price_stream`](#method.mark_price_stream).
#[derive(Clone, Debug, Builder)]
#[builder(pattern = "owned", build_fn(error = "ParamBuildError"))]
pub struct MarkPriceStreamParams {
    /// The symbol parameter
    ///
    /// This field is **required.
    #[builder(setter(into))]
    pub symbol: String,
    /// Unique WebSocket request ID.
    ///
    /// This field is **optional.
    #[builder(setter(into), default)]
    pub id: Option<String>,
    /// WebSocket stream update speed
    ///
    /// This field is **optional.
    #[builder(setter(into), default)]
    pub update_speed: Option<String>,
}

impl MarkPriceStreamParams {
    /// Create a builder for [`mark_price_stream`].
    ///
    /// Required parameters:
    ///
    /// * `symbol` — The symbol parameter
    ///
    #[must_use]
    pub fn builder(symbol: String) -> MarkPriceStreamParamsBuilder {
        MarkPriceStreamParamsBuilder::default().symbol(symbol)
    }
}
/// Request parameters for the [`partial_book_depth_streams`] operation.
///
/// This struct holds all of the inputs you can pass when calling
/// [`partial_book_depth_streams`](#method.partial_book_depth_streams).
#[derive(Clone, Debug, Builder)]
#[builder(pattern = "owned", build_fn(error = "ParamBuildError"))]
pub struct PartialBookDepthStreamsParams {
    /// The symbol parameter
    ///
    /// This field is **required.
    #[builder(setter(into))]
    pub symbol: String,
    /// The levels parameter
    ///
    /// This field is **required.
    #[builder(setter(into))]
    pub levels: i64,
    /// Unique WebSocket request ID.
    ///
    /// This field is **optional.
    #[builder(setter(into), default)]
    pub id: Option<String>,
    /// WebSocket stream update speed
    ///
    /// This field is **optional.
    #[builder(setter(into), default)]
    pub update_speed: Option<String>,
}

impl PartialBookDepthStreamsParams {
    /// Create a builder for [`partial_book_depth_streams`].
    ///
    /// Required parameters:
    ///
    /// * `symbol` — The symbol parameter
    /// * `levels` — The levels parameter
    ///
    #[must_use]
    pub fn builder(symbol: String, levels: i64) -> PartialBookDepthStreamsParamsBuilder {
        PartialBookDepthStreamsParamsBuilder::default()
            .symbol(symbol)
            .levels(levels)
    }
}

#[async_trait]
impl WebsocketMarketStreamsApi for WebsocketMarketStreamsApiClient {
    async fn aggregate_trade_streams(
        &self,
        params: AggregateTradeStreamsParams,
    ) -> anyhow::Result<Arc<WebsocketStream<models::AggregateTradeStreamsResponse>>> {
        let AggregateTradeStreamsParams { symbol, id } = params;

        let pairs: &[(&str, Option<String>)] =
            &[("symbol", Some(symbol.clone())), ("id", id.clone())];

        let vars: HashMap<_, _> = pairs
            .iter()
            .filter_map(|&(k, ref v)| v.clone().map(|v| (k, v)))
            .collect();

        let id_opt: Option<String> = vars.get("id").map(std::string::ToString::to_string);

        let stream = replace_websocket_streams_placeholders("/<symbol>@aggTrade", &vars);

        Ok(
            create_stream_handler::<models::AggregateTradeStreamsResponse>(
                WebsocketBase::WebsocketStreams(Arc::clone(&self.websocket_streams_base)),
                stream,
                id_opt.map(|s| {
                    if !s.is_empty() && s.bytes().all(|b| b.is_ascii_digit()) {
                        if let Ok(n) = s.parse::<u32>() {
                            return StreamId::Number(n);
                        }
                    }
                    StreamId::Str(s)
                }),
                None,
            )
            .await,
        )
    }

    async fn all_book_tickers_stream(
        &self,
        params: AllBookTickersStreamParams,
    ) -> anyhow::Result<Arc<WebsocketStream<models::AllBookTickersStreamResponse>>> {
        let AllBookTickersStreamParams { id } = params;

        let pairs: &[(&str, Option<String>)] = &[("id", id.clone())];

        let vars: HashMap<_, _> = pairs
            .iter()
            .filter_map(|&(k, ref v)| v.clone().map(|v| (k, v)))
            .collect();

        let id_opt: Option<String> = vars.get("id").map(std::string::ToString::to_string);

        let stream = replace_websocket_streams_placeholders("/!bookTicker", &vars);

        Ok(
            create_stream_handler::<models::AllBookTickersStreamResponse>(
                WebsocketBase::WebsocketStreams(Arc::clone(&self.websocket_streams_base)),
                stream,
                id_opt.map(|s| {
                    if !s.is_empty() && s.bytes().all(|b| b.is_ascii_digit()) {
                        if let Ok(n) = s.parse::<u32>() {
                            return StreamId::Number(n);
                        }
                    }
                    StreamId::Str(s)
                }),
                None,
            )
            .await,
        )
    }

    async fn all_market_liquidation_order_streams(
        &self,
        params: AllMarketLiquidationOrderStreamsParams,
    ) -> anyhow::Result<Arc<WebsocketStream<models::AllMarketLiquidationOrderStreamsResponse>>>
    {
        let AllMarketLiquidationOrderStreamsParams { id } = params;

        let pairs: &[(&str, Option<String>)] = &[("id", id.clone())];

        let vars: HashMap<_, _> = pairs
            .iter()
            .filter_map(|&(k, ref v)| v.clone().map(|v| (k, v)))
            .collect();

        let id_opt: Option<String> = vars.get("id").map(std::string::ToString::to_string);

        let stream = replace_websocket_streams_placeholders("/!forceOrder@arr", &vars);

        Ok(
            create_stream_handler::<models::AllMarketLiquidationOrderStreamsResponse>(
                WebsocketBase::WebsocketStreams(Arc::clone(&self.websocket_streams_base)),
                stream,
                id_opt.map(|s| {
                    if !s.is_empty() && s.bytes().all(|b| b.is_ascii_digit()) {
                        if let Ok(n) = s.parse::<u32>() {
                            return StreamId::Number(n);
                        }
                    }
                    StreamId::Str(s)
                }),
                None,
            )
            .await,
        )
    }

    async fn all_market_mini_tickers_stream(
        &self,
        params: AllMarketMiniTickersStreamParams,
    ) -> anyhow::Result<Arc<WebsocketStream<Vec<models::AllMarketMiniTickersStreamResponseInner>>>>
    {
        let AllMarketMiniTickersStreamParams { id } = params;

        let pairs: &[(&str, Option<String>)] = &[("id", id.clone())];

        let vars: HashMap<_, _> = pairs
            .iter()
            .filter_map(|&(k, ref v)| v.clone().map(|v| (k, v)))
            .collect();

        let id_opt: Option<String> = vars.get("id").map(std::string::ToString::to_string);

        let stream = replace_websocket_streams_placeholders("/!miniTicker@arr", &vars);

        Ok(
            create_stream_handler::<Vec<models::AllMarketMiniTickersStreamResponseInner>>(
                WebsocketBase::WebsocketStreams(Arc::clone(&self.websocket_streams_base)),
                stream,
                id_opt.map(|s| {
                    if !s.is_empty() && s.bytes().all(|b| b.is_ascii_digit()) {
                        if let Ok(n) = s.parse::<u32>() {
                            return StreamId::Number(n);
                        }
                    }
                    StreamId::Str(s)
                }),
                None,
            )
            .await,
        )
    }

    async fn all_market_tickers_streams(
        &self,
        params: AllMarketTickersStreamsParams,
    ) -> anyhow::Result<Arc<WebsocketStream<Vec<models::AllMarketTickersStreamsResponseInner>>>>
    {
        let AllMarketTickersStreamsParams { id } = params;

        let pairs: &[(&str, Option<String>)] = &[("id", id.clone())];

        let vars: HashMap<_, _> = pairs
            .iter()
            .filter_map(|&(k, ref v)| v.clone().map(|v| (k, v)))
            .collect();

        let id_opt: Option<String> = vars.get("id").map(std::string::ToString::to_string);

        let stream = replace_websocket_streams_placeholders("/!ticker@arr", &vars);

        Ok(
            create_stream_handler::<Vec<models::AllMarketTickersStreamsResponseInner>>(
                WebsocketBase::WebsocketStreams(Arc::clone(&self.websocket_streams_base)),
                stream,
                id_opt.map(|s| {
                    if !s.is_empty() && s.bytes().all(|b| b.is_ascii_digit()) {
                        if let Ok(n) = s.parse::<u32>() {
                            return StreamId::Number(n);
                        }
                    }
                    StreamId::Str(s)
                }),
                None,
            )
            .await,
        )
    }

    async fn continuous_contract_kline_candlestick_streams(
        &self,
        params: ContinuousContractKlineCandlestickStreamsParams,
    ) -> anyhow::Result<
        Arc<WebsocketStream<models::ContinuousContractKlineCandlestickStreamsResponse>>,
    > {
        let ContinuousContractKlineCandlestickStreamsParams {
            pair,
            contract_type,
            interval,
            id,
        } = params;

        let pairs: &[(&str, Option<String>)] = &[
            ("pair", Some(pair.clone())),
            ("contractType", Some(contract_type.clone())),
            ("interval", Some(interval.clone())),
            ("id", id.clone()),
        ];

        let vars: HashMap<_, _> = pairs
            .iter()
            .filter_map(|&(k, ref v)| v.clone().map(|v| (k, v)))
            .collect();

        let id_opt: Option<String> = vars.get("id").map(std::string::ToString::to_string);

        let stream = replace_websocket_streams_placeholders(
            "/<pair>_<contractType>@continuousKline_<interval>",
            &vars,
        );

        Ok(
            create_stream_handler::<models::ContinuousContractKlineCandlestickStreamsResponse>(
                WebsocketBase::WebsocketStreams(Arc::clone(&self.websocket_streams_base)),
                stream,
                id_opt.map(|s| {
                    if !s.is_empty() && s.bytes().all(|b| b.is_ascii_digit()) {
                        if let Ok(n) = s.parse::<u32>() {
                            return StreamId::Number(n);
                        }
                    }
                    StreamId::Str(s)
                }),
                None,
            )
            .await,
        )
    }

    async fn contract_info_stream(
        &self,
        params: ContractInfoStreamParams,
    ) -> anyhow::Result<Arc<WebsocketStream<models::ContractInfoStreamResponse>>> {
        let ContractInfoStreamParams { id } = params;

        let pairs: &[(&str, Option<String>)] = &[("id", id.clone())];

        let vars: HashMap<_, _> = pairs
            .iter()
            .filter_map(|&(k, ref v)| v.clone().map(|v| (k, v)))
            .collect();

        let id_opt: Option<String> = vars.get("id").map(std::string::ToString::to_string);

        let stream = replace_websocket_streams_placeholders("/!contractInfo", &vars);

        Ok(create_stream_handler::<models::ContractInfoStreamResponse>(
            WebsocketBase::WebsocketStreams(Arc::clone(&self.websocket_streams_base)),
            stream,
            id_opt.map(|s| {
                if !s.is_empty() && s.bytes().all(|b| b.is_ascii_digit()) {
                    if let Ok(n) = s.parse::<u32>() {
                        return StreamId::Number(n);
                    }
                }
                StreamId::Str(s)
            }),
            None,
        )
        .await)
    }

    async fn diff_book_depth_streams(
        &self,
        params: DiffBookDepthStreamsParams,
    ) -> anyhow::Result<Arc<WebsocketStream<models::DiffBookDepthStreamsResponse>>> {
        let DiffBookDepthStreamsParams {
            symbol,
            id,
            update_speed,
        } = params;

        let pairs: &[(&str, Option<String>)] = &[
            ("symbol", Some(symbol.clone())),
            ("id", id.clone()),
            ("updateSpeed", update_speed.clone()),
        ];

        let vars: HashMap<_, _> = pairs
            .iter()
            .filter_map(|&(k, ref v)| v.clone().map(|v| (k, v)))
            .collect();

        let id_opt: Option<String> = vars.get("id").map(std::string::ToString::to_string);

        let stream = replace_websocket_streams_placeholders("/<symbol>@depth@<updateSpeed>", &vars);

        Ok(
            create_stream_handler::<models::DiffBookDepthStreamsResponse>(
                WebsocketBase::WebsocketStreams(Arc::clone(&self.websocket_streams_base)),
                stream,
                id_opt.map(|s| {
                    if !s.is_empty() && s.bytes().all(|b| b.is_ascii_digit()) {
                        if let Ok(n) = s.parse::<u32>() {
                            return StreamId::Number(n);
                        }
                    }
                    StreamId::Str(s)
                }),
                None,
            )
            .await,
        )
    }

    async fn index_kline_candlestick_streams(
        &self,
        params: IndexKlineCandlestickStreamsParams,
    ) -> anyhow::Result<Arc<WebsocketStream<models::IndexKlineCandlestickStreamsResponse>>> {
        let IndexKlineCandlestickStreamsParams { pair, interval, id } = params;

        let pairs: &[(&str, Option<String>)] = &[
            ("pair", Some(pair.clone())),
            ("interval", Some(interval.clone())),
            ("id", id.clone()),
        ];

        let vars: HashMap<_, _> = pairs
            .iter()
            .filter_map(|&(k, ref v)| v.clone().map(|v| (k, v)))
            .collect();

        let id_opt: Option<String> = vars.get("id").map(std::string::ToString::to_string);

        let stream =
            replace_websocket_streams_placeholders("/<pair>@indexPriceKline_<interval>", &vars);

        Ok(
            create_stream_handler::<models::IndexKlineCandlestickStreamsResponse>(
                WebsocketBase::WebsocketStreams(Arc::clone(&self.websocket_streams_base)),
                stream,
                id_opt.map(|s| {
                    if !s.is_empty() && s.bytes().all(|b| b.is_ascii_digit()) {
                        if let Ok(n) = s.parse::<u32>() {
                            return StreamId::Number(n);
                        }
                    }
                    StreamId::Str(s)
                }),
                None,
            )
            .await,
        )
    }

    async fn index_price_stream(
        &self,
        params: IndexPriceStreamParams,
    ) -> anyhow::Result<Arc<WebsocketStream<models::IndexPriceStreamResponse>>> {
        let IndexPriceStreamParams {
            pair,
            id,
            update_speed,
        } = params;

        let pairs: &[(&str, Option<String>)] = &[
            ("pair", Some(pair.clone())),
            ("id", id.clone()),
            ("updateSpeed", update_speed.clone()),
        ];

        let vars: HashMap<_, _> = pairs
            .iter()
            .filter_map(|&(k, ref v)| v.clone().map(|v| (k, v)))
            .collect();

        let id_opt: Option<String> = vars.get("id").map(std::string::ToString::to_string);

        let stream =
            replace_websocket_streams_placeholders("/<pair>@indexPrice@<updateSpeed>", &vars);

        Ok(create_stream_handler::<models::IndexPriceStreamResponse>(
            WebsocketBase::WebsocketStreams(Arc::clone(&self.websocket_streams_base)),
            stream,
            id_opt.map(|s| {
                if !s.is_empty() && s.bytes().all(|b| b.is_ascii_digit()) {
                    if let Ok(n) = s.parse::<u32>() {
                        return StreamId::Number(n);
                    }
                }
                StreamId::Str(s)
            }),
            None,
        )
        .await)
    }

    async fn individual_symbol_book_ticker_streams(
        &self,
        params: IndividualSymbolBookTickerStreamsParams,
    ) -> anyhow::Result<Arc<WebsocketStream<models::IndividualSymbolBookTickerStreamsResponse>>>
    {
        let IndividualSymbolBookTickerStreamsParams { symbol, id } = params;

        let pairs: &[(&str, Option<String>)] =
            &[("symbol", Some(symbol.clone())), ("id", id.clone())];

        let vars: HashMap<_, _> = pairs
            .iter()
            .filter_map(|&(k, ref v)| v.clone().map(|v| (k, v)))
            .collect();

        let id_opt: Option<String> = vars.get("id").map(std::string::ToString::to_string);

        let stream = replace_websocket_streams_placeholders("/<symbol>@bookTicker", &vars);

        Ok(
            create_stream_handler::<models::IndividualSymbolBookTickerStreamsResponse>(
                WebsocketBase::WebsocketStreams(Arc::clone(&self.websocket_streams_base)),
                stream,
                id_opt.map(|s| {
                    if !s.is_empty() && s.bytes().all(|b| b.is_ascii_digit()) {
                        if let Ok(n) = s.parse::<u32>() {
                            return StreamId::Number(n);
                        }
                    }
                    StreamId::Str(s)
                }),
                None,
            )
            .await,
        )
    }

    async fn individual_symbol_mini_ticker_stream(
        &self,
        params: IndividualSymbolMiniTickerStreamParams,
    ) -> anyhow::Result<Arc<WebsocketStream<models::IndividualSymbolMiniTickerStreamResponse>>>
    {
        let IndividualSymbolMiniTickerStreamParams { symbol, id } = params;

        let pairs: &[(&str, Option<String>)] =
            &[("symbol", Some(symbol.clone())), ("id", id.clone())];

        let vars: HashMap<_, _> = pairs
            .iter()
            .filter_map(|&(k, ref v)| v.clone().map(|v| (k, v)))
            .collect();

        let id_opt: Option<String> = vars.get("id").map(std::string::ToString::to_string);

        let stream = replace_websocket_streams_placeholders("/<symbol>@miniTicker", &vars);

        Ok(
            create_stream_handler::<models::IndividualSymbolMiniTickerStreamResponse>(
                WebsocketBase::WebsocketStreams(Arc::clone(&self.websocket_streams_base)),
                stream,
                id_opt.map(|s| {
                    if !s.is_empty() && s.bytes().all(|b| b.is_ascii_digit()) {
                        if let Ok(n) = s.parse::<u32>() {
                            return StreamId::Number(n);
                        }
                    }
                    StreamId::Str(s)
                }),
                None,
            )
            .await,
        )
    }

    async fn individual_symbol_ticker_streams(
        &self,
        params: IndividualSymbolTickerStreamsParams,
    ) -> anyhow::Result<Arc<WebsocketStream<models::IndividualSymbolTickerStreamsResponse>>> {
        let IndividualSymbolTickerStreamsParams { symbol, id } = params;

        let pairs: &[(&str, Option<String>)] =
            &[("symbol", Some(symbol.clone())), ("id", id.clone())];

        let vars: HashMap<_, _> = pairs
            .iter()
            .filter_map(|&(k, ref v)| v.clone().map(|v| (k, v)))
            .collect();

        let id_opt: Option<String> = vars.get("id").map(std::string::ToString::to_string);

        let stream = replace_websocket_streams_placeholders("/<symbol>@ticker", &vars);

        Ok(
            create_stream_handler::<models::IndividualSymbolTickerStreamsResponse>(
                WebsocketBase::WebsocketStreams(Arc::clone(&self.websocket_streams_base)),
                stream,
                id_opt.map(|s| {
                    if !s.is_empty() && s.bytes().all(|b| b.is_ascii_digit()) {
                        if let Ok(n) = s.parse::<u32>() {
                            return StreamId::Number(n);
                        }
                    }
                    StreamId::Str(s)
                }),
                None,
            )
            .await,
        )
    }

    async fn kline_candlestick_streams(
        &self,
        params: KlineCandlestickStreamsParams,
    ) -> anyhow::Result<Arc<WebsocketStream<models::KlineCandlestickStreamsResponse>>> {
        let KlineCandlestickStreamsParams {
            symbol,
            interval,
            id,
        } = params;

        let pairs: &[(&str, Option<String>)] = &[
            ("symbol", Some(symbol.clone())),
            ("interval", Some(interval.clone())),
            ("id", id.clone()),
        ];

        let vars: HashMap<_, _> = pairs
            .iter()
            .filter_map(|&(k, ref v)| v.clone().map(|v| (k, v)))
            .collect();

        let id_opt: Option<String> = vars.get("id").map(std::string::ToString::to_string);

        let stream = replace_websocket_streams_placeholders("/<symbol>@kline_<interval>", &vars);

        Ok(
            create_stream_handler::<models::KlineCandlestickStreamsResponse>(
                WebsocketBase::WebsocketStreams(Arc::clone(&self.websocket_streams_base)),
                stream,
                id_opt.map(|s| {
                    if !s.is_empty() && s.bytes().all(|b| b.is_ascii_digit()) {
                        if let Ok(n) = s.parse::<u32>() {
                            return StreamId::Number(n);
                        }
                    }
                    StreamId::Str(s)
                }),
                None,
            )
            .await,
        )
    }

    async fn liquidation_order_streams(
        &self,
        params: LiquidationOrderStreamsParams,
    ) -> anyhow::Result<Arc<WebsocketStream<models::LiquidationOrderStreamsResponse>>> {
        let LiquidationOrderStreamsParams { symbol, id } = params;

        let pairs: &[(&str, Option<String>)] =
            &[("symbol", Some(symbol.clone())), ("id", id.clone())];

        let vars: HashMap<_, _> = pairs
            .iter()
            .filter_map(|&(k, ref v)| v.clone().map(|v| (k, v)))
            .collect();

        let id_opt: Option<String> = vars.get("id").map(std::string::ToString::to_string);

        let stream = replace_websocket_streams_placeholders("/<symbol>@forceOrder", &vars);

        Ok(
            create_stream_handler::<models::LiquidationOrderStreamsResponse>(
                WebsocketBase::WebsocketStreams(Arc::clone(&self.websocket_streams_base)),
                stream,
                id_opt.map(|s| {
                    if !s.is_empty() && s.bytes().all(|b| b.is_ascii_digit()) {
                        if let Ok(n) = s.parse::<u32>() {
                            return StreamId::Number(n);
                        }
                    }
                    StreamId::Str(s)
                }),
                None,
            )
            .await,
        )
    }

    async fn mark_price_kline_candlestick_streams(
        &self,
        params: MarkPriceKlineCandlestickStreamsParams,
    ) -> anyhow::Result<Arc<WebsocketStream<models::MarkPriceKlineCandlestickStreamsResponse>>>
    {
        let MarkPriceKlineCandlestickStreamsParams {
            symbol,
            interval,
            id,
        } = params;

        let pairs: &[(&str, Option<String>)] = &[
            ("symbol", Some(symbol.clone())),
            ("interval", Some(interval.clone())),
            ("id", id.clone()),
        ];

        let vars: HashMap<_, _> = pairs
            .iter()
            .filter_map(|&(k, ref v)| v.clone().map(|v| (k, v)))
            .collect();

        let id_opt: Option<String> = vars.get("id").map(std::string::ToString::to_string);

        let stream =
            replace_websocket_streams_placeholders("/<symbol>@markPriceKline_<interval>", &vars);

        Ok(
            create_stream_handler::<models::MarkPriceKlineCandlestickStreamsResponse>(
                WebsocketBase::WebsocketStreams(Arc::clone(&self.websocket_streams_base)),
                stream,
                id_opt.map(|s| {
                    if !s.is_empty() && s.bytes().all(|b| b.is_ascii_digit()) {
                        if let Ok(n) = s.parse::<u32>() {
                            return StreamId::Number(n);
                        }
                    }
                    StreamId::Str(s)
                }),
                None,
            )
            .await,
        )
    }

    async fn mark_price_of_all_symbols_of_a_pair(
        &self,
        params: MarkPriceOfAllSymbolsOfAPairParams,
    ) -> anyhow::Result<Arc<WebsocketStream<Vec<models::MarkPriceOfAllSymbolsOfAPairResponseInner>>>>
    {
        let MarkPriceOfAllSymbolsOfAPairParams {
            pair,
            id,
            update_speed,
        } = params;

        let pairs: &[(&str, Option<String>)] = &[
            ("pair", Some(pair.clone())),
            ("id", id.clone()),
            ("updateSpeed", update_speed.clone()),
        ];

        let vars: HashMap<_, _> = pairs
            .iter()
            .filter_map(|&(k, ref v)| v.clone().map(|v| (k, v)))
            .collect();

        let id_opt: Option<String> = vars.get("id").map(std::string::ToString::to_string);

        let stream =
            replace_websocket_streams_placeholders("/<pair>@markPrice@<updateSpeed>", &vars);

        Ok(
            create_stream_handler::<Vec<models::MarkPriceOfAllSymbolsOfAPairResponseInner>>(
                WebsocketBase::WebsocketStreams(Arc::clone(&self.websocket_streams_base)),
                stream,
                id_opt.map(|s| {
                    if !s.is_empty() && s.bytes().all(|b| b.is_ascii_digit()) {
                        if let Ok(n) = s.parse::<u32>() {
                            return StreamId::Number(n);
                        }
                    }
                    StreamId::Str(s)
                }),
                None,
            )
            .await,
        )
    }

    async fn mark_price_stream(
        &self,
        params: MarkPriceStreamParams,
    ) -> anyhow::Result<Arc<WebsocketStream<models::MarkPriceStreamResponse>>> {
        let MarkPriceStreamParams {
            symbol,
            id,
            update_speed,
        } = params;

        let pairs: &[(&str, Option<String>)] = &[
            ("symbol", Some(symbol.clone())),
            ("id", id.clone()),
            ("updateSpeed", update_speed.clone()),
        ];

        let vars: HashMap<_, _> = pairs
            .iter()
            .filter_map(|&(k, ref v)| v.clone().map(|v| (k, v)))
            .collect();

        let id_opt: Option<String> = vars.get("id").map(std::string::ToString::to_string);

        let stream =
            replace_websocket_streams_placeholders("/<symbol>@markPrice@<updateSpeed>", &vars);

        Ok(create_stream_handler::<models::MarkPriceStreamResponse>(
            WebsocketBase::WebsocketStreams(Arc::clone(&self.websocket_streams_base)),
            stream,
            id_opt.map(|s| {
                if !s.is_empty() && s.bytes().all(|b| b.is_ascii_digit()) {
                    if let Ok(n) = s.parse::<u32>() {
                        return StreamId::Number(n);
                    }
                }
                StreamId::Str(s)
            }),
            None,
        )
        .await)
    }

    async fn partial_book_depth_streams(
        &self,
        params: PartialBookDepthStreamsParams,
    ) -> anyhow::Result<Arc<WebsocketStream<models::PartialBookDepthStreamsResponse>>> {
        let PartialBookDepthStreamsParams {
            symbol,
            levels,
            id,
            update_speed,
        } = params;

        let pairs: &[(&str, Option<String>)] = &[
            ("symbol", Some(symbol.clone())),
            ("levels", Some(levels.to_string())),
            ("id", id.clone()),
            ("updateSpeed", update_speed.clone()),
        ];

        let vars: HashMap<_, _> = pairs
            .iter()
            .filter_map(|&(k, ref v)| v.clone().map(|v| (k, v)))
            .collect();

        let id_opt: Option<String> = vars.get("id").map(std::string::ToString::to_string);

        let stream =
            replace_websocket_streams_placeholders("/<symbol>@depth<levels>@<updateSpeed>", &vars);

        Ok(
            create_stream_handler::<models::PartialBookDepthStreamsResponse>(
                WebsocketBase::WebsocketStreams(Arc::clone(&self.websocket_streams_base)),
                stream,
                id_opt.map(|s| {
                    if !s.is_empty() && s.bytes().all(|b| b.is_ascii_digit()) {
                        if let Ok(n) = s.parse::<u32>() {
                            return StreamId::Number(n);
                        }
                    }
                    StreamId::Str(s)
                }),
                None,
            )
            .await,
        )
    }
}

#[cfg(all(test, feature = "derivatives_trading_coin_futures"))]
mod tests {
    use super::*;
    use crate::TOKIO_SHARED_RT;
    use crate::{
        common::websocket::{WebsocketConnection, WebsocketHandler},
        config::ConfigurationWebsocketStreams,
    };
    use serde_json::json;
    use std::sync::atomic::{AtomicBool, Ordering};
    use tokio::task::yield_now;

    async fn make_streams_base() -> (Arc<WebsocketStreams>, Arc<WebsocketConnection>) {
        let conn = WebsocketConnection::new("test");
        let config = ConfigurationWebsocketStreams::builder()
            .build()
            .expect("Failed to build configuration");
        let streams_base = WebsocketStreams::new(config, vec![conn.clone()], vec![]);
        conn.set_handler(streams_base.clone() as Arc<dyn WebsocketHandler>)
            .await;
        (streams_base, conn)
    }

    #[test]
    fn aggregate_trade_streams_should_execute_successfully() {
        TOKIO_SHARED_RT.block_on(async {
            let (streams_base, _) = make_streams_base().await;
            let api = WebsocketMarketStreamsApiClient::new(streams_base.clone());

            let id = "test-id-123".to_string();

            let params = AggregateTradeStreamsParams::builder("btcusdt".to_string())
                .id(Some(id.clone()))
                .build()
                .unwrap();

            let AggregateTradeStreamsParams { symbol, id } = params.clone();

            let pairs: &[(&str, Option<String>)] =
                &[("symbol", Some(symbol.clone())), ("id", id.clone())];

            let vars: HashMap<_, _> = pairs
                .iter()
                .filter_map(|&(k, ref v)| v.clone().map(|v| (k, v)))
                .collect();
            let stream = replace_websocket_streams_placeholders("/<symbol>@aggTrade", &vars);
            let ws_stream = api
                .aggregate_trade_streams(params)
                .await
                .expect("aggregate_trade_streams should return a WebsocketStream");

            assert!(
                streams_base.is_subscribed(&stream).await,
                "expected stream '{stream}' to be subscribed"
            );
            assert_eq!(ws_stream.id, Some(StreamId::Str("test-id-123".to_string())));
        });
    }

    #[test]
    fn aggregate_trade_streams_should_handle_incoming_message() {
        TOKIO_SHARED_RT.block_on(async {
            let (streams_base, conn) = make_streams_base().await;
            let api = WebsocketMarketStreamsApiClient::new(streams_base.clone());

            let id = "test-id-123".to_string();

            let params = AggregateTradeStreamsParams::builder("btcusdt".to_string(),).id(Some(id.clone())).build().unwrap();

            let AggregateTradeStreamsParams {
                symbol,id,
            } = params.clone();

            let pairs: &[(&str, Option<String>)] = &[
                ("symbol",
                        Some(symbol.clone())
                ),
                ("id",
                        id.clone()
                ),
            ];

            let vars: HashMap<_, _> = pairs
                .iter()
                .filter_map(|&(k, ref v)| v.clone().map(|v| (k, v)))
                .collect();
            let stream = replace_websocket_streams_placeholders("/<symbol>@aggTrade", &vars);

            let ws_stream = api.aggregate_trade_streams(params).await.unwrap();

            let called = Arc::new(AtomicBool::new(false));
            let called_with_message = called.clone();
            ws_stream.on_message(move |_payload: models::AggregateTradeStreamsResponse| {
                called_with_message.store(true, Ordering::SeqCst);
            });

            let payload: Value = serde_json::from_str(r#"{"e":"aggTrade","E":1591261134288,"a":424951,"s":"BTCUSD_200626","p":"9643.5","q":"2","f":606073,"l":606073,"T":1591261134199,"m":false}"#).unwrap();
            let msg = json!({
                "stream": stream,
                "data": payload,
            });

            streams_base.on_message(msg.to_string(), conn.clone()).await;
            yield_now().await;

            assert!(called.load(Ordering::SeqCst), "expected our callback to have been invoked");
        });
    }

    #[test]
    fn aggregate_trade_streams_should_not_fire_after_unsubscribe() {
        TOKIO_SHARED_RT.block_on(async {
            let (streams_base, conn) = make_streams_base().await;
            let api = WebsocketMarketStreamsApiClient::new(streams_base.clone());

            let id = "test-id-123".to_string();

            let params = AggregateTradeStreamsParams::builder("btcusdt".to_string(),).id(Some(id.clone())).build().unwrap();

            let AggregateTradeStreamsParams {
                symbol,id,
            } = params.clone();

            let pairs: &[(&str, Option<String>)] = &[
                ("symbol",
                        Some(symbol.clone())
                ),
                ("id",
                        id.clone()
                ),
            ];

            let vars: HashMap<_, _> = pairs
                .iter()
                .filter_map(|&(k, ref v)| v.clone().map(|v| (k, v)))
                .collect();
            let stream = replace_websocket_streams_placeholders("/<symbol>@aggTrade", &vars);

            let ws_stream = api.aggregate_trade_streams(params).await.unwrap();

            let called = Arc::new(AtomicBool::new(false));
            let called_clone = called.clone();
            ws_stream.on_message(move |_payload: models::AggregateTradeStreamsResponse| {
                called_clone.store(true, Ordering::SeqCst);
            });

            assert!(streams_base.is_subscribed(&stream).await, "should be subscribed before unsubscribe");

            ws_stream.unsubscribe().await;

            let payload: Value = serde_json::from_str(r#"{"e":"aggTrade","E":1591261134288,"a":424951,"s":"BTCUSD_200626","p":"9643.5","q":"2","f":606073,"l":606073,"T":1591261134199,"m":false}"#).unwrap();
            let msg = json!({
                "stream": stream,
                "data": payload,
            });

            streams_base.on_message(msg.to_string(), conn.clone()).await;

            yield_now().await;

            assert!(!called.load(Ordering::SeqCst), "callback should not be invoked after unsubscribe");
        });
    }

    #[test]
    fn all_book_tickers_stream_should_execute_successfully() {
        TOKIO_SHARED_RT.block_on(async {
            let (streams_base, _) = make_streams_base().await;
            let api = WebsocketMarketStreamsApiClient::new(streams_base.clone());

            let id = "test-id-123".to_string();

            let params = AllBookTickersStreamParams::builder()
                .id(Some(id.clone()))
                .build()
                .unwrap();

            let AllBookTickersStreamParams { id } = params.clone();

            let pairs: &[(&str, Option<String>)] = &[("id", id.clone())];

            let vars: HashMap<_, _> = pairs
                .iter()
                .filter_map(|&(k, ref v)| v.clone().map(|v| (k, v)))
                .collect();
            let stream = replace_websocket_streams_placeholders("/!bookTicker", &vars);
            let ws_stream = api
                .all_book_tickers_stream(params)
                .await
                .expect("all_book_tickers_stream should return a WebsocketStream");

            assert!(
                streams_base.is_subscribed(&stream).await,
                "expected stream '{stream}' to be subscribed"
            );
            assert_eq!(ws_stream.id, Some(StreamId::Str("test-id-123".to_string())));
        });
    }

    #[test]
    fn all_book_tickers_stream_should_handle_incoming_message() {
        TOKIO_SHARED_RT.block_on(async {
            let (streams_base, conn) = make_streams_base().await;
            let api = WebsocketMarketStreamsApiClient::new(streams_base.clone());

            let id = "test-id-123".to_string();

            let params = AllBookTickersStreamParams::builder().id(Some(id.clone())).build().unwrap();

            let AllBookTickersStreamParams {
                id,
            } = params.clone();

            let pairs: &[(&str, Option<String>)] = &[
                ("id",
                        id.clone()
                ),
            ];

            let vars: HashMap<_, _> = pairs
                .iter()
                .filter_map(|&(k, ref v)| v.clone().map(|v| (k, v)))
                .collect();
            let stream = replace_websocket_streams_placeholders("/!bookTicker", &vars);

            let ws_stream = api.all_book_tickers_stream(params).await.unwrap();

            let called = Arc::new(AtomicBool::new(false));
            let called_with_message = called.clone();
            ws_stream.on_message(move |_payload: models::AllBookTickersStreamResponse| {
                called_with_message.store(true, Ordering::SeqCst);
            });

            let payload: Value = serde_json::from_str(r#"{"e":"bookTicker","u":17242169,"s":"BTCUSD_200626","ps":"BTCUSD","b":"9548.1","B":"52","a":"9548.5","A":"11","T":1591268628155,"E":1591268628166}"#).unwrap();
            let msg = json!({
                "stream": stream,
                "data": payload,
            });

            streams_base.on_message(msg.to_string(), conn.clone()).await;
            yield_now().await;

            assert!(called.load(Ordering::SeqCst), "expected our callback to have been invoked");
        });
    }

    #[test]
    fn all_book_tickers_stream_should_not_fire_after_unsubscribe() {
        TOKIO_SHARED_RT.block_on(async {
            let (streams_base, conn) = make_streams_base().await;
            let api = WebsocketMarketStreamsApiClient::new(streams_base.clone());

            let id = "test-id-123".to_string();

            let params = AllBookTickersStreamParams::builder().id(Some(id.clone())).build().unwrap();

            let AllBookTickersStreamParams {
                id,
            } = params.clone();

            let pairs: &[(&str, Option<String>)] = &[
                ("id",
                        id.clone()
                ),
            ];

            let vars: HashMap<_, _> = pairs
                .iter()
                .filter_map(|&(k, ref v)| v.clone().map(|v| (k, v)))
                .collect();
            let stream = replace_websocket_streams_placeholders("/!bookTicker", &vars);

            let ws_stream = api.all_book_tickers_stream(params).await.unwrap();

            let called = Arc::new(AtomicBool::new(false));
            let called_clone = called.clone();
            ws_stream.on_message(move |_payload: models::AllBookTickersStreamResponse| {
                called_clone.store(true, Ordering::SeqCst);
            });

            assert!(streams_base.is_subscribed(&stream).await, "should be subscribed before unsubscribe");

            ws_stream.unsubscribe().await;

            let payload: Value = serde_json::from_str(r#"{"e":"bookTicker","u":17242169,"s":"BTCUSD_200626","ps":"BTCUSD","b":"9548.1","B":"52","a":"9548.5","A":"11","T":1591268628155,"E":1591268628166}"#).unwrap();
            let msg = json!({
                "stream": stream,
                "data": payload,
            });

            streams_base.on_message(msg.to_string(), conn.clone()).await;

            yield_now().await;

            assert!(!called.load(Ordering::SeqCst), "callback should not be invoked after unsubscribe");
        });
    }

    #[test]
    fn all_market_liquidation_order_streams_should_execute_successfully() {
        TOKIO_SHARED_RT.block_on(async {
            let (streams_base, _) = make_streams_base().await;
            let api = WebsocketMarketStreamsApiClient::new(streams_base.clone());

            let id = "test-id-123".to_string();

            let params = AllMarketLiquidationOrderStreamsParams::builder()
                .id(Some(id.clone()))
                .build()
                .unwrap();

            let AllMarketLiquidationOrderStreamsParams { id } = params.clone();

            let pairs: &[(&str, Option<String>)] = &[("id", id.clone())];

            let vars: HashMap<_, _> = pairs
                .iter()
                .filter_map(|&(k, ref v)| v.clone().map(|v| (k, v)))
                .collect();
            let stream = replace_websocket_streams_placeholders("/!forceOrder@arr", &vars);
            let ws_stream = api
                .all_market_liquidation_order_streams(params)
                .await
                .expect("all_market_liquidation_order_streams should return a WebsocketStream");

            assert!(
                streams_base.is_subscribed(&stream).await,
                "expected stream '{stream}' to be subscribed"
            );
            assert_eq!(ws_stream.id, Some(StreamId::Str("test-id-123".to_string())));
        });
    }

    #[test]
    fn all_market_liquidation_order_streams_should_handle_incoming_message() {
        TOKIO_SHARED_RT.block_on(async {
            let (streams_base, conn) = make_streams_base().await;
            let api = WebsocketMarketStreamsApiClient::new(streams_base.clone());

            let id = "test-id-123".to_string();

            let params = AllMarketLiquidationOrderStreamsParams::builder().id(Some(id.clone())).build().unwrap();

            let AllMarketLiquidationOrderStreamsParams {
                id,
            } = params.clone();

            let pairs: &[(&str, Option<String>)] = &[
                ("id",
                        id.clone()
                ),
            ];

            let vars: HashMap<_, _> = pairs
                .iter()
                .filter_map(|&(k, ref v)| v.clone().map(|v| (k, v)))
                .collect();
            let stream = replace_websocket_streams_placeholders("/!forceOrder@arr", &vars);

            let ws_stream = api.all_market_liquidation_order_streams(params).await.unwrap();

            let called = Arc::new(AtomicBool::new(false));
            let called_with_message = called.clone();
            ws_stream.on_message(move |_payload: models::AllMarketLiquidationOrderStreamsResponse| {
                called_with_message.store(true, Ordering::SeqCst);
            });

            let payload: Value = serde_json::from_str(r#"{"e":"forceOrder","E":1591154240950,"o":{"s":"BTCUSD_200925","ps":"BTCUSD","S":"SELL","o":"LIMIT","f":"IOC","q":"1","p":"9425.5","ap":"9496.5","X":"FILLED","l":"1","z":"1","T":1591154240949}}"#).unwrap();
            let msg = json!({
                "stream": stream,
                "data": payload,
            });

            streams_base.on_message(msg.to_string(), conn.clone()).await;
            yield_now().await;

            assert!(called.load(Ordering::SeqCst), "expected our callback to have been invoked");
        });
    }

    #[test]
    fn all_market_liquidation_order_streams_should_not_fire_after_unsubscribe() {
        TOKIO_SHARED_RT.block_on(async {
            let (streams_base, conn) = make_streams_base().await;
            let api = WebsocketMarketStreamsApiClient::new(streams_base.clone());

            let id = "test-id-123".to_string();

            let params = AllMarketLiquidationOrderStreamsParams::builder().id(Some(id.clone())).build().unwrap();

            let AllMarketLiquidationOrderStreamsParams {
                id,
            } = params.clone();

            let pairs: &[(&str, Option<String>)] = &[
                ("id",
                        id.clone()
                ),
            ];

            let vars: HashMap<_, _> = pairs
                .iter()
                .filter_map(|&(k, ref v)| v.clone().map(|v| (k, v)))
                .collect();
            let stream = replace_websocket_streams_placeholders("/!forceOrder@arr", &vars);

            let ws_stream = api.all_market_liquidation_order_streams(params).await.unwrap();

            let called = Arc::new(AtomicBool::new(false));
            let called_clone = called.clone();
            ws_stream.on_message(move |_payload: models::AllMarketLiquidationOrderStreamsResponse| {
                called_clone.store(true, Ordering::SeqCst);
            });

            assert!(streams_base.is_subscribed(&stream).await, "should be subscribed before unsubscribe");

            ws_stream.unsubscribe().await;

            let payload: Value = serde_json::from_str(r#"{"e":"forceOrder","E":1591154240950,"o":{"s":"BTCUSD_200925","ps":"BTCUSD","S":"SELL","o":"LIMIT","f":"IOC","q":"1","p":"9425.5","ap":"9496.5","X":"FILLED","l":"1","z":"1","T":1591154240949}}"#).unwrap();
            let msg = json!({
                "stream": stream,
                "data": payload,
            });

            streams_base.on_message(msg.to_string(), conn.clone()).await;

            yield_now().await;

            assert!(!called.load(Ordering::SeqCst), "callback should not be invoked after unsubscribe");
        });
    }

    #[test]
    fn all_market_mini_tickers_stream_should_execute_successfully() {
        TOKIO_SHARED_RT.block_on(async {
            let (streams_base, _) = make_streams_base().await;
            let api = WebsocketMarketStreamsApiClient::new(streams_base.clone());

            let id = "test-id-123".to_string();

            let params = AllMarketMiniTickersStreamParams::builder()
                .id(Some(id.clone()))
                .build()
                .unwrap();

            let AllMarketMiniTickersStreamParams { id } = params.clone();

            let pairs: &[(&str, Option<String>)] = &[("id", id.clone())];

            let vars: HashMap<_, _> = pairs
                .iter()
                .filter_map(|&(k, ref v)| v.clone().map(|v| (k, v)))
                .collect();
            let stream = replace_websocket_streams_placeholders("/!miniTicker@arr", &vars);
            let ws_stream = api
                .all_market_mini_tickers_stream(params)
                .await
                .expect("all_market_mini_tickers_stream should return a WebsocketStream");

            assert!(
                streams_base.is_subscribed(&stream).await,
                "expected stream '{stream}' to be subscribed"
            );
            assert_eq!(ws_stream.id, Some(StreamId::Str("test-id-123".to_string())));
        });
    }

    #[test]
    fn all_market_mini_tickers_stream_should_handle_incoming_message() {
        TOKIO_SHARED_RT.block_on(async {
            let (streams_base, conn) = make_streams_base().await;
            let api = WebsocketMarketStreamsApiClient::new(streams_base.clone());

            let id = "test-id-123".to_string();

            let params = AllMarketMiniTickersStreamParams::builder().id(Some(id.clone())).build().unwrap();

            let AllMarketMiniTickersStreamParams {
                id,
            } = params.clone();

            let pairs: &[(&str, Option<String>)] = &[
                ("id",
                        id.clone()
                ),
            ];

            let vars: HashMap<_, _> = pairs
                .iter()
                .filter_map(|&(k, ref v)| v.clone().map(|v| (k, v)))
                .collect();
            let stream = replace_websocket_streams_placeholders("/!miniTicker@arr", &vars);

            let ws_stream = api.all_market_mini_tickers_stream(params).await.unwrap();

            let called = Arc::new(AtomicBool::new(false));
            let called_with_message = called.clone();
            ws_stream.on_message(move |_payload: Vec<models::AllMarketMiniTickersStreamResponseInner>| {
                called_with_message.store(true, Ordering::SeqCst);
            });

            let payload: Value = serde_json::from_str(r#"[{"e":"24hrMiniTicker","E":1591267704450,"s":"BTCUSD_200626","ps":"BTCUSD","c":"9561.7","o":"9580.9","h":"10000.0","l":"7000.0","v":"487476","q":"33264343847.22378500"}]"#).unwrap();
            let msg = json!({
                "stream": stream,
                "data": payload,
            });

            streams_base.on_message(msg.to_string(), conn.clone()).await;
            yield_now().await;

            assert!(called.load(Ordering::SeqCst), "expected our callback to have been invoked");
        });
    }

    #[test]
    fn all_market_mini_tickers_stream_should_not_fire_after_unsubscribe() {
        TOKIO_SHARED_RT.block_on(async {
            let (streams_base, conn) = make_streams_base().await;
            let api = WebsocketMarketStreamsApiClient::new(streams_base.clone());

            let id = "test-id-123".to_string();

            let params = AllMarketMiniTickersStreamParams::builder().id(Some(id.clone())).build().unwrap();

            let AllMarketMiniTickersStreamParams {
                id,
            } = params.clone();

            let pairs: &[(&str, Option<String>)] = &[
                ("id",
                        id.clone()
                ),
            ];

            let vars: HashMap<_, _> = pairs
                .iter()
                .filter_map(|&(k, ref v)| v.clone().map(|v| (k, v)))
                .collect();
            let stream = replace_websocket_streams_placeholders("/!miniTicker@arr", &vars);

            let ws_stream = api.all_market_mini_tickers_stream(params).await.unwrap();

            let called = Arc::new(AtomicBool::new(false));
            let called_clone = called.clone();
            ws_stream.on_message(move |_payload: Vec<models::AllMarketMiniTickersStreamResponseInner>| {
                called_clone.store(true, Ordering::SeqCst);
            });

            assert!(streams_base.is_subscribed(&stream).await, "should be subscribed before unsubscribe");

            ws_stream.unsubscribe().await;

            let payload: Value = serde_json::from_str(r#"[{"e":"24hrMiniTicker","E":1591267704450,"s":"BTCUSD_200626","ps":"BTCUSD","c":"9561.7","o":"9580.9","h":"10000.0","l":"7000.0","v":"487476","q":"33264343847.22378500"}]"#).unwrap();
            let msg = json!({
                "stream": stream,
                "data": payload,
            });

            streams_base.on_message(msg.to_string(), conn.clone()).await;

            yield_now().await;

            assert!(!called.load(Ordering::SeqCst), "callback should not be invoked after unsubscribe");
        });
    }

    #[test]
    fn all_market_tickers_streams_should_execute_successfully() {
        TOKIO_SHARED_RT.block_on(async {
            let (streams_base, _) = make_streams_base().await;
            let api = WebsocketMarketStreamsApiClient::new(streams_base.clone());

            let id = "test-id-123".to_string();

            let params = AllMarketTickersStreamsParams::builder()
                .id(Some(id.clone()))
                .build()
                .unwrap();

            let AllMarketTickersStreamsParams { id } = params.clone();

            let pairs: &[(&str, Option<String>)] = &[("id", id.clone())];

            let vars: HashMap<_, _> = pairs
                .iter()
                .filter_map(|&(k, ref v)| v.clone().map(|v| (k, v)))
                .collect();
            let stream = replace_websocket_streams_placeholders("/!ticker@arr", &vars);
            let ws_stream = api
                .all_market_tickers_streams(params)
                .await
                .expect("all_market_tickers_streams should return a WebsocketStream");

            assert!(
                streams_base.is_subscribed(&stream).await,
                "expected stream '{stream}' to be subscribed"
            );
            assert_eq!(ws_stream.id, Some(StreamId::Str("test-id-123".to_string())));
        });
    }

    #[test]
    fn all_market_tickers_streams_should_handle_incoming_message() {
        TOKIO_SHARED_RT.block_on(async {
            let (streams_base, conn) = make_streams_base().await;
            let api = WebsocketMarketStreamsApiClient::new(streams_base.clone());

            let id = "test-id-123".to_string();

            let params = AllMarketTickersStreamsParams::builder().id(Some(id.clone())).build().unwrap();

            let AllMarketTickersStreamsParams {
                id,
            } = params.clone();

            let pairs: &[(&str, Option<String>)] = &[
                ("id",
                        id.clone()
                ),
            ];

            let vars: HashMap<_, _> = pairs
                .iter()
                .filter_map(|&(k, ref v)| v.clone().map(|v| (k, v)))
                .collect();
            let stream = replace_websocket_streams_placeholders("/!ticker@arr", &vars);

            let ws_stream = api.all_market_tickers_streams(params).await.unwrap();

            let called = Arc::new(AtomicBool::new(false));
            let called_with_message = called.clone();
            ws_stream.on_message(move |_payload: Vec<models::AllMarketTickersStreamsResponseInner>| {
                called_with_message.store(true, Ordering::SeqCst);
            });

            let payload: Value = serde_json::from_str(r#"[{"e":"24hrTicker","E":1591268262453,"s":"BTCUSD_200626","ps":"BTCUSD","p":"-43.4","P":"-0.452","w":"0.00147974","c":"9548.5","Q":"2","o":"9591.9","h":"10000.0","l":"7000.0","v":"487850","q":"32968676323.46222700","O":1591181820000,"C":1591268262442,"F":512014,"L":615289,"n":103272}]"#).unwrap();
            let msg = json!({
                "stream": stream,
                "data": payload,
            });

            streams_base.on_message(msg.to_string(), conn.clone()).await;
            yield_now().await;

            assert!(called.load(Ordering::SeqCst), "expected our callback to have been invoked");
        });
    }

    #[test]
    fn all_market_tickers_streams_should_not_fire_after_unsubscribe() {
        TOKIO_SHARED_RT.block_on(async {
            let (streams_base, conn) = make_streams_base().await;
            let api = WebsocketMarketStreamsApiClient::new(streams_base.clone());

            let id = "test-id-123".to_string();

            let params = AllMarketTickersStreamsParams::builder().id(Some(id.clone())).build().unwrap();

            let AllMarketTickersStreamsParams {
                id,
            } = params.clone();

            let pairs: &[(&str, Option<String>)] = &[
                ("id",
                        id.clone()
                ),
            ];

            let vars: HashMap<_, _> = pairs
                .iter()
                .filter_map(|&(k, ref v)| v.clone().map(|v| (k, v)))
                .collect();
            let stream = replace_websocket_streams_placeholders("/!ticker@arr", &vars);

            let ws_stream = api.all_market_tickers_streams(params).await.unwrap();

            let called = Arc::new(AtomicBool::new(false));
            let called_clone = called.clone();
            ws_stream.on_message(move |_payload: Vec<models::AllMarketTickersStreamsResponseInner>| {
                called_clone.store(true, Ordering::SeqCst);
            });

            assert!(streams_base.is_subscribed(&stream).await, "should be subscribed before unsubscribe");

            ws_stream.unsubscribe().await;

            let payload: Value = serde_json::from_str(r#"[{"e":"24hrTicker","E":1591268262453,"s":"BTCUSD_200626","ps":"BTCUSD","p":"-43.4","P":"-0.452","w":"0.00147974","c":"9548.5","Q":"2","o":"9591.9","h":"10000.0","l":"7000.0","v":"487850","q":"32968676323.46222700","O":1591181820000,"C":1591268262442,"F":512014,"L":615289,"n":103272}]"#).unwrap();
            let msg = json!({
                "stream": stream,
                "data": payload,
            });

            streams_base.on_message(msg.to_string(), conn.clone()).await;

            yield_now().await;

            assert!(!called.load(Ordering::SeqCst), "callback should not be invoked after unsubscribe");
        });
    }

    #[test]
    fn continuous_contract_kline_candlestick_streams_should_execute_successfully() {
        TOKIO_SHARED_RT.block_on(async {
            let (streams_base, _) = make_streams_base().await;
            let api = WebsocketMarketStreamsApiClient::new(streams_base.clone());

            let id = "test-id-123".to_string();

            let params = ContinuousContractKlineCandlestickStreamsParams::builder(
                "btcusdt".to_string(),
                "next_quarter".to_string(),
                "1m".to_string(),
            )
            .id(Some(id.clone()))
            .build()
            .unwrap();

            let ContinuousContractKlineCandlestickStreamsParams {
                pair,
                contract_type,
                interval,
                id,
            } = params.clone();

            let pairs: &[(&str, Option<String>)] = &[
                ("pair", Some(pair.clone())),
                ("contractType", Some(contract_type.clone())),
                ("interval", Some(interval.clone())),
                ("id", id.clone()),
            ];

            let vars: HashMap<_, _> = pairs
                .iter()
                .filter_map(|&(k, ref v)| v.clone().map(|v| (k, v)))
                .collect();
            let stream = replace_websocket_streams_placeholders(
                "/<pair>_<contractType>@continuousKline_<interval>",
                &vars,
            );
            let ws_stream = api
                .continuous_contract_kline_candlestick_streams(params)
                .await
                .expect(
                    "continuous_contract_kline_candlestick_streams should return a WebsocketStream",
                );

            assert!(
                streams_base.is_subscribed(&stream).await,
                "expected stream '{stream}' to be subscribed"
            );
            assert_eq!(ws_stream.id, Some(StreamId::Str("test-id-123".to_string())));
        });
    }

    #[test]
    fn continuous_contract_kline_candlestick_streams_should_handle_incoming_message() {
        TOKIO_SHARED_RT.block_on(async {
            let (streams_base, conn) = make_streams_base().await;
            let api = WebsocketMarketStreamsApiClient::new(streams_base.clone());

            let id = "test-id-123".to_string();

            let params = ContinuousContractKlineCandlestickStreamsParams::builder("btcusdt".to_string(),"next_quarter".to_string(),"1m".to_string(),).id(Some(id.clone())).build().unwrap();

            let ContinuousContractKlineCandlestickStreamsParams {
                pair,contract_type,interval,id,
            } = params.clone();

            let pairs: &[(&str, Option<String>)] = &[
                ("pair",
                        Some(pair.clone())
                ),
                ("contractType",
                        Some(contract_type.clone())
                ),
                ("interval",
                        Some(interval.clone())
                ),
                ("id",
                        id.clone()
                ),
            ];

            let vars: HashMap<_, _> = pairs
                .iter()
                .filter_map(|&(k, ref v)| v.clone().map(|v| (k, v)))
                .collect();
            let stream = replace_websocket_streams_placeholders("/<pair>_<contractType>@continuousKline_<interval>", &vars);

            let ws_stream = api.continuous_contract_kline_candlestick_streams(params).await.unwrap();

            let called = Arc::new(AtomicBool::new(false));
            let called_with_message = called.clone();
            ws_stream.on_message(move |_payload: models::ContinuousContractKlineCandlestickStreamsResponse| {
                called_with_message.store(true, Ordering::SeqCst);
            });

            let payload: Value = serde_json::from_str(r#"{"e":"continuous_kline","E":1591261542539,"ps":"BTCUSD","ct":"NEXT_QUARTER","k":{"t":1591261500000,"T":1591261559999,"i":"1m","f":606400,"L":606430,"o":"9638.9","c":"9639.8","h":"9639.8","l":"9638.6","v":"156","n":31,"x":false,"q":"1.61836886","V":"73","Q":"0.75731156","B":"0"}}"#).unwrap();
            let msg = json!({
                "stream": stream,
                "data": payload,
            });

            streams_base.on_message(msg.to_string(), conn.clone()).await;
            yield_now().await;

            assert!(called.load(Ordering::SeqCst), "expected our callback to have been invoked");
        });
    }

    #[test]
    fn continuous_contract_kline_candlestick_streams_should_not_fire_after_unsubscribe() {
        TOKIO_SHARED_RT.block_on(async {
            let (streams_base, conn) = make_streams_base().await;
            let api = WebsocketMarketStreamsApiClient::new(streams_base.clone());

            let id = "test-id-123".to_string();

            let params = ContinuousContractKlineCandlestickStreamsParams::builder("btcusdt".to_string(),"next_quarter".to_string(),"1m".to_string(),).id(Some(id.clone())).build().unwrap();

            let ContinuousContractKlineCandlestickStreamsParams {
                pair,contract_type,interval,id,
            } = params.clone();

            let pairs: &[(&str, Option<String>)] = &[
                ("pair",
                        Some(pair.clone())
                ),
                ("contractType",
                        Some(contract_type.clone())
                ),
                ("interval",
                        Some(interval.clone())
                ),
                ("id",
                        id.clone()
                ),
            ];

            let vars: HashMap<_, _> = pairs
                .iter()
                .filter_map(|&(k, ref v)| v.clone().map(|v| (k, v)))
                .collect();
            let stream = replace_websocket_streams_placeholders("/<pair>_<contractType>@continuousKline_<interval>", &vars);

            let ws_stream = api.continuous_contract_kline_candlestick_streams(params).await.unwrap();

            let called = Arc::new(AtomicBool::new(false));
            let called_clone = called.clone();
            ws_stream.on_message(move |_payload: models::ContinuousContractKlineCandlestickStreamsResponse| {
                called_clone.store(true, Ordering::SeqCst);
            });

            assert!(streams_base.is_subscribed(&stream).await, "should be subscribed before unsubscribe");

            ws_stream.unsubscribe().await;

            let payload: Value = serde_json::from_str(r#"{"e":"continuous_kline","E":1591261542539,"ps":"BTCUSD","ct":"NEXT_QUARTER","k":{"t":1591261500000,"T":1591261559999,"i":"1m","f":606400,"L":606430,"o":"9638.9","c":"9639.8","h":"9639.8","l":"9638.6","v":"156","n":31,"x":false,"q":"1.61836886","V":"73","Q":"0.75731156","B":"0"}}"#).unwrap();
            let msg = json!({
                "stream": stream,
                "data": payload,
            });

            streams_base.on_message(msg.to_string(), conn.clone()).await;

            yield_now().await;

            assert!(!called.load(Ordering::SeqCst), "callback should not be invoked after unsubscribe");
        });
    }

    #[test]
    fn contract_info_stream_should_execute_successfully() {
        TOKIO_SHARED_RT.block_on(async {
            let (streams_base, _) = make_streams_base().await;
            let api = WebsocketMarketStreamsApiClient::new(streams_base.clone());

            let id = "test-id-123".to_string();

            let params = ContractInfoStreamParams::builder()
                .id(Some(id.clone()))
                .build()
                .unwrap();

            let ContractInfoStreamParams { id } = params.clone();

            let pairs: &[(&str, Option<String>)] = &[("id", id.clone())];

            let vars: HashMap<_, _> = pairs
                .iter()
                .filter_map(|&(k, ref v)| v.clone().map(|v| (k, v)))
                .collect();
            let stream = replace_websocket_streams_placeholders("/!contractInfo", &vars);
            let ws_stream = api
                .contract_info_stream(params)
                .await
                .expect("contract_info_stream should return a WebsocketStream");

            assert!(
                streams_base.is_subscribed(&stream).await,
                "expected stream '{stream}' to be subscribed"
            );
            assert_eq!(ws_stream.id, Some(StreamId::Str("test-id-123".to_string())));
        });
    }

    #[test]
    fn contract_info_stream_should_handle_incoming_message() {
        TOKIO_SHARED_RT.block_on(async {
            let (streams_base, conn) = make_streams_base().await;
            let api = WebsocketMarketStreamsApiClient::new(streams_base.clone());

            let id = "test-id-123".to_string();

            let params = ContractInfoStreamParams::builder().id(Some(id.clone())).build().unwrap();

            let ContractInfoStreamParams {
                id,
            } = params.clone();

            let pairs: &[(&str, Option<String>)] = &[
                ("id",
                        id.clone()
                ),
            ];

            let vars: HashMap<_, _> = pairs
                .iter()
                .filter_map(|&(k, ref v)| v.clone().map(|v| (k, v)))
                .collect();
            let stream = replace_websocket_streams_placeholders("/!contractInfo", &vars);

            let ws_stream = api.contract_info_stream(params).await.unwrap();

            let called = Arc::new(AtomicBool::new(false));
            let called_with_message = called.clone();
            ws_stream.on_message(move |_payload: models::ContractInfoStreamResponse| {
                called_with_message.store(true, Ordering::SeqCst);
            });

            let payload: Value = serde_json::from_str(r#"{"e":"contractInfo","E":1669647330375,"s":"APTUSD_PERP","ps":"APTUSD","ct":"PERPETUAL","dt":4133404800000,"ot":1666594800000,"cs":"TRADING","bks":[{"bs":1,"bnf":0,"bnc":5000,"mmr":0.01,"cf":0,"mi":21,"ma":50},{"bs":2,"bnf":5000,"bnc":25000,"mmr":0.025,"cf":75,"mi":11,"ma":20}]}"#).unwrap();
            let msg = json!({
                "stream": stream,
                "data": payload,
            });

            streams_base.on_message(msg.to_string(), conn.clone()).await;
            yield_now().await;

            assert!(called.load(Ordering::SeqCst), "expected our callback to have been invoked");
        });
    }

    #[test]
    fn contract_info_stream_should_not_fire_after_unsubscribe() {
        TOKIO_SHARED_RT.block_on(async {
            let (streams_base, conn) = make_streams_base().await;
            let api = WebsocketMarketStreamsApiClient::new(streams_base.clone());

            let id = "test-id-123".to_string();

            let params = ContractInfoStreamParams::builder().id(Some(id.clone())).build().unwrap();

            let ContractInfoStreamParams {
                id,
            } = params.clone();

            let pairs: &[(&str, Option<String>)] = &[
                ("id",
                        id.clone()
                ),
            ];

            let vars: HashMap<_, _> = pairs
                .iter()
                .filter_map(|&(k, ref v)| v.clone().map(|v| (k, v)))
                .collect();
            let stream = replace_websocket_streams_placeholders("/!contractInfo", &vars);

            let ws_stream = api.contract_info_stream(params).await.unwrap();

            let called = Arc::new(AtomicBool::new(false));
            let called_clone = called.clone();
            ws_stream.on_message(move |_payload: models::ContractInfoStreamResponse| {
                called_clone.store(true, Ordering::SeqCst);
            });

            assert!(streams_base.is_subscribed(&stream).await, "should be subscribed before unsubscribe");

            ws_stream.unsubscribe().await;

            let payload: Value = serde_json::from_str(r#"{"e":"contractInfo","E":1669647330375,"s":"APTUSD_PERP","ps":"APTUSD","ct":"PERPETUAL","dt":4133404800000,"ot":1666594800000,"cs":"TRADING","bks":[{"bs":1,"bnf":0,"bnc":5000,"mmr":0.01,"cf":0,"mi":21,"ma":50},{"bs":2,"bnf":5000,"bnc":25000,"mmr":0.025,"cf":75,"mi":11,"ma":20}]}"#).unwrap();
            let msg = json!({
                "stream": stream,
                "data": payload,
            });

            streams_base.on_message(msg.to_string(), conn.clone()).await;

            yield_now().await;

            assert!(!called.load(Ordering::SeqCst), "callback should not be invoked after unsubscribe");
        });
    }

    #[test]
    fn diff_book_depth_streams_should_execute_successfully() {
        TOKIO_SHARED_RT.block_on(async {
            let (streams_base, _) = make_streams_base().await;
            let api = WebsocketMarketStreamsApiClient::new(streams_base.clone());

            let id = "test-id-123".to_string();

            let params = DiffBookDepthStreamsParams::builder("btcusdt".to_string())
                .id(Some(id.clone()))
                .build()
                .unwrap();

            let DiffBookDepthStreamsParams {
                symbol,
                id,
                update_speed,
            } = params.clone();

            let pairs: &[(&str, Option<String>)] = &[
                ("symbol", Some(symbol.clone())),
                ("id", id.clone()),
                ("updateSpeed", update_speed.clone()),
            ];

            let vars: HashMap<_, _> = pairs
                .iter()
                .filter_map(|&(k, ref v)| v.clone().map(|v| (k, v)))
                .collect();
            let stream =
                replace_websocket_streams_placeholders("/<symbol>@depth@<updateSpeed>", &vars);
            let ws_stream = api
                .diff_book_depth_streams(params)
                .await
                .expect("diff_book_depth_streams should return a WebsocketStream");

            assert!(
                streams_base.is_subscribed(&stream).await,
                "expected stream '{stream}' to be subscribed"
            );
            assert_eq!(ws_stream.id, Some(StreamId::Str("test-id-123".to_string())));
        });
    }

    #[test]
    fn diff_book_depth_streams_should_handle_incoming_message() {
        TOKIO_SHARED_RT.block_on(async {
            let (streams_base, conn) = make_streams_base().await;
            let api = WebsocketMarketStreamsApiClient::new(streams_base.clone());

            let id = "test-id-123".to_string();

            let params = DiffBookDepthStreamsParams::builder("btcusdt".to_string(),).id(Some(id.clone())).build().unwrap();

            let DiffBookDepthStreamsParams {
                symbol,id,update_speed,
            } = params.clone();

            let pairs: &[(&str, Option<String>)] = &[
                ("symbol",
                        Some(symbol.clone())
                ),
                ("id",
                        id.clone()
                ),
                ("updateSpeed",
                        update_speed.clone()
                ),
            ];

            let vars: HashMap<_, _> = pairs
                .iter()
                .filter_map(|&(k, ref v)| v.clone().map(|v| (k, v)))
                .collect();
            let stream = replace_websocket_streams_placeholders("/<symbol>@depth@<updateSpeed>", &vars);

            let ws_stream = api.diff_book_depth_streams(params).await.unwrap();

            let called = Arc::new(AtomicBool::new(false));
            let called_with_message = called.clone();
            ws_stream.on_message(move |_payload: models::DiffBookDepthStreamsResponse| {
                called_with_message.store(true, Ordering::SeqCst);
            });

            let payload: Value = serde_json::from_str(r#"{"e":"depthUpdate","E":1591270260907,"T":1591270260891,"s":"BTCUSD_200626","ps":"BTCUSD","U":17285681,"u":17285702,"pu":17285675,"b":[["9517.6","10"]],"a":[["9518.5","45"]]}"#).unwrap();
            let msg = json!({
                "stream": stream,
                "data": payload,
            });

            streams_base.on_message(msg.to_string(), conn.clone()).await;
            yield_now().await;

            assert!(called.load(Ordering::SeqCst), "expected our callback to have been invoked");
        });
    }

    #[test]
    fn diff_book_depth_streams_should_not_fire_after_unsubscribe() {
        TOKIO_SHARED_RT.block_on(async {
            let (streams_base, conn) = make_streams_base().await;
            let api = WebsocketMarketStreamsApiClient::new(streams_base.clone());

            let id = "test-id-123".to_string();

            let params = DiffBookDepthStreamsParams::builder("btcusdt".to_string(),).id(Some(id.clone())).build().unwrap();

            let DiffBookDepthStreamsParams {
                symbol,id,update_speed,
            } = params.clone();

            let pairs: &[(&str, Option<String>)] = &[
                ("symbol",
                        Some(symbol.clone())
                ),
                ("id",
                        id.clone()
                ),
                ("updateSpeed",
                        update_speed.clone()
                ),
            ];

            let vars: HashMap<_, _> = pairs
                .iter()
                .filter_map(|&(k, ref v)| v.clone().map(|v| (k, v)))
                .collect();
            let stream = replace_websocket_streams_placeholders("/<symbol>@depth@<updateSpeed>", &vars);

            let ws_stream = api.diff_book_depth_streams(params).await.unwrap();

            let called = Arc::new(AtomicBool::new(false));
            let called_clone = called.clone();
            ws_stream.on_message(move |_payload: models::DiffBookDepthStreamsResponse| {
                called_clone.store(true, Ordering::SeqCst);
            });

            assert!(streams_base.is_subscribed(&stream).await, "should be subscribed before unsubscribe");

            ws_stream.unsubscribe().await;

            let payload: Value = serde_json::from_str(r#"{"e":"depthUpdate","E":1591270260907,"T":1591270260891,"s":"BTCUSD_200626","ps":"BTCUSD","U":17285681,"u":17285702,"pu":17285675,"b":[["9517.6","10"]],"a":[["9518.5","45"]]}"#).unwrap();
            let msg = json!({
                "stream": stream,
                "data": payload,
            });

            streams_base.on_message(msg.to_string(), conn.clone()).await;

            yield_now().await;

            assert!(!called.load(Ordering::SeqCst), "callback should not be invoked after unsubscribe");
        });
    }

    #[test]
    fn index_kline_candlestick_streams_should_execute_successfully() {
        TOKIO_SHARED_RT.block_on(async {
            let (streams_base, _) = make_streams_base().await;
            let api = WebsocketMarketStreamsApiClient::new(streams_base.clone());

            let id = "test-id-123".to_string();

            let params = IndexKlineCandlestickStreamsParams::builder(
                "btcusdt".to_string(),
                "1m".to_string(),
            )
            .id(Some(id.clone()))
            .build()
            .unwrap();

            let IndexKlineCandlestickStreamsParams { pair, interval, id } = params.clone();

            let pairs: &[(&str, Option<String>)] = &[
                ("pair", Some(pair.clone())),
                ("interval", Some(interval.clone())),
                ("id", id.clone()),
            ];

            let vars: HashMap<_, _> = pairs
                .iter()
                .filter_map(|&(k, ref v)| v.clone().map(|v| (k, v)))
                .collect();
            let stream =
                replace_websocket_streams_placeholders("/<pair>@indexPriceKline_<interval>", &vars);
            let ws_stream = api
                .index_kline_candlestick_streams(params)
                .await
                .expect("index_kline_candlestick_streams should return a WebsocketStream");

            assert!(
                streams_base.is_subscribed(&stream).await,
                "expected stream '{stream}' to be subscribed"
            );
            assert_eq!(ws_stream.id, Some(StreamId::Str("test-id-123".to_string())));
        });
    }

    #[test]
    fn index_kline_candlestick_streams_should_handle_incoming_message() {
        TOKIO_SHARED_RT.block_on(async {
            let (streams_base, conn) = make_streams_base().await;
            let api = WebsocketMarketStreamsApiClient::new(streams_base.clone());

            let id = "test-id-123".to_string();

            let params = IndexKlineCandlestickStreamsParams::builder("btcusdt".to_string(),"1m".to_string(),).id(Some(id.clone())).build().unwrap();

            let IndexKlineCandlestickStreamsParams {
                pair,interval,id,
            } = params.clone();

            let pairs: &[(&str, Option<String>)] = &[
                ("pair",
                        Some(pair.clone())
                ),
                ("interval",
                        Some(interval.clone())
                ),
                ("id",
                        id.clone()
                ),
            ];

            let vars: HashMap<_, _> = pairs
                .iter()
                .filter_map(|&(k, ref v)| v.clone().map(|v| (k, v)))
                .collect();
            let stream = replace_websocket_streams_placeholders("/<pair>@indexPriceKline_<interval>", &vars);

            let ws_stream = api.index_kline_candlestick_streams(params).await.unwrap();

            let called = Arc::new(AtomicBool::new(false));
            let called_with_message = called.clone();
            ws_stream.on_message(move |_payload: models::IndexKlineCandlestickStreamsResponse| {
                called_with_message.store(true, Ordering::SeqCst);
            });

            let payload: Value = serde_json::from_str(r#"{"e":"indexPrice_kline","E":1591267070033,"ps":"BTCUSD","k":{"t":1591267020000,"T":1591267079999,"s":"0","i":"1m","f":1591267020000,"L":1591267070000,"o":"9542.21900000","c":"9542.50440000","h":"9542.71640000","l":"9542.21040000","v":"0","n":51,"x":false,"q":"0","V":"0","Q":"0","B":"0"}}"#).unwrap();
            let msg = json!({
                "stream": stream,
                "data": payload,
            });

            streams_base.on_message(msg.to_string(), conn.clone()).await;
            yield_now().await;

            assert!(called.load(Ordering::SeqCst), "expected our callback to have been invoked");
        });
    }

    #[test]
    fn index_kline_candlestick_streams_should_not_fire_after_unsubscribe() {
        TOKIO_SHARED_RT.block_on(async {
            let (streams_base, conn) = make_streams_base().await;
            let api = WebsocketMarketStreamsApiClient::new(streams_base.clone());

            let id = "test-id-123".to_string();

            let params = IndexKlineCandlestickStreamsParams::builder("btcusdt".to_string(),"1m".to_string(),).id(Some(id.clone())).build().unwrap();

            let IndexKlineCandlestickStreamsParams {
                pair,interval,id,
            } = params.clone();

            let pairs: &[(&str, Option<String>)] = &[
                ("pair",
                        Some(pair.clone())
                ),
                ("interval",
                        Some(interval.clone())
                ),
                ("id",
                        id.clone()
                ),
            ];

            let vars: HashMap<_, _> = pairs
                .iter()
                .filter_map(|&(k, ref v)| v.clone().map(|v| (k, v)))
                .collect();
            let stream = replace_websocket_streams_placeholders("/<pair>@indexPriceKline_<interval>", &vars);

            let ws_stream = api.index_kline_candlestick_streams(params).await.unwrap();

            let called = Arc::new(AtomicBool::new(false));
            let called_clone = called.clone();
            ws_stream.on_message(move |_payload: models::IndexKlineCandlestickStreamsResponse| {
                called_clone.store(true, Ordering::SeqCst);
            });

            assert!(streams_base.is_subscribed(&stream).await, "should be subscribed before unsubscribe");

            ws_stream.unsubscribe().await;

            let payload: Value = serde_json::from_str(r#"{"e":"indexPrice_kline","E":1591267070033,"ps":"BTCUSD","k":{"t":1591267020000,"T":1591267079999,"s":"0","i":"1m","f":1591267020000,"L":1591267070000,"o":"9542.21900000","c":"9542.50440000","h":"9542.71640000","l":"9542.21040000","v":"0","n":51,"x":false,"q":"0","V":"0","Q":"0","B":"0"}}"#).unwrap();
            let msg = json!({
                "stream": stream,
                "data": payload,
            });

            streams_base.on_message(msg.to_string(), conn.clone()).await;

            yield_now().await;

            assert!(!called.load(Ordering::SeqCst), "callback should not be invoked after unsubscribe");
        });
    }

    #[test]
    fn index_price_stream_should_execute_successfully() {
        TOKIO_SHARED_RT.block_on(async {
            let (streams_base, _) = make_streams_base().await;
            let api = WebsocketMarketStreamsApiClient::new(streams_base.clone());

            let id = "test-id-123".to_string();

            let params = IndexPriceStreamParams::builder("btcusdt".to_string())
                .id(Some(id.clone()))
                .build()
                .unwrap();

            let IndexPriceStreamParams {
                pair,
                id,
                update_speed,
            } = params.clone();

            let pairs: &[(&str, Option<String>)] = &[
                ("pair", Some(pair.clone())),
                ("id", id.clone()),
                ("updateSpeed", update_speed.clone()),
            ];

            let vars: HashMap<_, _> = pairs
                .iter()
                .filter_map(|&(k, ref v)| v.clone().map(|v| (k, v)))
                .collect();
            let stream =
                replace_websocket_streams_placeholders("/<pair>@indexPrice@<updateSpeed>", &vars);
            let ws_stream = api
                .index_price_stream(params)
                .await
                .expect("index_price_stream should return a WebsocketStream");

            assert!(
                streams_base.is_subscribed(&stream).await,
                "expected stream '{stream}' to be subscribed"
            );
            assert_eq!(ws_stream.id, Some(StreamId::Str("test-id-123".to_string())));
        });
    }

    #[test]
    fn index_price_stream_should_handle_incoming_message() {
        TOKIO_SHARED_RT.block_on(async {
            let (streams_base, conn) = make_streams_base().await;
            let api = WebsocketMarketStreamsApiClient::new(streams_base.clone());

            let id = "test-id-123".to_string();

            let params = IndexPriceStreamParams::builder("btcusdt".to_string())
                .id(Some(id.clone()))
                .build()
                .unwrap();

            let IndexPriceStreamParams {
                pair,
                id,
                update_speed,
            } = params.clone();

            let pairs: &[(&str, Option<String>)] = &[
                ("pair", Some(pair.clone())),
                ("id", id.clone()),
                ("updateSpeed", update_speed.clone()),
            ];

            let vars: HashMap<_, _> = pairs
                .iter()
                .filter_map(|&(k, ref v)| v.clone().map(|v| (k, v)))
                .collect();
            let stream =
                replace_websocket_streams_placeholders("/<pair>@indexPrice@<updateSpeed>", &vars);

            let ws_stream = api.index_price_stream(params).await.unwrap();

            let called = Arc::new(AtomicBool::new(false));
            let called_with_message = called.clone();
            ws_stream.on_message(move |_payload: models::IndexPriceStreamResponse| {
                called_with_message.store(true, Ordering::SeqCst);
            });

            let payload: Value = serde_json::from_str(
                r#"{"e":"indexPriceUpdate","E":1591261236000,"i":"BTCUSD","p":"9636.57860000"}"#,
            )
            .unwrap();
            let msg = json!({
                "stream": stream,
                "data": payload,
            });

            streams_base.on_message(msg.to_string(), conn.clone()).await;
            yield_now().await;

            assert!(
                called.load(Ordering::SeqCst),
                "expected our callback to have been invoked"
            );
        });
    }

    #[test]
    fn index_price_stream_should_not_fire_after_unsubscribe() {
        TOKIO_SHARED_RT.block_on(async {
            let (streams_base, conn) = make_streams_base().await;
            let api = WebsocketMarketStreamsApiClient::new(streams_base.clone());

            let id = "test-id-123".to_string();

            let params = IndexPriceStreamParams::builder("btcusdt".to_string())
                .id(Some(id.clone()))
                .build()
                .unwrap();

            let IndexPriceStreamParams {
                pair,
                id,
                update_speed,
            } = params.clone();

            let pairs: &[(&str, Option<String>)] = &[
                ("pair", Some(pair.clone())),
                ("id", id.clone()),
                ("updateSpeed", update_speed.clone()),
            ];

            let vars: HashMap<_, _> = pairs
                .iter()
                .filter_map(|&(k, ref v)| v.clone().map(|v| (k, v)))
                .collect();
            let stream =
                replace_websocket_streams_placeholders("/<pair>@indexPrice@<updateSpeed>", &vars);

            let ws_stream = api.index_price_stream(params).await.unwrap();

            let called = Arc::new(AtomicBool::new(false));
            let called_clone = called.clone();
            ws_stream.on_message(move |_payload: models::IndexPriceStreamResponse| {
                called_clone.store(true, Ordering::SeqCst);
            });

            assert!(
                streams_base.is_subscribed(&stream).await,
                "should be subscribed before unsubscribe"
            );

            ws_stream.unsubscribe().await;

            let payload: Value = serde_json::from_str(
                r#"{"e":"indexPriceUpdate","E":1591261236000,"i":"BTCUSD","p":"9636.57860000"}"#,
            )
            .unwrap();
            let msg = json!({
                "stream": stream,
                "data": payload,
            });

            streams_base.on_message(msg.to_string(), conn.clone()).await;

            yield_now().await;

            assert!(
                !called.load(Ordering::SeqCst),
                "callback should not be invoked after unsubscribe"
            );
        });
    }

    #[test]
    fn individual_symbol_book_ticker_streams_should_execute_successfully() {
        TOKIO_SHARED_RT.block_on(async {
            let (streams_base, _) = make_streams_base().await;
            let api = WebsocketMarketStreamsApiClient::new(streams_base.clone());

            let id = "test-id-123".to_string();

            let params = IndividualSymbolBookTickerStreamsParams::builder("btcusdt".to_string())
                .id(Some(id.clone()))
                .build()
                .unwrap();

            let IndividualSymbolBookTickerStreamsParams { symbol, id } = params.clone();

            let pairs: &[(&str, Option<String>)] =
                &[("symbol", Some(symbol.clone())), ("id", id.clone())];

            let vars: HashMap<_, _> = pairs
                .iter()
                .filter_map(|&(k, ref v)| v.clone().map(|v| (k, v)))
                .collect();
            let stream = replace_websocket_streams_placeholders("/<symbol>@bookTicker", &vars);
            let ws_stream = api
                .individual_symbol_book_ticker_streams(params)
                .await
                .expect("individual_symbol_book_ticker_streams should return a WebsocketStream");

            assert!(
                streams_base.is_subscribed(&stream).await,
                "expected stream '{stream}' to be subscribed"
            );
            assert_eq!(ws_stream.id, Some(StreamId::Str("test-id-123".to_string())));
        });
    }

    #[test]
    fn individual_symbol_book_ticker_streams_should_handle_incoming_message() {
        TOKIO_SHARED_RT.block_on(async {
            let (streams_base, conn) = make_streams_base().await;
            let api = WebsocketMarketStreamsApiClient::new(streams_base.clone());

            let id = "test-id-123".to_string();

            let params = IndividualSymbolBookTickerStreamsParams::builder("btcusdt".to_string(),).id(Some(id.clone())).build().unwrap();

            let IndividualSymbolBookTickerStreamsParams {
                symbol,id,
            } = params.clone();

            let pairs: &[(&str, Option<String>)] = &[
                ("symbol",
                        Some(symbol.clone())
                ),
                ("id",
                        id.clone()
                ),
            ];

            let vars: HashMap<_, _> = pairs
                .iter()
                .filter_map(|&(k, ref v)| v.clone().map(|v| (k, v)))
                .collect();
            let stream = replace_websocket_streams_placeholders("/<symbol>@bookTicker", &vars);

            let ws_stream = api.individual_symbol_book_ticker_streams(params).await.unwrap();

            let called = Arc::new(AtomicBool::new(false));
            let called_with_message = called.clone();
            ws_stream.on_message(move |_payload: models::IndividualSymbolBookTickerStreamsResponse| {
                called_with_message.store(true, Ordering::SeqCst);
            });

            let payload: Value = serde_json::from_str(r#"{"e":"bookTicker","u":17242169,"s":"BTCUSD_200626","ps":"BTCUSD","b":"9548.1","B":"52","a":"9548.5","A":"11","T":1591268628155,"E":1591268628166}"#).unwrap();
            let msg = json!({
                "stream": stream,
                "data": payload,
            });

            streams_base.on_message(msg.to_string(), conn.clone()).await;
            yield_now().await;

            assert!(called.load(Ordering::SeqCst), "expected our callback to have been invoked");
        });
    }

    #[test]
    fn individual_symbol_book_ticker_streams_should_not_fire_after_unsubscribe() {
        TOKIO_SHARED_RT.block_on(async {
            let (streams_base, conn) = make_streams_base().await;
            let api = WebsocketMarketStreamsApiClient::new(streams_base.clone());

            let id = "test-id-123".to_string();

            let params = IndividualSymbolBookTickerStreamsParams::builder("btcusdt".to_string(),).id(Some(id.clone())).build().unwrap();

            let IndividualSymbolBookTickerStreamsParams {
                symbol,id,
            } = params.clone();

            let pairs: &[(&str, Option<String>)] = &[
                ("symbol",
                        Some(symbol.clone())
                ),
                ("id",
                        id.clone()
                ),
            ];

            let vars: HashMap<_, _> = pairs
                .iter()
                .filter_map(|&(k, ref v)| v.clone().map(|v| (k, v)))
                .collect();
            let stream = replace_websocket_streams_placeholders("/<symbol>@bookTicker", &vars);

            let ws_stream = api.individual_symbol_book_ticker_streams(params).await.unwrap();

            let called = Arc::new(AtomicBool::new(false));
            let called_clone = called.clone();
            ws_stream.on_message(move |_payload: models::IndividualSymbolBookTickerStreamsResponse| {
                called_clone.store(true, Ordering::SeqCst);
            });

            assert!(streams_base.is_subscribed(&stream).await, "should be subscribed before unsubscribe");

            ws_stream.unsubscribe().await;

            let payload: Value = serde_json::from_str(r#"{"e":"bookTicker","u":17242169,"s":"BTCUSD_200626","ps":"BTCUSD","b":"9548.1","B":"52","a":"9548.5","A":"11","T":1591268628155,"E":1591268628166}"#).unwrap();
            let msg = json!({
                "stream": stream,
                "data": payload,
            });

            streams_base.on_message(msg.to_string(), conn.clone()).await;

            yield_now().await;

            assert!(!called.load(Ordering::SeqCst), "callback should not be invoked after unsubscribe");
        });
    }

    #[test]
    fn individual_symbol_mini_ticker_stream_should_execute_successfully() {
        TOKIO_SHARED_RT.block_on(async {
            let (streams_base, _) = make_streams_base().await;
            let api = WebsocketMarketStreamsApiClient::new(streams_base.clone());

            let id = "test-id-123".to_string();

            let params = IndividualSymbolMiniTickerStreamParams::builder("btcusdt".to_string())
                .id(Some(id.clone()))
                .build()
                .unwrap();

            let IndividualSymbolMiniTickerStreamParams { symbol, id } = params.clone();

            let pairs: &[(&str, Option<String>)] =
                &[("symbol", Some(symbol.clone())), ("id", id.clone())];

            let vars: HashMap<_, _> = pairs
                .iter()
                .filter_map(|&(k, ref v)| v.clone().map(|v| (k, v)))
                .collect();
            let stream = replace_websocket_streams_placeholders("/<symbol>@miniTicker", &vars);
            let ws_stream = api
                .individual_symbol_mini_ticker_stream(params)
                .await
                .expect("individual_symbol_mini_ticker_stream should return a WebsocketStream");

            assert!(
                streams_base.is_subscribed(&stream).await,
                "expected stream '{stream}' to be subscribed"
            );
            assert_eq!(ws_stream.id, Some(StreamId::Str("test-id-123".to_string())));
        });
    }

    #[test]
    fn individual_symbol_mini_ticker_stream_should_handle_incoming_message() {
        TOKIO_SHARED_RT.block_on(async {
            let (streams_base, conn) = make_streams_base().await;
            let api = WebsocketMarketStreamsApiClient::new(streams_base.clone());

            let id = "test-id-123".to_string();

            let params = IndividualSymbolMiniTickerStreamParams::builder("btcusdt".to_string(),).id(Some(id.clone())).build().unwrap();

            let IndividualSymbolMiniTickerStreamParams {
                symbol,id,
            } = params.clone();

            let pairs: &[(&str, Option<String>)] = &[
                ("symbol",
                        Some(symbol.clone())
                ),
                ("id",
                        id.clone()
                ),
            ];

            let vars: HashMap<_, _> = pairs
                .iter()
                .filter_map(|&(k, ref v)| v.clone().map(|v| (k, v)))
                .collect();
            let stream = replace_websocket_streams_placeholders("/<symbol>@miniTicker", &vars);

            let ws_stream = api.individual_symbol_mini_ticker_stream(params).await.unwrap();

            let called = Arc::new(AtomicBool::new(false));
            let called_with_message = called.clone();
            ws_stream.on_message(move |_payload: models::IndividualSymbolMiniTickerStreamResponse| {
                called_with_message.store(true, Ordering::SeqCst);
            });

            let payload: Value = serde_json::from_str(r#"{"e":"24hrMiniTicker","E":1591267704450,"s":"BTCUSD_200626","ps":"BTCUSD","c":"9561.7","o":"9580.9","h":"10000.0","l":"7000.0","v":"487476","q":"33264343847.22378500"}"#).unwrap();
            let msg = json!({
                "stream": stream,
                "data": payload,
            });

            streams_base.on_message(msg.to_string(), conn.clone()).await;
            yield_now().await;

            assert!(called.load(Ordering::SeqCst), "expected our callback to have been invoked");
        });
    }

    #[test]
    fn individual_symbol_mini_ticker_stream_should_not_fire_after_unsubscribe() {
        TOKIO_SHARED_RT.block_on(async {
            let (streams_base, conn) = make_streams_base().await;
            let api = WebsocketMarketStreamsApiClient::new(streams_base.clone());

            let id = "test-id-123".to_string();

            let params = IndividualSymbolMiniTickerStreamParams::builder("btcusdt".to_string(),).id(Some(id.clone())).build().unwrap();

            let IndividualSymbolMiniTickerStreamParams {
                symbol,id,
            } = params.clone();

            let pairs: &[(&str, Option<String>)] = &[
                ("symbol",
                        Some(symbol.clone())
                ),
                ("id",
                        id.clone()
                ),
            ];

            let vars: HashMap<_, _> = pairs
                .iter()
                .filter_map(|&(k, ref v)| v.clone().map(|v| (k, v)))
                .collect();
            let stream = replace_websocket_streams_placeholders("/<symbol>@miniTicker", &vars);

            let ws_stream = api.individual_symbol_mini_ticker_stream(params).await.unwrap();

            let called = Arc::new(AtomicBool::new(false));
            let called_clone = called.clone();
            ws_stream.on_message(move |_payload: models::IndividualSymbolMiniTickerStreamResponse| {
                called_clone.store(true, Ordering::SeqCst);
            });

            assert!(streams_base.is_subscribed(&stream).await, "should be subscribed before unsubscribe");

            ws_stream.unsubscribe().await;

            let payload: Value = serde_json::from_str(r#"{"e":"24hrMiniTicker","E":1591267704450,"s":"BTCUSD_200626","ps":"BTCUSD","c":"9561.7","o":"9580.9","h":"10000.0","l":"7000.0","v":"487476","q":"33264343847.22378500"}"#).unwrap();
            let msg = json!({
                "stream": stream,
                "data": payload,
            });

            streams_base.on_message(msg.to_string(), conn.clone()).await;

            yield_now().await;

            assert!(!called.load(Ordering::SeqCst), "callback should not be invoked after unsubscribe");
        });
    }

    #[test]
    fn individual_symbol_ticker_streams_should_execute_successfully() {
        TOKIO_SHARED_RT.block_on(async {
            let (streams_base, _) = make_streams_base().await;
            let api = WebsocketMarketStreamsApiClient::new(streams_base.clone());

            let id = "test-id-123".to_string();

            let params = IndividualSymbolTickerStreamsParams::builder("btcusdt".to_string())
                .id(Some(id.clone()))
                .build()
                .unwrap();

            let IndividualSymbolTickerStreamsParams { symbol, id } = params.clone();

            let pairs: &[(&str, Option<String>)] =
                &[("symbol", Some(symbol.clone())), ("id", id.clone())];

            let vars: HashMap<_, _> = pairs
                .iter()
                .filter_map(|&(k, ref v)| v.clone().map(|v| (k, v)))
                .collect();
            let stream = replace_websocket_streams_placeholders("/<symbol>@ticker", &vars);
            let ws_stream = api
                .individual_symbol_ticker_streams(params)
                .await
                .expect("individual_symbol_ticker_streams should return a WebsocketStream");

            assert!(
                streams_base.is_subscribed(&stream).await,
                "expected stream '{stream}' to be subscribed"
            );
            assert_eq!(ws_stream.id, Some(StreamId::Str("test-id-123".to_string())));
        });
    }

    #[test]
    fn individual_symbol_ticker_streams_should_handle_incoming_message() {
        TOKIO_SHARED_RT.block_on(async {
            let (streams_base, conn) = make_streams_base().await;
            let api = WebsocketMarketStreamsApiClient::new(streams_base.clone());

            let id = "test-id-123".to_string();

            let params = IndividualSymbolTickerStreamsParams::builder("btcusdt".to_string(),).id(Some(id.clone())).build().unwrap();

            let IndividualSymbolTickerStreamsParams {
                symbol,id,
            } = params.clone();

            let pairs: &[(&str, Option<String>)] = &[
                ("symbol",
                        Some(symbol.clone())
                ),
                ("id",
                        id.clone()
                ),
            ];

            let vars: HashMap<_, _> = pairs
                .iter()
                .filter_map(|&(k, ref v)| v.clone().map(|v| (k, v)))
                .collect();
            let stream = replace_websocket_streams_placeholders("/<symbol>@ticker", &vars);

            let ws_stream = api.individual_symbol_ticker_streams(params).await.unwrap();

            let called = Arc::new(AtomicBool::new(false));
            let called_with_message = called.clone();
            ws_stream.on_message(move |_payload: models::IndividualSymbolTickerStreamsResponse| {
                called_with_message.store(true, Ordering::SeqCst);
            });

            let payload: Value = serde_json::from_str(r#"{"e":"24hrTicker","E":1591268262453,"s":"BTCUSD_200626","ps":"BTCUSD","p":"-43.4","P":"-0.452","w":"0.00147974","c":"9548.5","Q":"2","o":"9591.9","h":"10000.0","l":"7000.0","v":"487850","q":"32968676323.46222700","O":1591181820000,"C":1591268262442,"F":512014,"L":615289,"n":103272}"#).unwrap();
            let msg = json!({
                "stream": stream,
                "data": payload,
            });

            streams_base.on_message(msg.to_string(), conn.clone()).await;
            yield_now().await;

            assert!(called.load(Ordering::SeqCst), "expected our callback to have been invoked");
        });
    }

    #[test]
    fn individual_symbol_ticker_streams_should_not_fire_after_unsubscribe() {
        TOKIO_SHARED_RT.block_on(async {
            let (streams_base, conn) = make_streams_base().await;
            let api = WebsocketMarketStreamsApiClient::new(streams_base.clone());

            let id = "test-id-123".to_string();

            let params = IndividualSymbolTickerStreamsParams::builder("btcusdt".to_string(),).id(Some(id.clone())).build().unwrap();

            let IndividualSymbolTickerStreamsParams {
                symbol,id,
            } = params.clone();

            let pairs: &[(&str, Option<String>)] = &[
                ("symbol",
                        Some(symbol.clone())
                ),
                ("id",
                        id.clone()
                ),
            ];

            let vars: HashMap<_, _> = pairs
                .iter()
                .filter_map(|&(k, ref v)| v.clone().map(|v| (k, v)))
                .collect();
            let stream = replace_websocket_streams_placeholders("/<symbol>@ticker", &vars);

            let ws_stream = api.individual_symbol_ticker_streams(params).await.unwrap();

            let called = Arc::new(AtomicBool::new(false));
            let called_clone = called.clone();
            ws_stream.on_message(move |_payload: models::IndividualSymbolTickerStreamsResponse| {
                called_clone.store(true, Ordering::SeqCst);
            });

            assert!(streams_base.is_subscribed(&stream).await, "should be subscribed before unsubscribe");

            ws_stream.unsubscribe().await;

            let payload: Value = serde_json::from_str(r#"{"e":"24hrTicker","E":1591268262453,"s":"BTCUSD_200626","ps":"BTCUSD","p":"-43.4","P":"-0.452","w":"0.00147974","c":"9548.5","Q":"2","o":"9591.9","h":"10000.0","l":"7000.0","v":"487850","q":"32968676323.46222700","O":1591181820000,"C":1591268262442,"F":512014,"L":615289,"n":103272}"#).unwrap();
            let msg = json!({
                "stream": stream,
                "data": payload,
            });

            streams_base.on_message(msg.to_string(), conn.clone()).await;

            yield_now().await;

            assert!(!called.load(Ordering::SeqCst), "callback should not be invoked after unsubscribe");
        });
    }

    #[test]
    fn kline_candlestick_streams_should_execute_successfully() {
        TOKIO_SHARED_RT.block_on(async {
            let (streams_base, _) = make_streams_base().await;
            let api = WebsocketMarketStreamsApiClient::new(streams_base.clone());

            let id = "test-id-123".to_string();

            let params =
                KlineCandlestickStreamsParams::builder("btcusdt".to_string(), "1m".to_string())
                    .id(Some(id.clone()))
                    .build()
                    .unwrap();

            let KlineCandlestickStreamsParams {
                symbol,
                interval,
                id,
            } = params.clone();

            let pairs: &[(&str, Option<String>)] = &[
                ("symbol", Some(symbol.clone())),
                ("interval", Some(interval.clone())),
                ("id", id.clone()),
            ];

            let vars: HashMap<_, _> = pairs
                .iter()
                .filter_map(|&(k, ref v)| v.clone().map(|v| (k, v)))
                .collect();
            let stream =
                replace_websocket_streams_placeholders("/<symbol>@kline_<interval>", &vars);
            let ws_stream = api
                .kline_candlestick_streams(params)
                .await
                .expect("kline_candlestick_streams should return a WebsocketStream");

            assert!(
                streams_base.is_subscribed(&stream).await,
                "expected stream '{stream}' to be subscribed"
            );
            assert_eq!(ws_stream.id, Some(StreamId::Str("test-id-123".to_string())));
        });
    }

    #[test]
    fn kline_candlestick_streams_should_handle_incoming_message() {
        TOKIO_SHARED_RT.block_on(async {
            let (streams_base, conn) = make_streams_base().await;
            let api = WebsocketMarketStreamsApiClient::new(streams_base.clone());

            let id = "test-id-123".to_string();

            let params = KlineCandlestickStreamsParams::builder("btcusdt".to_string(),"1m".to_string(),).id(Some(id.clone())).build().unwrap();

            let KlineCandlestickStreamsParams {
                symbol,interval,id,
            } = params.clone();

            let pairs: &[(&str, Option<String>)] = &[
                ("symbol",
                        Some(symbol.clone())
                ),
                ("interval",
                        Some(interval.clone())
                ),
                ("id",
                        id.clone()
                ),
            ];

            let vars: HashMap<_, _> = pairs
                .iter()
                .filter_map(|&(k, ref v)| v.clone().map(|v| (k, v)))
                .collect();
            let stream = replace_websocket_streams_placeholders("/<symbol>@kline_<interval>", &vars);

            let ws_stream = api.kline_candlestick_streams(params).await.unwrap();

            let called = Arc::new(AtomicBool::new(false));
            let called_with_message = called.clone();
            ws_stream.on_message(move |_payload: models::KlineCandlestickStreamsResponse| {
                called_with_message.store(true, Ordering::SeqCst);
            });

            let payload: Value = serde_json::from_str(r#"{"e":"kline","E":1591261542539,"s":"BTCUSD_200626","k":{"t":1591261500000,"T":1591261559999,"s":"BTCUSD_200626","i":"1m","f":606400,"L":606430,"o":"9638.9","c":"9639.8","h":"9639.8","l":"9638.6","v":"156","n":31,"x":false,"q":"1.61836886","V":"73","Q":"0.75731156","B":"0"}}"#).unwrap();
            let msg = json!({
                "stream": stream,
                "data": payload,
            });

            streams_base.on_message(msg.to_string(), conn.clone()).await;
            yield_now().await;

            assert!(called.load(Ordering::SeqCst), "expected our callback to have been invoked");
        });
    }

    #[test]
    fn kline_candlestick_streams_should_not_fire_after_unsubscribe() {
        TOKIO_SHARED_RT.block_on(async {
            let (streams_base, conn) = make_streams_base().await;
            let api = WebsocketMarketStreamsApiClient::new(streams_base.clone());

            let id = "test-id-123".to_string();

            let params = KlineCandlestickStreamsParams::builder("btcusdt".to_string(),"1m".to_string(),).id(Some(id.clone())).build().unwrap();

            let KlineCandlestickStreamsParams {
                symbol,interval,id,
            } = params.clone();

            let pairs: &[(&str, Option<String>)] = &[
                ("symbol",
                        Some(symbol.clone())
                ),
                ("interval",
                        Some(interval.clone())
                ),
                ("id",
                        id.clone()
                ),
            ];

            let vars: HashMap<_, _> = pairs
                .iter()
                .filter_map(|&(k, ref v)| v.clone().map(|v| (k, v)))
                .collect();
            let stream = replace_websocket_streams_placeholders("/<symbol>@kline_<interval>", &vars);

            let ws_stream = api.kline_candlestick_streams(params).await.unwrap();

            let called = Arc::new(AtomicBool::new(false));
            let called_clone = called.clone();
            ws_stream.on_message(move |_payload: models::KlineCandlestickStreamsResponse| {
                called_clone.store(true, Ordering::SeqCst);
            });

            assert!(streams_base.is_subscribed(&stream).await, "should be subscribed before unsubscribe");

            ws_stream.unsubscribe().await;

            let payload: Value = serde_json::from_str(r#"{"e":"kline","E":1591261542539,"s":"BTCUSD_200626","k":{"t":1591261500000,"T":1591261559999,"s":"BTCUSD_200626","i":"1m","f":606400,"L":606430,"o":"9638.9","c":"9639.8","h":"9639.8","l":"9638.6","v":"156","n":31,"x":false,"q":"1.61836886","V":"73","Q":"0.75731156","B":"0"}}"#).unwrap();
            let msg = json!({
                "stream": stream,
                "data": payload,
            });

            streams_base.on_message(msg.to_string(), conn.clone()).await;

            yield_now().await;

            assert!(!called.load(Ordering::SeqCst), "callback should not be invoked after unsubscribe");
        });
    }

    #[test]
    fn liquidation_order_streams_should_execute_successfully() {
        TOKIO_SHARED_RT.block_on(async {
            let (streams_base, _) = make_streams_base().await;
            let api = WebsocketMarketStreamsApiClient::new(streams_base.clone());

            let id = "test-id-123".to_string();

            let params = LiquidationOrderStreamsParams::builder("btcusdt".to_string())
                .id(Some(id.clone()))
                .build()
                .unwrap();

            let LiquidationOrderStreamsParams { symbol, id } = params.clone();

            let pairs: &[(&str, Option<String>)] =
                &[("symbol", Some(symbol.clone())), ("id", id.clone())];

            let vars: HashMap<_, _> = pairs
                .iter()
                .filter_map(|&(k, ref v)| v.clone().map(|v| (k, v)))
                .collect();
            let stream = replace_websocket_streams_placeholders("/<symbol>@forceOrder", &vars);
            let ws_stream = api
                .liquidation_order_streams(params)
                .await
                .expect("liquidation_order_streams should return a WebsocketStream");

            assert!(
                streams_base.is_subscribed(&stream).await,
                "expected stream '{stream}' to be subscribed"
            );
            assert_eq!(ws_stream.id, Some(StreamId::Str("test-id-123".to_string())));
        });
    }

    #[test]
    fn liquidation_order_streams_should_handle_incoming_message() {
        TOKIO_SHARED_RT.block_on(async {
            let (streams_base, conn) = make_streams_base().await;
            let api = WebsocketMarketStreamsApiClient::new(streams_base.clone());

            let id = "test-id-123".to_string();

            let params = LiquidationOrderStreamsParams::builder("btcusdt".to_string(),).id(Some(id.clone())).build().unwrap();

            let LiquidationOrderStreamsParams {
                symbol,id,
            } = params.clone();

            let pairs: &[(&str, Option<String>)] = &[
                ("symbol",
                        Some(symbol.clone())
                ),
                ("id",
                        id.clone()
                ),
            ];

            let vars: HashMap<_, _> = pairs
                .iter()
                .filter_map(|&(k, ref v)| v.clone().map(|v| (k, v)))
                .collect();
            let stream = replace_websocket_streams_placeholders("/<symbol>@forceOrder", &vars);

            let ws_stream = api.liquidation_order_streams(params).await.unwrap();

            let called = Arc::new(AtomicBool::new(false));
            let called_with_message = called.clone();
            ws_stream.on_message(move |_payload: models::LiquidationOrderStreamsResponse| {
                called_with_message.store(true, Ordering::SeqCst);
            });

            let payload: Value = serde_json::from_str(r#"{"e":"forceOrder","E":1591154240950,"o":{"s":"BTCUSD_200925","ps":"BTCUSD","S":"SELL","o":"LIMIT","f":"IOC","q":"1","p":"9425.5","ap":"9496.5","X":"FILLED","l":"1","z":"1","T":1591154240949}}"#).unwrap();
            let msg = json!({
                "stream": stream,
                "data": payload,
            });

            streams_base.on_message(msg.to_string(), conn.clone()).await;
            yield_now().await;

            assert!(called.load(Ordering::SeqCst), "expected our callback to have been invoked");
        });
    }

    #[test]
    fn liquidation_order_streams_should_not_fire_after_unsubscribe() {
        TOKIO_SHARED_RT.block_on(async {
            let (streams_base, conn) = make_streams_base().await;
            let api = WebsocketMarketStreamsApiClient::new(streams_base.clone());

            let id = "test-id-123".to_string();

            let params = LiquidationOrderStreamsParams::builder("btcusdt".to_string(),).id(Some(id.clone())).build().unwrap();

            let LiquidationOrderStreamsParams {
                symbol,id,
            } = params.clone();

            let pairs: &[(&str, Option<String>)] = &[
                ("symbol",
                        Some(symbol.clone())
                ),
                ("id",
                        id.clone()
                ),
            ];

            let vars: HashMap<_, _> = pairs
                .iter()
                .filter_map(|&(k, ref v)| v.clone().map(|v| (k, v)))
                .collect();
            let stream = replace_websocket_streams_placeholders("/<symbol>@forceOrder", &vars);

            let ws_stream = api.liquidation_order_streams(params).await.unwrap();

            let called = Arc::new(AtomicBool::new(false));
            let called_clone = called.clone();
            ws_stream.on_message(move |_payload: models::LiquidationOrderStreamsResponse| {
                called_clone.store(true, Ordering::SeqCst);
            });

            assert!(streams_base.is_subscribed(&stream).await, "should be subscribed before unsubscribe");

            ws_stream.unsubscribe().await;

            let payload: Value = serde_json::from_str(r#"{"e":"forceOrder","E":1591154240950,"o":{"s":"BTCUSD_200925","ps":"BTCUSD","S":"SELL","o":"LIMIT","f":"IOC","q":"1","p":"9425.5","ap":"9496.5","X":"FILLED","l":"1","z":"1","T":1591154240949}}"#).unwrap();
            let msg = json!({
                "stream": stream,
                "data": payload,
            });

            streams_base.on_message(msg.to_string(), conn.clone()).await;

            yield_now().await;

            assert!(!called.load(Ordering::SeqCst), "callback should not be invoked after unsubscribe");
        });
    }

    #[test]
    fn mark_price_kline_candlestick_streams_should_execute_successfully() {
        TOKIO_SHARED_RT.block_on(async {
            let (streams_base, _) = make_streams_base().await;
            let api = WebsocketMarketStreamsApiClient::new(streams_base.clone());

            let id = "test-id-123".to_string();

            let params = MarkPriceKlineCandlestickStreamsParams::builder(
                "btcusdt".to_string(),
                "1m".to_string(),
            )
            .id(Some(id.clone()))
            .build()
            .unwrap();

            let MarkPriceKlineCandlestickStreamsParams {
                symbol,
                interval,
                id,
            } = params.clone();

            let pairs: &[(&str, Option<String>)] = &[
                ("symbol", Some(symbol.clone())),
                ("interval", Some(interval.clone())),
                ("id", id.clone()),
            ];

            let vars: HashMap<_, _> = pairs
                .iter()
                .filter_map(|&(k, ref v)| v.clone().map(|v| (k, v)))
                .collect();
            let stream = replace_websocket_streams_placeholders(
                "/<symbol>@markPriceKline_<interval>",
                &vars,
            );
            let ws_stream = api
                .mark_price_kline_candlestick_streams(params)
                .await
                .expect("mark_price_kline_candlestick_streams should return a WebsocketStream");

            assert!(
                streams_base.is_subscribed(&stream).await,
                "expected stream '{stream}' to be subscribed"
            );
            assert_eq!(ws_stream.id, Some(StreamId::Str("test-id-123".to_string())));
        });
    }

    #[test]
    fn mark_price_kline_candlestick_streams_should_handle_incoming_message() {
        TOKIO_SHARED_RT.block_on(async {
            let (streams_base, conn) = make_streams_base().await;
            let api = WebsocketMarketStreamsApiClient::new(streams_base.clone());

            let id = "test-id-123".to_string();

            let params = MarkPriceKlineCandlestickStreamsParams::builder("btcusdt".to_string(),"1m".to_string(),).id(Some(id.clone())).build().unwrap();

            let MarkPriceKlineCandlestickStreamsParams {
                symbol,interval,id,
            } = params.clone();

            let pairs: &[(&str, Option<String>)] = &[
                ("symbol",
                        Some(symbol.clone())
                ),
                ("interval",
                        Some(interval.clone())
                ),
                ("id",
                        id.clone()
                ),
            ];

            let vars: HashMap<_, _> = pairs
                .iter()
                .filter_map(|&(k, ref v)| v.clone().map(|v| (k, v)))
                .collect();
            let stream = replace_websocket_streams_placeholders("/<symbol>@markPriceKline_<interval>", &vars);

            let ws_stream = api.mark_price_kline_candlestick_streams(params).await.unwrap();

            let called = Arc::new(AtomicBool::new(false));
            let called_with_message = called.clone();
            ws_stream.on_message(move |_payload: models::MarkPriceKlineCandlestickStreamsResponse| {
                called_with_message.store(true, Ordering::SeqCst);
            });

            let payload: Value = serde_json::from_str(r#"{"e":"markPrice_kline","E":1591267398004,"ps":"BTCUSD","k":{"t":1591267380000,"T":1591267439999,"s":"BTCUSD_200626","i":"1m","f":1591267380000,"L":1591267398000,"o":"9539.67161333","c":"9540.82761333","h":"9540.82761333","l":"9539.66961333","v":"0","n":19,"x":false,"q":"0","V":"0","Q":"0","B":"0"}}"#).unwrap();
            let msg = json!({
                "stream": stream,
                "data": payload,
            });

            streams_base.on_message(msg.to_string(), conn.clone()).await;
            yield_now().await;

            assert!(called.load(Ordering::SeqCst), "expected our callback to have been invoked");
        });
    }

    #[test]
    fn mark_price_kline_candlestick_streams_should_not_fire_after_unsubscribe() {
        TOKIO_SHARED_RT.block_on(async {
            let (streams_base, conn) = make_streams_base().await;
            let api = WebsocketMarketStreamsApiClient::new(streams_base.clone());

            let id = "test-id-123".to_string();

            let params = MarkPriceKlineCandlestickStreamsParams::builder("btcusdt".to_string(),"1m".to_string(),).id(Some(id.clone())).build().unwrap();

            let MarkPriceKlineCandlestickStreamsParams {
                symbol,interval,id,
            } = params.clone();

            let pairs: &[(&str, Option<String>)] = &[
                ("symbol",
                        Some(symbol.clone())
                ),
                ("interval",
                        Some(interval.clone())
                ),
                ("id",
                        id.clone()
                ),
            ];

            let vars: HashMap<_, _> = pairs
                .iter()
                .filter_map(|&(k, ref v)| v.clone().map(|v| (k, v)))
                .collect();
            let stream = replace_websocket_streams_placeholders("/<symbol>@markPriceKline_<interval>", &vars);

            let ws_stream = api.mark_price_kline_candlestick_streams(params).await.unwrap();

            let called = Arc::new(AtomicBool::new(false));
            let called_clone = called.clone();
            ws_stream.on_message(move |_payload: models::MarkPriceKlineCandlestickStreamsResponse| {
                called_clone.store(true, Ordering::SeqCst);
            });

            assert!(streams_base.is_subscribed(&stream).await, "should be subscribed before unsubscribe");

            ws_stream.unsubscribe().await;

            let payload: Value = serde_json::from_str(r#"{"e":"markPrice_kline","E":1591267398004,"ps":"BTCUSD","k":{"t":1591267380000,"T":1591267439999,"s":"BTCUSD_200626","i":"1m","f":1591267380000,"L":1591267398000,"o":"9539.67161333","c":"9540.82761333","h":"9540.82761333","l":"9539.66961333","v":"0","n":19,"x":false,"q":"0","V":"0","Q":"0","B":"0"}}"#).unwrap();
            let msg = json!({
                "stream": stream,
                "data": payload,
            });

            streams_base.on_message(msg.to_string(), conn.clone()).await;

            yield_now().await;

            assert!(!called.load(Ordering::SeqCst), "callback should not be invoked after unsubscribe");
        });
    }

    #[test]
    fn mark_price_of_all_symbols_of_a_pair_should_execute_successfully() {
        TOKIO_SHARED_RT.block_on(async {
            let (streams_base, _) = make_streams_base().await;
            let api = WebsocketMarketStreamsApiClient::new(streams_base.clone());

            let id = "test-id-123".to_string();

            let params = MarkPriceOfAllSymbolsOfAPairParams::builder("btcusdt".to_string())
                .id(Some(id.clone()))
                .build()
                .unwrap();

            let MarkPriceOfAllSymbolsOfAPairParams {
                pair,
                id,
                update_speed,
            } = params.clone();

            let pairs: &[(&str, Option<String>)] = &[
                ("pair", Some(pair.clone())),
                ("id", id.clone()),
                ("updateSpeed", update_speed.clone()),
            ];

            let vars: HashMap<_, _> = pairs
                .iter()
                .filter_map(|&(k, ref v)| v.clone().map(|v| (k, v)))
                .collect();
            let stream =
                replace_websocket_streams_placeholders("/<pair>@markPrice@<updateSpeed>", &vars);
            let ws_stream = api
                .mark_price_of_all_symbols_of_a_pair(params)
                .await
                .expect("mark_price_of_all_symbols_of_a_pair should return a WebsocketStream");

            assert!(
                streams_base.is_subscribed(&stream).await,
                "expected stream '{stream}' to be subscribed"
            );
            assert_eq!(ws_stream.id, Some(StreamId::Str("test-id-123".to_string())));
        });
    }

    #[test]
    fn mark_price_of_all_symbols_of_a_pair_should_handle_incoming_message() {
        TOKIO_SHARED_RT.block_on(async {
            let (streams_base, conn) = make_streams_base().await;
            let api = WebsocketMarketStreamsApiClient::new(streams_base.clone());

            let id = "test-id-123".to_string();

            let params = MarkPriceOfAllSymbolsOfAPairParams::builder("btcusdt".to_string(),).id(Some(id.clone())).build().unwrap();

            let MarkPriceOfAllSymbolsOfAPairParams {
                pair,id,update_speed,
            } = params.clone();

            let pairs: &[(&str, Option<String>)] = &[
                ("pair",
                        Some(pair.clone())
                ),
                ("id",
                        id.clone()
                ),
                ("updateSpeed",
                        update_speed.clone()
                ),
            ];

            let vars: HashMap<_, _> = pairs
                .iter()
                .filter_map(|&(k, ref v)| v.clone().map(|v| (k, v)))
                .collect();
            let stream = replace_websocket_streams_placeholders("/<pair>@markPrice@<updateSpeed>", &vars);

            let ws_stream = api.mark_price_of_all_symbols_of_a_pair(params).await.unwrap();

            let called = Arc::new(AtomicBool::new(false));
            let called_with_message = called.clone();
            ws_stream.on_message(move |_payload: Vec<models::MarkPriceOfAllSymbolsOfAPairResponseInner>| {
                called_with_message.store(true, Ordering::SeqCst);
            });

            let payload: Value = serde_json::from_str(r#"[{"e":"markPriceUpdate","E":1596095725000,"s":"BTCUSD_201225","p":"10934.62615417","P":"10962.17178236","i":"10933.62615417","r":"","T":0},{"e":"markPriceUpdate","E":1596095725000,"s":"BTCUSD_PERP","p":"11012.31359011","P":"10962.17178236","i":"10933.62615417","r":"0.00000000","T":1596096000000}]"#).unwrap();
            let msg = json!({
                "stream": stream,
                "data": payload,
            });

            streams_base.on_message(msg.to_string(), conn.clone()).await;
            yield_now().await;

            assert!(called.load(Ordering::SeqCst), "expected our callback to have been invoked");
        });
    }

    #[test]
    fn mark_price_of_all_symbols_of_a_pair_should_not_fire_after_unsubscribe() {
        TOKIO_SHARED_RT.block_on(async {
            let (streams_base, conn) = make_streams_base().await;
            let api = WebsocketMarketStreamsApiClient::new(streams_base.clone());

            let id = "test-id-123".to_string();

            let params = MarkPriceOfAllSymbolsOfAPairParams::builder("btcusdt".to_string(),).id(Some(id.clone())).build().unwrap();

            let MarkPriceOfAllSymbolsOfAPairParams {
                pair,id,update_speed,
            } = params.clone();

            let pairs: &[(&str, Option<String>)] = &[
                ("pair",
                        Some(pair.clone())
                ),
                ("id",
                        id.clone()
                ),
                ("updateSpeed",
                        update_speed.clone()
                ),
            ];

            let vars: HashMap<_, _> = pairs
                .iter()
                .filter_map(|&(k, ref v)| v.clone().map(|v| (k, v)))
                .collect();
            let stream = replace_websocket_streams_placeholders("/<pair>@markPrice@<updateSpeed>", &vars);

            let ws_stream = api.mark_price_of_all_symbols_of_a_pair(params).await.unwrap();

            let called = Arc::new(AtomicBool::new(false));
            let called_clone = called.clone();
            ws_stream.on_message(move |_payload: Vec<models::MarkPriceOfAllSymbolsOfAPairResponseInner>| {
                called_clone.store(true, Ordering::SeqCst);
            });

            assert!(streams_base.is_subscribed(&stream).await, "should be subscribed before unsubscribe");

            ws_stream.unsubscribe().await;

            let payload: Value = serde_json::from_str(r#"[{"e":"markPriceUpdate","E":1596095725000,"s":"BTCUSD_201225","p":"10934.62615417","P":"10962.17178236","i":"10933.62615417","r":"","T":0},{"e":"markPriceUpdate","E":1596095725000,"s":"BTCUSD_PERP","p":"11012.31359011","P":"10962.17178236","i":"10933.62615417","r":"0.00000000","T":1596096000000}]"#).unwrap();
            let msg = json!({
                "stream": stream,
                "data": payload,
            });

            streams_base.on_message(msg.to_string(), conn.clone()).await;

            yield_now().await;

            assert!(!called.load(Ordering::SeqCst), "callback should not be invoked after unsubscribe");
        });
    }

    #[test]
    fn mark_price_stream_should_execute_successfully() {
        TOKIO_SHARED_RT.block_on(async {
            let (streams_base, _) = make_streams_base().await;
            let api = WebsocketMarketStreamsApiClient::new(streams_base.clone());

            let id = "test-id-123".to_string();

            let params = MarkPriceStreamParams::builder("btcusdt".to_string())
                .id(Some(id.clone()))
                .build()
                .unwrap();

            let MarkPriceStreamParams {
                symbol,
                id,
                update_speed,
            } = params.clone();

            let pairs: &[(&str, Option<String>)] = &[
                ("symbol", Some(symbol.clone())),
                ("id", id.clone()),
                ("updateSpeed", update_speed.clone()),
            ];

            let vars: HashMap<_, _> = pairs
                .iter()
                .filter_map(|&(k, ref v)| v.clone().map(|v| (k, v)))
                .collect();
            let stream =
                replace_websocket_streams_placeholders("/<symbol>@markPrice@<updateSpeed>", &vars);
            let ws_stream = api
                .mark_price_stream(params)
                .await
                .expect("mark_price_stream should return a WebsocketStream");

            assert!(
                streams_base.is_subscribed(&stream).await,
                "expected stream '{stream}' to be subscribed"
            );
            assert_eq!(ws_stream.id, Some(StreamId::Str("test-id-123".to_string())));
        });
    }

    #[test]
    fn mark_price_stream_should_handle_incoming_message() {
        TOKIO_SHARED_RT.block_on(async {
            let (streams_base, conn) = make_streams_base().await;
            let api = WebsocketMarketStreamsApiClient::new(streams_base.clone());

            let id = "test-id-123".to_string();

            let params = MarkPriceStreamParams::builder("btcusdt".to_string(),).id(Some(id.clone())).build().unwrap();

            let MarkPriceStreamParams {
                symbol,id,update_speed,
            } = params.clone();

            let pairs: &[(&str, Option<String>)] = &[
                ("symbol",
                        Some(symbol.clone())
                ),
                ("id",
                        id.clone()
                ),
                ("updateSpeed",
                        update_speed.clone()
                ),
            ];

            let vars: HashMap<_, _> = pairs
                .iter()
                .filter_map(|&(k, ref v)| v.clone().map(|v| (k, v)))
                .collect();
            let stream = replace_websocket_streams_placeholders("/<symbol>@markPrice@<updateSpeed>", &vars);

            let ws_stream = api.mark_price_stream(params).await.unwrap();

            let called = Arc::new(AtomicBool::new(false));
            let called_with_message = called.clone();
            ws_stream.on_message(move |_payload: models::MarkPriceStreamResponse| {
                called_with_message.store(true, Ordering::SeqCst);
            });

            let payload: Value = serde_json::from_str(r#"{"e":"markPriceUpdate","E":1596095725000,"s":"BTCUSD_201225","p":"10934.62615417","P":"10962.17178236","i":"10933.62615417","r":"","T":0}"#).unwrap();
            let msg = json!({
                "stream": stream,
                "data": payload,
            });

            streams_base.on_message(msg.to_string(), conn.clone()).await;
            yield_now().await;

            assert!(called.load(Ordering::SeqCst), "expected our callback to have been invoked");
        });
    }

    #[test]
    fn mark_price_stream_should_not_fire_after_unsubscribe() {
        TOKIO_SHARED_RT.block_on(async {
            let (streams_base, conn) = make_streams_base().await;
            let api = WebsocketMarketStreamsApiClient::new(streams_base.clone());

            let id = "test-id-123".to_string();

            let params = MarkPriceStreamParams::builder("btcusdt".to_string(),).id(Some(id.clone())).build().unwrap();

            let MarkPriceStreamParams {
                symbol,id,update_speed,
            } = params.clone();

            let pairs: &[(&str, Option<String>)] = &[
                ("symbol",
                        Some(symbol.clone())
                ),
                ("id",
                        id.clone()
                ),
                ("updateSpeed",
                        update_speed.clone()
                ),
            ];

            let vars: HashMap<_, _> = pairs
                .iter()
                .filter_map(|&(k, ref v)| v.clone().map(|v| (k, v)))
                .collect();
            let stream = replace_websocket_streams_placeholders("/<symbol>@markPrice@<updateSpeed>", &vars);

            let ws_stream = api.mark_price_stream(params).await.unwrap();

            let called = Arc::new(AtomicBool::new(false));
            let called_clone = called.clone();
            ws_stream.on_message(move |_payload: models::MarkPriceStreamResponse| {
                called_clone.store(true, Ordering::SeqCst);
            });

            assert!(streams_base.is_subscribed(&stream).await, "should be subscribed before unsubscribe");

            ws_stream.unsubscribe().await;

            let payload: Value = serde_json::from_str(r#"{"e":"markPriceUpdate","E":1596095725000,"s":"BTCUSD_201225","p":"10934.62615417","P":"10962.17178236","i":"10933.62615417","r":"","T":0}"#).unwrap();
            let msg = json!({
                "stream": stream,
                "data": payload,
            });

            streams_base.on_message(msg.to_string(), conn.clone()).await;

            yield_now().await;

            assert!(!called.load(Ordering::SeqCst), "callback should not be invoked after unsubscribe");
        });
    }

    #[test]
    fn partial_book_depth_streams_should_execute_successfully() {
        TOKIO_SHARED_RT.block_on(async {
            let (streams_base, _) = make_streams_base().await;
            let api = WebsocketMarketStreamsApiClient::new(streams_base.clone());

            let id = "test-id-123".to_string();

            let params = PartialBookDepthStreamsParams::builder("btcusdt".to_string(), 10)
                .id(Some(id.clone()))
                .build()
                .unwrap();

            let PartialBookDepthStreamsParams {
                symbol,
                levels,
                id,
                update_speed,
            } = params.clone();

            let pairs: &[(&str, Option<String>)] = &[
                ("symbol", Some(symbol.clone())),
                ("levels", Some(levels.to_string())),
                ("id", id.clone()),
                ("updateSpeed", update_speed.clone()),
            ];

            let vars: HashMap<_, _> = pairs
                .iter()
                .filter_map(|&(k, ref v)| v.clone().map(|v| (k, v)))
                .collect();
            let stream = replace_websocket_streams_placeholders(
                "/<symbol>@depth<levels>@<updateSpeed>",
                &vars,
            );
            let ws_stream = api
                .partial_book_depth_streams(params)
                .await
                .expect("partial_book_depth_streams should return a WebsocketStream");

            assert!(
                streams_base.is_subscribed(&stream).await,
                "expected stream '{stream}' to be subscribed"
            );
            assert_eq!(ws_stream.id, Some(StreamId::Str("test-id-123".to_string())));
        });
    }

    #[test]
    fn partial_book_depth_streams_should_handle_incoming_message() {
        TOKIO_SHARED_RT.block_on(async {
            let (streams_base, conn) = make_streams_base().await;
            let api = WebsocketMarketStreamsApiClient::new(streams_base.clone());

            let id = "test-id-123".to_string();

            let params = PartialBookDepthStreamsParams::builder("btcusdt".to_string(),10,).id(Some(id.clone())).build().unwrap();

            let PartialBookDepthStreamsParams {
                symbol,levels,id,update_speed,
            } = params.clone();

            let pairs: &[(&str, Option<String>)] = &[
                ("symbol",
                        Some(symbol.clone())
                ),
                ("levels",
                        Some(levels.to_string().to_string())
                ),
                ("id",
                        id.clone()
                ),
                ("updateSpeed",
                        update_speed.clone()
                ),
            ];

            let vars: HashMap<_, _> = pairs
                .iter()
                .filter_map(|&(k, ref v)| v.clone().map(|v| (k, v)))
                .collect();
            let stream = replace_websocket_streams_placeholders("/<symbol>@depth<levels>@<updateSpeed>", &vars);

            let ws_stream = api.partial_book_depth_streams(params).await.unwrap();

            let called = Arc::new(AtomicBool::new(false));
            let called_with_message = called.clone();
            ws_stream.on_message(move |_payload: models::PartialBookDepthStreamsResponse| {
                called_with_message.store(true, Ordering::SeqCst);
            });

            let payload: Value = serde_json::from_str(r#"{"e":"depthUpdate","E":1591269996801,"T":1591269996646,"s":"BTCUSD_200626","ps":"BTCUSD","U":17276694,"u":17276701,"pu":17276678,"b":[["9523.0","5"],["9522.8","8"],["9522.6","2"],["9522.4","1"],["9522.0","5"]],"a":[["9524.6","2"],["9524.7","3"],["9524.9","16"],["9525.1","10"],["9525.3","6"]]}"#).unwrap();
            let msg = json!({
                "stream": stream,
                "data": payload,
            });

            streams_base.on_message(msg.to_string(), conn.clone()).await;
            yield_now().await;

            assert!(called.load(Ordering::SeqCst), "expected our callback to have been invoked");
        });
    }

    #[test]
    fn partial_book_depth_streams_should_not_fire_after_unsubscribe() {
        TOKIO_SHARED_RT.block_on(async {
            let (streams_base, conn) = make_streams_base().await;
            let api = WebsocketMarketStreamsApiClient::new(streams_base.clone());

            let id = "test-id-123".to_string();

            let params = PartialBookDepthStreamsParams::builder("btcusdt".to_string(),10,).id(Some(id.clone())).build().unwrap();

            let PartialBookDepthStreamsParams {
                symbol,levels,id,update_speed,
            } = params.clone();

            let pairs: &[(&str, Option<String>)] = &[
                ("symbol",
                        Some(symbol.clone())
                ),
                ("levels",
                        Some(levels.to_string().to_string())
                ),
                ("id",
                        id.clone()
                ),
                ("updateSpeed",
                        update_speed.clone()
                ),
            ];

            let vars: HashMap<_, _> = pairs
                .iter()
                .filter_map(|&(k, ref v)| v.clone().map(|v| (k, v)))
                .collect();
            let stream = replace_websocket_streams_placeholders("/<symbol>@depth<levels>@<updateSpeed>", &vars);

            let ws_stream = api.partial_book_depth_streams(params).await.unwrap();

            let called = Arc::new(AtomicBool::new(false));
            let called_clone = called.clone();
            ws_stream.on_message(move |_payload: models::PartialBookDepthStreamsResponse| {
                called_clone.store(true, Ordering::SeqCst);
            });

            assert!(streams_base.is_subscribed(&stream).await, "should be subscribed before unsubscribe");

            ws_stream.unsubscribe().await;

            let payload: Value = serde_json::from_str(r#"{"e":"depthUpdate","E":1591269996801,"T":1591269996646,"s":"BTCUSD_200626","ps":"BTCUSD","U":17276694,"u":17276701,"pu":17276678,"b":[["9523.0","5"],["9522.8","8"],["9522.6","2"],["9522.4","1"],["9522.0","5"]],"a":[["9524.6","2"],["9524.7","3"],["9524.9","16"],["9525.1","10"],["9525.3","6"]]}"#).unwrap();
            let msg = json!({
                "stream": stream,
                "data": payload,
            });

            streams_base.on_message(msg.to_string(), conn.clone()).await;

            yield_now().await;

            assert!(!called.load(Ordering::SeqCst), "callback should not be invoked after unsubscribe");
        });
    }
}
