binance-sdk 61.0.0

/*
 * Futures (COIN-M) WebSocket Market Streams
 *
 * Access market data, manage accounts, and trade COIN-M perpetual and delivery futures.
 *
 * 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 Api: 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 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 market_liquidation_order_streams(
        &self,
        params: MarketLiquidationOrderStreamsParams,
    ) -> anyhow::Result<Arc<WebsocketStream<models::MarketLiquidationOrderStreamsResponse>>>;
    async fn partial_book_depth_streams(
        &self,
        params: PartialBookDepthStreamsParams,
    ) -> anyhow::Result<Arc<WebsocketStream<models::PartialBookDepthStreamsResponse>>>;
}

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

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

#[allow(non_camel_case_types)]
#[derive(Debug, Clone, Serialize, Deserialize)]
pub enum ContinuousContractKlineCandlestickStreamsContractTypeEnum {
    #[serde(rename = "perpetual")]
    Perpetual,
    #[serde(rename = "current_quarter")]
    CurrentQuarter,
    #[serde(rename = "next_quarter")]
    NextQuarter,
}

impl ContinuousContractKlineCandlestickStreamsContractTypeEnum {
    #[must_use]
    pub fn as_str(&self) -> &'static str {
        match self {
            Self::Perpetual => "perpetual",
            Self::CurrentQuarter => "current_quarter",
            Self::NextQuarter => "next_quarter",
        }
    }
}

impl std::str::FromStr for ContinuousContractKlineCandlestickStreamsContractTypeEnum {
    type Err = Box<dyn std::error::Error + Send + Sync>;

    fn from_str(s: &str) -> Result<Self, Self::Err> {
        match s {
            "perpetual" => Ok(Self::Perpetual),
            "current_quarter" => Ok(Self::CurrentQuarter),
            "next_quarter" => Ok(Self::NextQuarter),
            other => Err(format!(
                "invalid ContinuousContractKlineCandlestickStreamsContractTypeEnum: {}",
                other
            )
            .into()),
        }
    }
}

#[allow(non_camel_case_types)]
#[derive(Debug, Clone, Serialize, Deserialize)]
pub enum ContinuousContractKlineCandlestickStreamsIntervalEnum {
    #[serde(rename = "1m")]
    Interval1m,
    #[serde(rename = "3m")]
    Interval3m,
    #[serde(rename = "5m")]
    Interval5m,
    #[serde(rename = "15m")]
    Interval15m,
    #[serde(rename = "30m")]
    Interval30m,
    #[serde(rename = "1h")]
    Interval1h,
    #[serde(rename = "2h")]
    Interval2h,
    #[serde(rename = "4h")]
    Interval4h,
    #[serde(rename = "6h")]
    Interval6h,
    #[serde(rename = "8h")]
    Interval8h,
    #[serde(rename = "12h")]
    Interval12h,
    #[serde(rename = "1d")]
    Interval1d,
    #[serde(rename = "3d")]
    Interval3d,
    #[serde(rename = "1w")]
    Interval1w,
    #[serde(rename = "1M")]
    Interval1M,
}

impl ContinuousContractKlineCandlestickStreamsIntervalEnum {
    #[must_use]
    pub fn as_str(&self) -> &'static str {
        match self {
            Self::Interval1m => "1m",
            Self::Interval3m => "3m",
            Self::Interval5m => "5m",
            Self::Interval15m => "15m",
            Self::Interval30m => "30m",
            Self::Interval1h => "1h",
            Self::Interval2h => "2h",
            Self::Interval4h => "4h",
            Self::Interval6h => "6h",
            Self::Interval8h => "8h",
            Self::Interval12h => "12h",
            Self::Interval1d => "1d",
            Self::Interval3d => "3d",
            Self::Interval1w => "1w",
            Self::Interval1M => "1M",
        }
    }
}

impl std::str::FromStr for ContinuousContractKlineCandlestickStreamsIntervalEnum {
    type Err = Box<dyn std::error::Error + Send + Sync>;

    fn from_str(s: &str) -> Result<Self, Self::Err> {
        match s {
            "1m" => Ok(Self::Interval1m),
            "3m" => Ok(Self::Interval3m),
            "5m" => Ok(Self::Interval5m),
            "15m" => Ok(Self::Interval15m),
            "30m" => Ok(Self::Interval30m),
            "1h" => Ok(Self::Interval1h),
            "2h" => Ok(Self::Interval2h),
            "4h" => Ok(Self::Interval4h),
            "6h" => Ok(Self::Interval6h),
            "8h" => Ok(Self::Interval8h),
            "12h" => Ok(Self::Interval12h),
            "1d" => Ok(Self::Interval1d),
            "3d" => Ok(Self::Interval3d),
            "1w" => Ok(Self::Interval1w),
            "1M" => Ok(Self::Interval1M),
            other => Err(format!(
                "invalid ContinuousContractKlineCandlestickStreamsIntervalEnum: {}",
                other
            )
            .into()),
        }
    }
}

#[allow(non_camel_case_types)]
#[derive(Debug, Clone, Serialize, Deserialize)]
pub enum DiffBookDepthStreamsUpdateSpeedEnum {
    #[serde(rename = "100ms")]
    UpdateSpeed100ms,
    #[serde(rename = "500ms")]
    UpdateSpeed500ms,
}

impl DiffBookDepthStreamsUpdateSpeedEnum {
    #[must_use]
    pub fn as_str(&self) -> &'static str {
        match self {
            Self::UpdateSpeed100ms => "100ms",
            Self::UpdateSpeed500ms => "500ms",
        }
    }
}

impl std::str::FromStr for DiffBookDepthStreamsUpdateSpeedEnum {
    type Err = Box<dyn std::error::Error + Send + Sync>;

    fn from_str(s: &str) -> Result<Self, Self::Err> {
        match s {
            "100ms" => Ok(Self::UpdateSpeed100ms),
            "500ms" => Ok(Self::UpdateSpeed500ms),
            other => Err(format!("invalid DiffBookDepthStreamsUpdateSpeedEnum: {}", other).into()),
        }
    }
}

#[allow(non_camel_case_types)]
#[derive(Debug, Clone, Serialize, Deserialize)]
pub enum IndexKlineCandlestickStreamsIntervalEnum {
    #[serde(rename = "1m")]
    Interval1m,
    #[serde(rename = "3m")]
    Interval3m,
    #[serde(rename = "5m")]
    Interval5m,
    #[serde(rename = "15m")]
    Interval15m,
    #[serde(rename = "30m")]
    Interval30m,
    #[serde(rename = "1h")]
    Interval1h,
    #[serde(rename = "2h")]
    Interval2h,
    #[serde(rename = "4h")]
    Interval4h,
    #[serde(rename = "6h")]
    Interval6h,
    #[serde(rename = "8h")]
    Interval8h,
    #[serde(rename = "12h")]
    Interval12h,
    #[serde(rename = "1d")]
    Interval1d,
    #[serde(rename = "3d")]
    Interval3d,
    #[serde(rename = "1w")]
    Interval1w,
    #[serde(rename = "1M")]
    Interval1M,
}

impl IndexKlineCandlestickStreamsIntervalEnum {
    #[must_use]
    pub fn as_str(&self) -> &'static str {
        match self {
            Self::Interval1m => "1m",
            Self::Interval3m => "3m",
            Self::Interval5m => "5m",
            Self::Interval15m => "15m",
            Self::Interval30m => "30m",
            Self::Interval1h => "1h",
            Self::Interval2h => "2h",
            Self::Interval4h => "4h",
            Self::Interval6h => "6h",
            Self::Interval8h => "8h",
            Self::Interval12h => "12h",
            Self::Interval1d => "1d",
            Self::Interval3d => "3d",
            Self::Interval1w => "1w",
            Self::Interval1M => "1M",
        }
    }
}

impl std::str::FromStr for IndexKlineCandlestickStreamsIntervalEnum {
    type Err = Box<dyn std::error::Error + Send + Sync>;

    fn from_str(s: &str) -> Result<Self, Self::Err> {
        match s {
            "1m" => Ok(Self::Interval1m),
            "3m" => Ok(Self::Interval3m),
            "5m" => Ok(Self::Interval5m),
            "15m" => Ok(Self::Interval15m),
            "30m" => Ok(Self::Interval30m),
            "1h" => Ok(Self::Interval1h),
            "2h" => Ok(Self::Interval2h),
            "4h" => Ok(Self::Interval4h),
            "6h" => Ok(Self::Interval6h),
            "8h" => Ok(Self::Interval8h),
            "12h" => Ok(Self::Interval12h),
            "1d" => Ok(Self::Interval1d),
            "3d" => Ok(Self::Interval3d),
            "1w" => Ok(Self::Interval1w),
            "1M" => Ok(Self::Interval1M),
            other => Err(format!(
                "invalid IndexKlineCandlestickStreamsIntervalEnum: {}",
                other
            )
            .into()),
        }
    }
}

#[allow(non_camel_case_types)]
#[derive(Debug, Clone, Serialize, Deserialize)]
pub enum IndexPriceStreamUpdateSpeedEnum {
    #[serde(rename = "1s")]
    UpdateSpeed1s,
}

impl IndexPriceStreamUpdateSpeedEnum {
    #[must_use]
    pub fn as_str(&self) -> &'static str {
        match self {
            Self::UpdateSpeed1s => "1s",
        }
    }
}

impl std::str::FromStr for IndexPriceStreamUpdateSpeedEnum {
    type Err = Box<dyn std::error::Error + Send + Sync>;

    fn from_str(s: &str) -> Result<Self, Self::Err> {
        match s {
            "1s" => Ok(Self::UpdateSpeed1s),
            other => Err(format!("invalid IndexPriceStreamUpdateSpeedEnum: {}", other).into()),
        }
    }
}

#[allow(non_camel_case_types)]
#[derive(Debug, Clone, Serialize, Deserialize)]
pub enum KlineCandlestickStreamsIntervalEnum {
    #[serde(rename = "1m")]
    Interval1m,
    #[serde(rename = "3m")]
    Interval3m,
    #[serde(rename = "5m")]
    Interval5m,
    #[serde(rename = "15m")]
    Interval15m,
    #[serde(rename = "30m")]
    Interval30m,
    #[serde(rename = "1h")]
    Interval1h,
    #[serde(rename = "2h")]
    Interval2h,
    #[serde(rename = "4h")]
    Interval4h,
    #[serde(rename = "6h")]
    Interval6h,
    #[serde(rename = "8h")]
    Interval8h,
    #[serde(rename = "12h")]
    Interval12h,
    #[serde(rename = "1d")]
    Interval1d,
    #[serde(rename = "3d")]
    Interval3d,
    #[serde(rename = "1w")]
    Interval1w,
    #[serde(rename = "1M")]
    Interval1M,
}

impl KlineCandlestickStreamsIntervalEnum {
    #[must_use]
    pub fn as_str(&self) -> &'static str {
        match self {
            Self::Interval1m => "1m",
            Self::Interval3m => "3m",
            Self::Interval5m => "5m",
            Self::Interval15m => "15m",
            Self::Interval30m => "30m",
            Self::Interval1h => "1h",
            Self::Interval2h => "2h",
            Self::Interval4h => "4h",
            Self::Interval6h => "6h",
            Self::Interval8h => "8h",
            Self::Interval12h => "12h",
            Self::Interval1d => "1d",
            Self::Interval3d => "3d",
            Self::Interval1w => "1w",
            Self::Interval1M => "1M",
        }
    }
}

impl std::str::FromStr for KlineCandlestickStreamsIntervalEnum {
    type Err = Box<dyn std::error::Error + Send + Sync>;

    fn from_str(s: &str) -> Result<Self, Self::Err> {
        match s {
            "1m" => Ok(Self::Interval1m),
            "3m" => Ok(Self::Interval3m),
            "5m" => Ok(Self::Interval5m),
            "15m" => Ok(Self::Interval15m),
            "30m" => Ok(Self::Interval30m),
            "1h" => Ok(Self::Interval1h),
            "2h" => Ok(Self::Interval2h),
            "4h" => Ok(Self::Interval4h),
            "6h" => Ok(Self::Interval6h),
            "8h" => Ok(Self::Interval8h),
            "12h" => Ok(Self::Interval12h),
            "1d" => Ok(Self::Interval1d),
            "3d" => Ok(Self::Interval3d),
            "1w" => Ok(Self::Interval1w),
            "1M" => Ok(Self::Interval1M),
            other => Err(format!("invalid KlineCandlestickStreamsIntervalEnum: {}", other).into()),
        }
    }
}

#[allow(non_camel_case_types)]
#[derive(Debug, Clone, Serialize, Deserialize)]
pub enum MarkPriceKlineCandlestickStreamsIntervalEnum {
    #[serde(rename = "1m")]
    Interval1m,
    #[serde(rename = "3m")]
    Interval3m,
    #[serde(rename = "5m")]
    Interval5m,
    #[serde(rename = "15m")]
    Interval15m,
    #[serde(rename = "30m")]
    Interval30m,
    #[serde(rename = "1h")]
    Interval1h,
    #[serde(rename = "2h")]
    Interval2h,
    #[serde(rename = "4h")]
    Interval4h,
    #[serde(rename = "6h")]
    Interval6h,
    #[serde(rename = "8h")]
    Interval8h,
    #[serde(rename = "12h")]
    Interval12h,
    #[serde(rename = "1d")]
    Interval1d,
    #[serde(rename = "3d")]
    Interval3d,
    #[serde(rename = "1w")]
    Interval1w,
    #[serde(rename = "1M")]
    Interval1M,
}

impl MarkPriceKlineCandlestickStreamsIntervalEnum {
    #[must_use]
    pub fn as_str(&self) -> &'static str {
        match self {
            Self::Interval1m => "1m",
            Self::Interval3m => "3m",
            Self::Interval5m => "5m",
            Self::Interval15m => "15m",
            Self::Interval30m => "30m",
            Self::Interval1h => "1h",
            Self::Interval2h => "2h",
            Self::Interval4h => "4h",
            Self::Interval6h => "6h",
            Self::Interval8h => "8h",
            Self::Interval12h => "12h",
            Self::Interval1d => "1d",
            Self::Interval3d => "3d",
            Self::Interval1w => "1w",
            Self::Interval1M => "1M",
        }
    }
}

impl std::str::FromStr for MarkPriceKlineCandlestickStreamsIntervalEnum {
    type Err = Box<dyn std::error::Error + Send + Sync>;

    fn from_str(s: &str) -> Result<Self, Self::Err> {
        match s {
            "1m" => Ok(Self::Interval1m),
            "3m" => Ok(Self::Interval3m),
            "5m" => Ok(Self::Interval5m),
            "15m" => Ok(Self::Interval15m),
            "30m" => Ok(Self::Interval30m),
            "1h" => Ok(Self::Interval1h),
            "2h" => Ok(Self::Interval2h),
            "4h" => Ok(Self::Interval4h),
            "6h" => Ok(Self::Interval6h),
            "8h" => Ok(Self::Interval8h),
            "12h" => Ok(Self::Interval12h),
            "1d" => Ok(Self::Interval1d),
            "3d" => Ok(Self::Interval3d),
            "1w" => Ok(Self::Interval1w),
            "1M" => Ok(Self::Interval1M),
            other => Err(format!(
                "invalid MarkPriceKlineCandlestickStreamsIntervalEnum: {}",
                other
            )
            .into()),
        }
    }
}

#[allow(non_camel_case_types)]
#[derive(Debug, Clone, Serialize, Deserialize)]
pub enum MarkPriceOfAllSymbolsOfAPairUpdateSpeedEnum {
    #[serde(rename = "1s")]
    UpdateSpeed1s,
}

impl MarkPriceOfAllSymbolsOfAPairUpdateSpeedEnum {
    #[must_use]
    pub fn as_str(&self) -> &'static str {
        match self {
            Self::UpdateSpeed1s => "1s",
        }
    }
}

impl std::str::FromStr for MarkPriceOfAllSymbolsOfAPairUpdateSpeedEnum {
    type Err = Box<dyn std::error::Error + Send + Sync>;

    fn from_str(s: &str) -> Result<Self, Self::Err> {
        match s {
            "1s" => Ok(Self::UpdateSpeed1s),
            other => Err(format!(
                "invalid MarkPriceOfAllSymbolsOfAPairUpdateSpeedEnum: {}",
                other
            )
            .into()),
        }
    }
}

#[allow(non_camel_case_types)]
#[derive(Debug, Clone, Serialize, Deserialize)]
pub enum MarkPriceStreamUpdateSpeedEnum {
    #[serde(rename = "1s")]
    UpdateSpeed1s,
}

impl MarkPriceStreamUpdateSpeedEnum {
    #[must_use]
    pub fn as_str(&self) -> &'static str {
        match self {
            Self::UpdateSpeed1s => "1s",
        }
    }
}

impl std::str::FromStr for MarkPriceStreamUpdateSpeedEnum {
    type Err = Box<dyn std::error::Error + Send + Sync>;

    fn from_str(s: &str) -> Result<Self, Self::Err> {
        match s {
            "1s" => Ok(Self::UpdateSpeed1s),
            other => Err(format!("invalid MarkPriceStreamUpdateSpeedEnum: {}", other).into()),
        }
    }
}

#[allow(non_camel_case_types)]
#[derive(Debug, Clone, Serialize, Deserialize)]
pub enum PartialBookDepthStreamsLevelsEnum {
    #[serde(rename = "5")]
    Levels5,
    #[serde(rename = "10")]
    Levels10,
    #[serde(rename = "20")]
    Levels20,
}

impl PartialBookDepthStreamsLevelsEnum {
    #[must_use]
    pub fn as_str(&self) -> &'static str {
        match self {
            Self::Levels5 => "5",
            Self::Levels10 => "10",
            Self::Levels20 => "20",
        }
    }
}

impl std::str::FromStr for PartialBookDepthStreamsLevelsEnum {
    type Err = Box<dyn std::error::Error + Send + Sync>;

    fn from_str(s: &str) -> Result<Self, Self::Err> {
        match s {
            "5" => Ok(Self::Levels5),
            "10" => Ok(Self::Levels10),
            "20" => Ok(Self::Levels20),
            other => Err(format!("invalid PartialBookDepthStreamsLevelsEnum: {}", other).into()),
        }
    }
}

#[allow(non_camel_case_types)]
#[derive(Debug, Clone, Serialize, Deserialize)]
pub enum PartialBookDepthStreamsUpdateSpeedEnum {
    #[serde(rename = "100ms")]
    UpdateSpeed100ms,
    #[serde(rename = "500ms")]
    UpdateSpeed500ms,
}

impl PartialBookDepthStreamsUpdateSpeedEnum {
    #[must_use]
    pub fn as_str(&self) -> &'static str {
        match self {
            Self::UpdateSpeed100ms => "100ms",
            Self::UpdateSpeed500ms => "500ms",
        }
    }
}

impl std::str::FromStr for PartialBookDepthStreamsUpdateSpeedEnum {
    type Err = Box<dyn std::error::Error + Send + Sync>;

    fn from_str(s: &str) -> Result<Self, Self::Err> {
        match s {
            "100ms" => Ok(Self::UpdateSpeed100ms),
            "500ms" => Ok(Self::UpdateSpeed500ms),
            other => {
                Err(format!("invalid PartialBookDepthStreamsUpdateSpeedEnum: {}", other).into())
            }
        }
    }
}

/// 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, Deserialize)]
#[builder(pattern = "owned", build_fn(error = "ParamBuildError"))]
pub struct AggregateTradeStreamsParams {
    /// The symbol parameter
    ///
    /// This field is **required.
    #[builder(setter(into))]
    #[serde(rename = "symbol")]
    pub symbol: String,
    /// Unique WebSocket request ID.
    ///
    /// This field is **optional.
    #[builder(setter(into), default)]
    #[serde(rename = "id", 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, Deserialize, Default)]
#[builder(pattern = "owned", build_fn(error = "ParamBuildError"))]
pub struct AllBookTickersStreamParams {
    /// Unique WebSocket request ID.
    ///
    /// This field is **optional.
    #[builder(setter(into), default)]
    #[serde(rename = "id", 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, Deserialize, Default)]
#[builder(pattern = "owned", build_fn(error = "ParamBuildError"))]
pub struct AllMarketLiquidationOrderStreamsParams {
    /// Unique WebSocket request ID.
    ///
    /// This field is **optional.
    #[builder(setter(into), default)]
    #[serde(rename = "id", 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, Deserialize, Default)]
#[builder(pattern = "owned", build_fn(error = "ParamBuildError"))]
pub struct AllMarketMiniTickersStreamParams {
    /// Unique WebSocket request ID.
    ///
    /// This field is **optional.
    #[builder(setter(into), default)]
    #[serde(rename = "id", 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, Deserialize, Default)]
#[builder(pattern = "owned", build_fn(error = "ParamBuildError"))]
pub struct AllMarketTickersStreamsParams {
    /// Unique WebSocket request ID.
    ///
    /// This field is **optional.
    #[builder(setter(into), default)]
    #[serde(rename = "id", 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, Deserialize)]
#[builder(pattern = "owned", build_fn(error = "ParamBuildError"))]
pub struct ContinuousContractKlineCandlestickStreamsParams {
    /// The pair parameter
    ///
    /// This field is **required.
    #[builder(setter(into))]
    #[serde(rename = "pair")]
    pub pair: String,
    /// The contractType parameter
    ///
    /// This field is **required.
    #[builder(setter(into))]
    #[serde(rename = "contractType")]
    pub contract_type: ContinuousContractKlineCandlestickStreamsContractTypeEnum,
    /// The interval parameter
    ///
    /// This field is **required.
    #[builder(setter(into))]
    #[serde(rename = "interval")]
    pub interval: ContinuousContractKlineCandlestickStreamsIntervalEnum,
    /// Unique WebSocket request ID.
    ///
    /// This field is **optional.
    #[builder(setter(into), default)]
    #[serde(rename = "id", 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: ContinuousContractKlineCandlestickStreamsContractTypeEnum,
        interval: ContinuousContractKlineCandlestickStreamsIntervalEnum,
    ) -> 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, Deserialize, Default)]
#[builder(pattern = "owned", build_fn(error = "ParamBuildError"))]
pub struct ContractInfoStreamParams {
    /// Unique WebSocket request ID.
    ///
    /// This field is **optional.
    #[builder(setter(into), default)]
    #[serde(rename = "id", 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, Deserialize)]
#[builder(pattern = "owned", build_fn(error = "ParamBuildError"))]
pub struct DiffBookDepthStreamsParams {
    /// The symbol parameter
    ///
    /// This field is **required.
    #[builder(setter(into))]
    #[serde(rename = "symbol")]
    pub symbol: String,
    /// Unique WebSocket request ID.
    ///
    /// This field is **optional.
    #[builder(setter(into), default)]
    #[serde(rename = "id", default)]
    pub id: Option<String>,
    /// WebSocket stream update speed
    ///
    /// This field is **optional.
    #[builder(setter(into), default)]
    #[serde(rename = "updateSpeed", default)]
    pub update_speed: Option<DiffBookDepthStreamsUpdateSpeedEnum>,
}

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, Deserialize)]
#[builder(pattern = "owned", build_fn(error = "ParamBuildError"))]
pub struct IndexKlineCandlestickStreamsParams {
    /// The pair parameter
    ///
    /// This field is **required.
    #[builder(setter(into))]
    #[serde(rename = "pair")]
    pub pair: String,
    /// The interval parameter
    ///
    /// This field is **required.
    #[builder(setter(into))]
    #[serde(rename = "interval")]
    pub interval: IndexKlineCandlestickStreamsIntervalEnum,
    /// Unique WebSocket request ID.
    ///
    /// This field is **optional.
    #[builder(setter(into), default)]
    #[serde(rename = "id", 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: IndexKlineCandlestickStreamsIntervalEnum,
    ) -> 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, Deserialize)]
#[builder(pattern = "owned", build_fn(error = "ParamBuildError"))]
pub struct IndexPriceStreamParams {
    /// The pair parameter
    ///
    /// This field is **required.
    #[builder(setter(into))]
    #[serde(rename = "pair")]
    pub pair: String,
    /// Unique WebSocket request ID.
    ///
    /// This field is **optional.
    #[builder(setter(into), default)]
    #[serde(rename = "id", default)]
    pub id: Option<String>,
    /// WebSocket stream update speed
    ///
    /// This field is **optional.
    #[builder(setter(into), default)]
    #[serde(rename = "updateSpeed", default)]
    pub update_speed: Option<IndexPriceStreamUpdateSpeedEnum>,
}

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, Deserialize)]
#[builder(pattern = "owned", build_fn(error = "ParamBuildError"))]
pub struct IndividualSymbolBookTickerStreamsParams {
    /// The symbol parameter
    ///
    /// This field is **required.
    #[builder(setter(into))]
    #[serde(rename = "symbol")]
    pub symbol: String,
    /// Unique WebSocket request ID.
    ///
    /// This field is **optional.
    #[builder(setter(into), default)]
    #[serde(rename = "id", 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, Deserialize)]
#[builder(pattern = "owned", build_fn(error = "ParamBuildError"))]
pub struct IndividualSymbolMiniTickerStreamParams {
    /// The symbol parameter
    ///
    /// This field is **required.
    #[builder(setter(into))]
    #[serde(rename = "symbol")]
    pub symbol: String,
    /// Unique WebSocket request ID.
    ///
    /// This field is **optional.
    #[builder(setter(into), default)]
    #[serde(rename = "id", 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, Deserialize)]
#[builder(pattern = "owned", build_fn(error = "ParamBuildError"))]
pub struct IndividualSymbolTickerStreamsParams {
    /// The symbol parameter
    ///
    /// This field is **required.
    #[builder(setter(into))]
    #[serde(rename = "symbol")]
    pub symbol: String,
    /// Unique WebSocket request ID.
    ///
    /// This field is **optional.
    #[builder(setter(into), default)]
    #[serde(rename = "id", 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, Deserialize)]
#[builder(pattern = "owned", build_fn(error = "ParamBuildError"))]
pub struct KlineCandlestickStreamsParams {
    /// The symbol parameter
    ///
    /// This field is **required.
    #[builder(setter(into))]
    #[serde(rename = "symbol")]
    pub symbol: String,
    /// The interval parameter
    ///
    /// This field is **required.
    #[builder(setter(into))]
    #[serde(rename = "interval")]
    pub interval: KlineCandlestickStreamsIntervalEnum,
    /// Unique WebSocket request ID.
    ///
    /// This field is **optional.
    #[builder(setter(into), default)]
    #[serde(rename = "id", 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: KlineCandlestickStreamsIntervalEnum,
    ) -> KlineCandlestickStreamsParamsBuilder {
        KlineCandlestickStreamsParamsBuilder::default()
            .symbol(symbol)
            .interval(interval)
    }
}
/// 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, Deserialize)]
#[builder(pattern = "owned", build_fn(error = "ParamBuildError"))]
pub struct MarkPriceKlineCandlestickStreamsParams {
    /// The symbol parameter
    ///
    /// This field is **required.
    #[builder(setter(into))]
    #[serde(rename = "symbol")]
    pub symbol: String,
    /// The interval parameter
    ///
    /// This field is **required.
    #[builder(setter(into))]
    #[serde(rename = "interval")]
    pub interval: MarkPriceKlineCandlestickStreamsIntervalEnum,
    /// Unique WebSocket request ID.
    ///
    /// This field is **optional.
    #[builder(setter(into), default)]
    #[serde(rename = "id", 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: MarkPriceKlineCandlestickStreamsIntervalEnum,
    ) -> 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, Deserialize)]
#[builder(pattern = "owned", build_fn(error = "ParamBuildError"))]
pub struct MarkPriceOfAllSymbolsOfAPairParams {
    /// The pair parameter
    ///
    /// This field is **required.
    #[builder(setter(into))]
    #[serde(rename = "pair")]
    pub pair: String,
    /// Unique WebSocket request ID.
    ///
    /// This field is **optional.
    #[builder(setter(into), default)]
    #[serde(rename = "id", default)]
    pub id: Option<String>,
    /// WebSocket stream update speed
    ///
    /// This field is **optional.
    #[builder(setter(into), default)]
    #[serde(rename = "updateSpeed", default)]
    pub update_speed: Option<MarkPriceOfAllSymbolsOfAPairUpdateSpeedEnum>,
}

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, Deserialize)]
#[builder(pattern = "owned", build_fn(error = "ParamBuildError"))]
pub struct MarkPriceStreamParams {
    /// The symbol parameter
    ///
    /// This field is **required.
    #[builder(setter(into))]
    #[serde(rename = "symbol")]
    pub symbol: String,
    /// Unique WebSocket request ID.
    ///
    /// This field is **optional.
    #[builder(setter(into), default)]
    #[serde(rename = "id", default)]
    pub id: Option<String>,
    /// WebSocket stream update speed
    ///
    /// This field is **optional.
    #[builder(setter(into), default)]
    #[serde(rename = "updateSpeed", default)]
    pub update_speed: Option<MarkPriceStreamUpdateSpeedEnum>,
}

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 [`market_liquidation_order_streams`] operation.
///
/// This struct holds all of the inputs you can pass when calling
/// [`market_liquidation_order_streams`](#method.market_liquidation_order_streams).
#[derive(Clone, Debug, Builder, Deserialize)]
#[builder(pattern = "owned", build_fn(error = "ParamBuildError"))]
pub struct MarketLiquidationOrderStreamsParams {
    /// The symbol parameter
    ///
    /// This field is **required.
    #[builder(setter(into))]
    #[serde(rename = "symbol")]
    pub symbol: String,
    /// Unique WebSocket request ID.
    ///
    /// This field is **optional.
    #[builder(setter(into), default)]
    #[serde(rename = "id", default)]
    pub id: Option<String>,
}

impl MarketLiquidationOrderStreamsParams {
    /// Create a builder for [`market_liquidation_order_streams`].
    ///
    /// Required parameters:
    ///
    /// * `symbol` — The symbol parameter
    ///
    #[must_use]
    pub fn builder(symbol: String) -> MarketLiquidationOrderStreamsParamsBuilder {
        MarketLiquidationOrderStreamsParamsBuilder::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, Deserialize)]
#[builder(pattern = "owned", build_fn(error = "ParamBuildError"))]
pub struct PartialBookDepthStreamsParams {
    /// The symbol parameter
    ///
    /// This field is **required.
    #[builder(setter(into))]
    #[serde(rename = "symbol")]
    pub symbol: String,
    /// The levels parameter
    ///
    /// This field is **required.
    #[builder(setter(into))]
    #[serde(rename = "levels")]
    pub levels: PartialBookDepthStreamsLevelsEnum,
    /// Unique WebSocket request ID.
    ///
    /// This field is **optional.
    #[builder(setter(into), default)]
    #[serde(rename = "id", default)]
    pub id: Option<String>,
    /// WebSocket stream update speed
    ///
    /// This field is **optional.
    #[builder(setter(into), default)]
    #[serde(rename = "updateSpeed", default)]
    pub update_speed: Option<PartialBookDepthStreamsUpdateSpeedEnum>,
}

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: PartialBookDepthStreamsLevelsEnum,
    ) -> PartialBookDepthStreamsParamsBuilder {
        PartialBookDepthStreamsParamsBuilder::default()
            .symbol(symbol)
            .levels(levels)
    }
}

#[async_trait]
impl Api for ApiClient {
    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").cloned();

        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").cloned();

        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").cloned();

        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").cloned();

        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").cloned();

        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.as_str().to_string())),
            ("interval", Some(interval.as_str().to_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").cloned();

        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").cloned();

        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().map(|v| v.as_str().to_string()),
            ),
        ];

        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").cloned();

        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.as_str().to_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").cloned();

        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().map(|v| v.as_str().to_string()),
            ),
        ];

        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").cloned();

        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").cloned();

        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").cloned();

        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").cloned();

        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.as_str().to_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").cloned();

        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 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.as_str().to_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").cloned();

        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().map(|v| v.as_str().to_string()),
            ),
        ];

        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").cloned();

        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().map(|v| v.as_str().to_string()),
            ),
        ];

        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").cloned();

        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 market_liquidation_order_streams(
        &self,
        params: MarketLiquidationOrderStreamsParams,
    ) -> anyhow::Result<Arc<WebsocketStream<models::MarketLiquidationOrderStreamsResponse>>> {
        let MarketLiquidationOrderStreamsParams { 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").cloned();

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

        Ok(
            create_stream_handler::<models::MarketLiquidationOrderStreamsResponse>(
                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.as_str().to_string())),
            ("id", id.clone()),
            (
                "updateSpeed",
                update_speed.clone().map(|v| v.as_str().to_string()),
            ),
        ];

        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").cloned();

        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 = ApiClient::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 = ApiClient::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,"st":1}"#).unwrap_or_else(|_| serde_json::json!({}));
            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 = ApiClient::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,"st":1}"#).unwrap_or_else(|_| serde_json::json!({}));
            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 = ApiClient::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 = ApiClient::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","b":"9548.1","B":"52","a":"9548.5","A":"11","T":1591268628155,"E":1591268628166,"ps":"BTCUSD","st":1}"#).unwrap_or_else(|_| serde_json::json!({}));
            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 = ApiClient::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","b":"9548.1","B":"52","a":"9548.5","A":"11","T":1591268628155,"E":1591268628166,"ps":"BTCUSD","st":1}"#).unwrap_or_else(|_| serde_json::json!({}));
            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 = ApiClient::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 = ApiClient::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},"st":1}"#).unwrap_or_else(|_| serde_json::json!({}));
            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 = ApiClient::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},"st":1}"#).unwrap_or_else(|_| serde_json::json!({}));
            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 = ApiClient::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 = ApiClient::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","st":1}]"#).unwrap_or_else(|_| serde_json::json!({}));
            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 = ApiClient::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","st":1}]"#).unwrap_or_else(|_| serde_json::json!({}));
            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 = ApiClient::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 = ApiClient::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","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,"ps":"BTCUSD","st":1}]"#).unwrap_or_else(|_| serde_json::json!({}));
            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 = ApiClient::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","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,"ps":"BTCUSD","st":1}]"#).unwrap_or_else(|_| serde_json::json!({}));
            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 = ApiClient::new(streams_base.clone());

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

            let params = ContinuousContractKlineCandlestickStreamsParams::builder(
                "btcusdt".to_string(),
                ContinuousContractKlineCandlestickStreamsContractTypeEnum::Perpetual,
                ContinuousContractKlineCandlestickStreamsIntervalEnum::Interval1m,
            )
            .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.as_str().to_string())),
                ("interval", Some(interval.as_str().to_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(
                "/<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 = ApiClient::new(streams_base.clone());

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

            let params = ContinuousContractKlineCandlestickStreamsParams::builder("btcusdt".to_string(),ContinuousContractKlineCandlestickStreamsContractTypeEnum::Perpetual,ContinuousContractKlineCandlestickStreamsIntervalEnum::Interval1m,).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.as_str().to_string())
                ),
                ("interval",
                        Some(interval.as_str().to_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("/<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_or_else(|_| serde_json::json!({}));
            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 = ApiClient::new(streams_base.clone());

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

            let params = ContinuousContractKlineCandlestickStreamsParams::builder("btcusdt".to_string(),ContinuousContractKlineCandlestickStreamsContractTypeEnum::Perpetual,ContinuousContractKlineCandlestickStreamsIntervalEnum::Interval1m,).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.as_str().to_string())
                ),
                ("interval",
                        Some(interval.as_str().to_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("/<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_or_else(|_| serde_json::json!({}));
            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 = ApiClient::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 = ApiClient::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":500000,"mmr":0.0065,"cf":75,"mi":51,"ma":75}],"st":1}"#).unwrap_or_else(|_| serde_json::json!({}));
            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 = ApiClient::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":500000,"mmr":0.0065,"cf":75,"mi":51,"ma":75}],"st":1}"#).unwrap_or_else(|_| serde_json::json!({}));
            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 = ApiClient::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().map(|v| v.as_str().to_string()),
                ),
            ];

            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 = ApiClient::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().map(|v| v.as_str().to_string())
                ),
            ];

            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"]],"a":[["9518.5"]],"st":1}"#).unwrap_or_else(|_| serde_json::json!({}));
            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 = ApiClient::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().map(|v| v.as_str().to_string())
                ),
            ];

            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"]],"a":[["9518.5"]],"st":1}"#).unwrap_or_else(|_| serde_json::json!({}));
            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 = ApiClient::new(streams_base.clone());

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

            let params = IndexKlineCandlestickStreamsParams::builder(
                "btcusdt".to_string(),
                IndexKlineCandlestickStreamsIntervalEnum::Interval1m,
            )
            .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.as_str().to_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("/<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 = ApiClient::new(streams_base.clone());

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

            let params = IndexKlineCandlestickStreamsParams::builder("btcusdt".to_string(),IndexKlineCandlestickStreamsIntervalEnum::Interval1m,).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.as_str().to_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("/<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_or_else(|_| serde_json::json!({}));
            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 = ApiClient::new(streams_base.clone());

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

            let params = IndexKlineCandlestickStreamsParams::builder("btcusdt".to_string(),IndexKlineCandlestickStreamsIntervalEnum::Interval1m,).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.as_str().to_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("/<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_or_else(|_| serde_json::json!({}));
            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 = ApiClient::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().map(|v| v.as_str().to_string()),
                ),
            ];

            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 = ApiClient::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().map(|v| v.as_str().to_string()),
                ),
            ];

            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_or_else(|_| serde_json::json!({}));
            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 = ApiClient::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().map(|v| v.as_str().to_string()),
                ),
            ];

            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_or_else(|_| serde_json::json!({}));
            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 = ApiClient::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 = ApiClient::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","b":"9548.1","B":"52","a":"9548.5","A":"11","T":1591268628155,"E":1591268628166,"ps":"BTCUSD","st":2}"#).unwrap_or_else(|_| serde_json::json!({}));
            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 = ApiClient::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","b":"9548.1","B":"52","a":"9548.5","A":"11","T":1591268628155,"E":1591268628166,"ps":"BTCUSD","st":2}"#).unwrap_or_else(|_| serde_json::json!({}));
            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 = ApiClient::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 = ApiClient::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","st":1}"#).unwrap_or_else(|_| serde_json::json!({}));
            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 = ApiClient::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","st":1}"#).unwrap_or_else(|_| serde_json::json!({}));
            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 = ApiClient::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 = ApiClient::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","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,"ps":"BTCUSD","st":1}"#).unwrap_or_else(|_| serde_json::json!({}));
            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 = ApiClient::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","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,"ps":"BTCUSD","st":1}"#).unwrap_or_else(|_| serde_json::json!({}));
            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 = ApiClient::new(streams_base.clone());

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

            let params = KlineCandlestickStreamsParams::builder(
                "btcusdt".to_string(),
                KlineCandlestickStreamsIntervalEnum::Interval1m,
            )
            .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.as_str().to_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("/<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 = ApiClient::new(streams_base.clone());

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

            let params = KlineCandlestickStreamsParams::builder("btcusdt".to_string(),KlineCandlestickStreamsIntervalEnum::Interval1m,).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.as_str().to_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("/<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_or_else(|_| serde_json::json!({}));
            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 = ApiClient::new(streams_base.clone());

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

            let params = KlineCandlestickStreamsParams::builder("btcusdt".to_string(),KlineCandlestickStreamsIntervalEnum::Interval1m,).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.as_str().to_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("/<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_or_else(|_| serde_json::json!({}));
            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 = ApiClient::new(streams_base.clone());

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

            let params = MarkPriceKlineCandlestickStreamsParams::builder(
                "btcusdt".to_string(),
                MarkPriceKlineCandlestickStreamsIntervalEnum::Interval1m,
            )
            .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.as_str().to_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(
                "/<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 = ApiClient::new(streams_base.clone());

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

            let params = MarkPriceKlineCandlestickStreamsParams::builder("btcusdt".to_string(),MarkPriceKlineCandlestickStreamsIntervalEnum::Interval1m,).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.as_str().to_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("/<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_or_else(|_| serde_json::json!({}));
            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 = ApiClient::new(streams_base.clone());

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

            let params = MarkPriceKlineCandlestickStreamsParams::builder("btcusdt".to_string(),MarkPriceKlineCandlestickStreamsIntervalEnum::Interval1m,).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.as_str().to_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("/<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_or_else(|_| serde_json::json!({}));
            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 = ApiClient::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().map(|v| v.as_str().to_string()),
                ),
            ];

            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 = ApiClient::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().map(|v| v.as_str().to_string())
                ),
            ];

            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,"st":1}]"#).unwrap_or_else(|_| serde_json::json!({}));
            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 = ApiClient::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().map(|v| v.as_str().to_string())
                ),
            ];

            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,"st":1}]"#).unwrap_or_else(|_| serde_json::json!({}));
            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 = ApiClient::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().map(|v| v.as_str().to_string()),
                ),
            ];

            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 = ApiClient::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().map(|v| v.as_str().to_string())
                ),
            ];

            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,"st":1}"#).unwrap_or_else(|_| serde_json::json!({}));
            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 = ApiClient::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().map(|v| v.as_str().to_string())
                ),
            ];

            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,"st":1}"#).unwrap_or_else(|_| serde_json::json!({}));
            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 market_liquidation_order_streams_should_execute_successfully() {
        TOKIO_SHARED_RT.block_on(async {
            let (streams_base, _) = make_streams_base().await;
            let api = ApiClient::new(streams_base.clone());

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

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

            let MarketLiquidationOrderStreamsParams { 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
                .market_liquidation_order_streams(params)
                .await
                .expect("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 market_liquidation_order_streams_should_handle_incoming_message() {
        TOKIO_SHARED_RT.block_on(async {
            let (streams_base, conn) = make_streams_base().await;
            let api = ApiClient::new(streams_base.clone());

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

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

            let MarketLiquidationOrderStreamsParams {
                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.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::MarketLiquidationOrderStreamsResponse| {
                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_or_else(|_| serde_json::json!({}));
            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 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 = ApiClient::new(streams_base.clone());

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

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

            let MarketLiquidationOrderStreamsParams {
                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.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::MarketLiquidationOrderStreamsResponse| {
                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_or_else(|_| serde_json::json!({}));
            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 = ApiClient::new(streams_base.clone());

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

            let params = PartialBookDepthStreamsParams::builder(
                "btcusdt".to_string(),
                PartialBookDepthStreamsLevelsEnum::Levels5,
            )
            .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.as_str().to_string())),
                ("id", id.clone()),
                (
                    "updateSpeed",
                    update_speed.clone().map(|v| v.as_str().to_string()),
                ),
            ];

            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 = ApiClient::new(streams_base.clone());

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

            let params = PartialBookDepthStreamsParams::builder("btcusdt".to_string(),PartialBookDepthStreamsLevelsEnum::Levels5,).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.as_str().to_string())
                ),
                ("id",
                        id.clone()
                ),
                ("updateSpeed",
                        update_speed.clone().map(|v| v.as_str().to_string())
                ),
            ];

            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"]],"a":[["9524.6"]],"st":1}"#).unwrap_or_else(|_| serde_json::json!({}));
            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 = ApiClient::new(streams_base.clone());

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

            let params = PartialBookDepthStreamsParams::builder("btcusdt".to_string(),PartialBookDepthStreamsLevelsEnum::Levels5,).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.as_str().to_string())
                ),
                ("id",
                        id.clone()
                ),
                ("updateSpeed",
                        update_speed.clone().map(|v| v.as_str().to_string())
                ),
            ];

            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"]],"a":[["9524.6"]],"st":1}"#).unwrap_or_else(|_| serde_json::json!({}));
            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");
        });
    }
}