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

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

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

#[async_trait]
pub trait WebsocketMarketStreamsApi: Send + Sync {
    async fn index_price_streams(
        &self,
        params: IndexPriceStreamsParams,
    ) -> anyhow::Result<Arc<WebsocketStream<models::IndexPriceStreamsResponse>>>;
    async fn kline_candlestick_streams(
        &self,
        params: KlineCandlestickStreamsParams,
    ) -> anyhow::Result<Arc<WebsocketStream<models::KlineCandlestickStreamsResponse>>>;
    async fn mark_price(
        &self,
        params: MarkPriceParams,
    ) -> anyhow::Result<Arc<WebsocketStream<Vec<models::MarkPriceResponseInner>>>>;
    async fn new_symbol_info(
        &self,
        params: NewSymbolInfoParams,
    ) -> anyhow::Result<Arc<WebsocketStream<models::NewSymbolInfoResponse>>>;
    async fn open_interest(
        &self,
        params: OpenInterestParams,
    ) -> anyhow::Result<Arc<WebsocketStream<Vec<models::OpenInterestResponseInner>>>>;
    async fn partial_book_depth_streams(
        &self,
        params: PartialBookDepthStreamsParams,
    ) -> anyhow::Result<Arc<WebsocketStream<models::PartialBookDepthStreamsResponse>>>;
    async fn ticker24_hour(
        &self,
        params: Ticker24HourParams,
    ) -> anyhow::Result<Arc<WebsocketStream<models::Ticker24HourResponse>>>;
    async fn ticker24_hour_by_underlying_asset_and_expiration_data(
        &self,
        params: Ticker24HourByUnderlyingAssetAndExpirationDataParams,
    ) -> anyhow::Result<
        Arc<
            WebsocketStream<
                Vec<models::Ticker24HourByUnderlyingAssetAndExpirationDataResponseInner>,
            >,
        >,
    >;
    async fn trade_streams(
        &self,
        params: TradeStreamsParams,
    ) -> anyhow::Result<Arc<WebsocketStream<models::TradeStreamsResponse>>>;
}

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

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

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

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

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

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

impl NewSymbolInfoParams {
    /// Create a builder for [`new_symbol_info`].
    ///
    #[must_use]
    pub fn builder() -> NewSymbolInfoParamsBuilder {
        NewSymbolInfoParamsBuilder::default()
    }
}
/// Request parameters for the [`open_interest`] operation.
///
/// This struct holds all of the inputs you can pass when calling
/// [`open_interest`](#method.open_interest).
#[derive(Clone, Debug, Builder)]
#[builder(pattern = "owned", build_fn(error = "ParamBuildError"))]
pub struct OpenInterestParams {
    /// The underlyingAsset parameter
    ///
    /// This field is **required.
    #[builder(setter(into))]
    pub underlying_asset: String,
    /// The expirationDate parameter
    ///
    /// This field is **required.
    #[builder(setter(into))]
    pub expiration_date: String,
    /// Unique WebSocket request ID.
    ///
    /// This field is **optional.
    #[builder(setter(into), default)]
    pub id: Option<String>,
}

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

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

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

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

impl TradeStreamsParams {
    /// Create a builder for [`trade_streams`].
    ///
    /// Required parameters:
    ///
    /// * `symbol` — The symbol parameter
    ///
    #[must_use]
    pub fn builder(symbol: String) -> TradeStreamsParamsBuilder {
        TradeStreamsParamsBuilder::default().symbol(symbol)
    }
}

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

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

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

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

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

        Ok(create_stream_handler::<models::IndexPriceStreamsResponse>(
            WebsocketBase::WebsocketStreams(Arc::clone(&self.websocket_streams_base)),
            stream,
            id_opt,
        )
        .await)
    }

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

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

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

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

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

        Ok(
            create_stream_handler::<models::KlineCandlestickStreamsResponse>(
                WebsocketBase::WebsocketStreams(Arc::clone(&self.websocket_streams_base)),
                stream,
                id_opt,
            )
            .await,
        )
    }

    async fn mark_price(
        &self,
        params: MarkPriceParams,
    ) -> anyhow::Result<Arc<WebsocketStream<Vec<models::MarkPriceResponseInner>>>> {
        let MarkPriceParams {
            underlying_asset,
            id,
        } = params;

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

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

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

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

        Ok(
            create_stream_handler::<Vec<models::MarkPriceResponseInner>>(
                WebsocketBase::WebsocketStreams(Arc::clone(&self.websocket_streams_base)),
                stream,
                id_opt,
            )
            .await,
        )
    }

    async fn new_symbol_info(
        &self,
        params: NewSymbolInfoParams,
    ) -> anyhow::Result<Arc<WebsocketStream<models::NewSymbolInfoResponse>>> {
        let NewSymbolInfoParams { id } = params;

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

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

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

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

        Ok(create_stream_handler::<models::NewSymbolInfoResponse>(
            WebsocketBase::WebsocketStreams(Arc::clone(&self.websocket_streams_base)),
            stream,
            id_opt,
        )
        .await)
    }

    async fn open_interest(
        &self,
        params: OpenInterestParams,
    ) -> anyhow::Result<Arc<WebsocketStream<Vec<models::OpenInterestResponseInner>>>> {
        let OpenInterestParams {
            underlying_asset,
            expiration_date,
            id,
        } = params;

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

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

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

        let stream = replace_websocket_streams_placeholders(
            "/<underlyingAsset>@openInterest@<expirationDate>",
            &vars,
        );

        Ok(
            create_stream_handler::<Vec<models::OpenInterestResponseInner>>(
                WebsocketBase::WebsocketStreams(Arc::clone(&self.websocket_streams_base)),
                stream,
                id_opt,
            )
            .await,
        )
    }

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

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

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

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

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

        Ok(
            create_stream_handler::<models::PartialBookDepthStreamsResponse>(
                WebsocketBase::WebsocketStreams(Arc::clone(&self.websocket_streams_base)),
                stream,
                id_opt,
            )
            .await,
        )
    }

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

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

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

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

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

        Ok(create_stream_handler::<models::Ticker24HourResponse>(
            WebsocketBase::WebsocketStreams(Arc::clone(&self.websocket_streams_base)),
            stream,
            id_opt,
        )
        .await)
    }

    async fn ticker24_hour_by_underlying_asset_and_expiration_data(
        &self,
        params: Ticker24HourByUnderlyingAssetAndExpirationDataParams,
    ) -> anyhow::Result<
        Arc<
            WebsocketStream<
                Vec<models::Ticker24HourByUnderlyingAssetAndExpirationDataResponseInner>,
            >,
        >,
    > {
        let Ticker24HourByUnderlyingAssetAndExpirationDataParams {
            underlying_asset,
            expiration_date,
            id,
        } = params;

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

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

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

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

        Ok(create_stream_handler::<
            Vec<models::Ticker24HourByUnderlyingAssetAndExpirationDataResponseInner>,
        >(
            WebsocketBase::WebsocketStreams(Arc::clone(&self.websocket_streams_base)),
            stream,
            id_opt,
        )
        .await)
    }

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

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

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

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

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

        Ok(create_stream_handler::<models::TradeStreamsResponse>(
            WebsocketBase::WebsocketStreams(Arc::clone(&self.websocket_streams_base)),
            stream,
            id_opt,
        )
        .await)
    }
}

#[cfg(all(test, feature = "derivatives_trading_options"))]
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()]);
        conn.set_handler(streams_base.clone() as Arc<dyn WebsocketHandler>)
            .await;
        (streams_base, conn)
    }

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

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

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

            let IndexPriceStreamsParams { 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>@index", &vars);
            let ws_stream = api
                .index_price_streams(params)
                .await
                .expect("index_price_streams should return a WebsocketStream");

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

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

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

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

            let IndexPriceStreamsParams { 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>@index", &vars);

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

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

            let payload: Value = serde_json::from_str(
                r#"{"e":"index","E":1661415480351,"s":"ETHUSDT","p":"1707.89008607"}"#,
            )
            .unwrap();
            let msg = json!({
                "stream": stream,
                "data": payload,
            });

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

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

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

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

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

            let IndexPriceStreamsParams { 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>@index", &vars);

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

            let called = Arc::new(AtomicBool::new(false));
            let called_clone = called.clone();
            ws_stream.on_message(move |_payload: models::IndexPriceStreamsResponse| {
                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":"index","E":1661415480351,"s":"ETHUSDT","p":"1707.89008607"}"#,
            )
            .unwrap();
            let msg = json!({
                "stream": stream,
                "data": payload,
            });

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

            yield_now().await;

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

            let payload: Value = serde_json::from_str(r#"{"e":"kline","E":1638747660000,"s":"BTC-200630-9000-P","k":{"t":1638747660000,"T":1638747719999,"s":"BTC-200630-9000-P","i":"1m","F":0,"L":0,"o":"1000","c":"1000","h":"1000","l":"1000","v":"0","n":0,"x":false,"q":"0","V":"0","Q":"0"}}"#).unwrap();
            let msg = json!({
                "stream": stream,
                "data": payload,
            });

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

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

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

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

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

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

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

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

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

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

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

            ws_stream.unsubscribe().await;

            let payload: Value = serde_json::from_str(r#"{"e":"kline","E":1638747660000,"s":"BTC-200630-9000-P","k":{"t":1638747660000,"T":1638747719999,"s":"BTC-200630-9000-P","i":"1m","F":0,"L":0,"o":"1000","c":"1000","h":"1000","l":"1000","v":"0","n":0,"x":false,"q":"0","V":"0","Q":"0"}}"#).unwrap();
            let msg = json!({
                "stream": stream,
                "data": payload,
            });

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

            yield_now().await;

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

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

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

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

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

            let pairs: &[(&str, Option<String>)] = &[
                ("underlyingAsset", Some(underlying_asset.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("/<underlyingAsset>@markPrice", &vars);
            let ws_stream = api
                .mark_price(params)
                .await
                .expect("mark_price should return a WebsocketStream");

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

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

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

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

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

            let pairs: &[(&str, Option<String>)] = &[
                ("underlyingAsset",
                        Some(underlying_asset.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("/<underlyingAsset>@markPrice", &vars);

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

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

            let payload: Value = serde_json::from_str(r#"[{"e":"markPrice","E":1663684594227,"s":"ETH-220930-1500-C","mp":"30.3"},{"e":"markPrice","E":1663684594228,"s":"ETH-220923-1000-C","mp":"341.5"}]"#).unwrap();
            let msg = json!({
                "stream": stream,
                "data": payload,
            });

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

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

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

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

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

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

            let pairs: &[(&str, Option<String>)] = &[
                ("underlyingAsset",
                        Some(underlying_asset.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("/<underlyingAsset>@markPrice", &vars);

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

            let called = Arc::new(AtomicBool::new(false));
            let called_clone = called.clone();
            ws_stream.on_message(move |_payload: Vec<models::MarkPriceResponseInner>| {
                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","E":1663684594227,"s":"ETH-220930-1500-C","mp":"30.3"},{"e":"markPrice","E":1663684594228,"s":"ETH-220923-1000-C","mp":"341.5"}]"#).unwrap();
            let msg = json!({
                "stream": stream,
                "data": payload,
            });

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

            yield_now().await;

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

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

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

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

            let NewSymbolInfoParams { 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("/option_pair", &vars);
            let ws_stream = api
                .new_symbol_info(params)
                .await
                .expect("new_symbol_info should return a WebsocketStream");

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

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

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

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

            let NewSymbolInfoParams {
                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("/option_pair", &vars);

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

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

            let payload: Value = serde_json::from_str(r#"{"e":"OPTION_PAIR","E":1668573571842,"u":"BTCUSDT","qa":"USDT","s":"BTC-221116-21000-C","unit":1,"mq":"0.01","d":"CALL","sp":"21000","ed":1668585600000}"#).unwrap();
            let msg = json!({
                "stream": stream,
                "data": payload,
            });

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

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

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

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

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

            let NewSymbolInfoParams {
                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("/option_pair", &vars);

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

            let called = Arc::new(AtomicBool::new(false));
            let called_clone = called.clone();
            ws_stream.on_message(move |_payload: models::NewSymbolInfoResponse| {
                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":"OPTION_PAIR","E":1668573571842,"u":"BTCUSDT","qa":"USDT","s":"BTC-221116-21000-C","unit":1,"mq":"0.01","d":"CALL","sp":"21000","ed":1668585600000}"#).unwrap();
            let msg = json!({
                "stream": stream,
                "data": payload,
            });

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

            yield_now().await;

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

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

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

            let params = OpenInterestParams::builder("ETH".to_string(), "220930".to_string())
                .id(Some(id.clone()))
                .build()
                .unwrap();

            let OpenInterestParams {
                underlying_asset,
                expiration_date,
                id,
            } = params.clone();

            let pairs: &[(&str, Option<String>)] = &[
                ("underlyingAsset", Some(underlying_asset.clone())),
                ("expirationDate", Some(expiration_date.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(
                "/<underlyingAsset>@openInterest@<expirationDate>",
                &vars,
            );
            let ws_stream = api
                .open_interest(params)
                .await
                .expect("open_interest should return a WebsocketStream");

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

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

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

            let params = OpenInterestParams::builder("ETH".to_string(),"220930".to_string(),).id(Some(id.clone())).build().unwrap();

            let OpenInterestParams {
                underlying_asset,expiration_date,id,
            } = params.clone();

            let pairs: &[(&str, Option<String>)] = &[
                ("underlyingAsset",
                        Some(underlying_asset.clone())
                ),
                ("expirationDate",
                        Some(expiration_date.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("/<underlyingAsset>@openInterest@<expirationDate>", &vars);

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

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

            let payload: Value = serde_json::from_str(r#"[{"e":"openInterest","E":1668759300045,"s":"ETH-221125-2700-C","o":"1580.87","h":"1912992.178168204"}]"#).unwrap();
            let msg = json!({
                "stream": stream,
                "data": payload,
            });

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

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

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

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

            let params = OpenInterestParams::builder("ETH".to_string(),"220930".to_string(),).id(Some(id.clone())).build().unwrap();

            let OpenInterestParams {
                underlying_asset,expiration_date,id,
            } = params.clone();

            let pairs: &[(&str, Option<String>)] = &[
                ("underlyingAsset",
                        Some(underlying_asset.clone())
                ),
                ("expirationDate",
                        Some(expiration_date.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("/<underlyingAsset>@openInterest@<expirationDate>", &vars);

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

            let called = Arc::new(AtomicBool::new(false));
            let called_clone = called.clone();
            ws_stream.on_message(move |_payload: Vec<models::OpenInterestResponseInner>| {
                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":"openInterest","E":1668759300045,"s":"ETH-221125-2700-C","o":"1580.87","h":"1912992.178168204"}]"#).unwrap();
            let msg = json!({
                "stream": stream,
                "data": payload,
            });

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

            yield_now().await;

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

            let payload: Value = serde_json::from_str(r#"{"e":"depth","E":1591695934010,"T":1591695934000,"s":"BTC-200630-9000-P","u":162,"pu":162,"b":[["200","3"],["101","1"],["100","2"]],"a":[["1000","89"]]}"#).unwrap();
            let msg = json!({
                "stream": stream,
                "data": payload,
            });

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

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

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

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

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

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

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

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

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

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

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

            ws_stream.unsubscribe().await;

            let payload: Value = serde_json::from_str(r#"{"e":"depth","E":1591695934010,"T":1591695934000,"s":"BTC-200630-9000-P","u":162,"pu":162,"b":[["200","3"],["101","1"],["100","2"]],"a":[["1000","89"]]}"#).unwrap();
            let msg = json!({
                "stream": stream,
                "data": payload,
            });

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

            yield_now().await;

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

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

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

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

            let Ticker24HourParams { 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
                .ticker24_hour(params)
                .await
                .expect("ticker24_hour should return a WebsocketStream");

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

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

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

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

            let Ticker24HourParams {
                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.ticker24_hour(params).await.unwrap();

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

            let payload: Value = serde_json::from_str(r#"{"e":"24hrTicker","E":1657706425200,"T":1657706425220,"s":"BTC-220930-18000-C","o":"2000","h":"2020","l":"2000","c":"2020","V":"1.42","A":"2841","P":"0.01","p":"20","Q":"0.01","F":"27","L":"48","n":22,"bo":"2012","ao":"2020","bq":"4.9","aq":"0.03","b":"0.1202","a":"0.1318","d":"0.98911","t":"-0.16961","g":"0.00004","v":"2.66584","vo":"0.10001","mp":"2003.5102","hl":"2023.511","ll":"1983.511","eep":"0"}"#).unwrap();
            let msg = json!({
                "stream": stream,
                "data": payload,
            });

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

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

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

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

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

            let Ticker24HourParams {
                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.ticker24_hour(params).await.unwrap();

            let called = Arc::new(AtomicBool::new(false));
            let called_clone = called.clone();
            ws_stream.on_message(move |_payload: models::Ticker24HourResponse| {
                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":1657706425200,"T":1657706425220,"s":"BTC-220930-18000-C","o":"2000","h":"2020","l":"2000","c":"2020","V":"1.42","A":"2841","P":"0.01","p":"20","Q":"0.01","F":"27","L":"48","n":22,"bo":"2012","ao":"2020","bq":"4.9","aq":"0.03","b":"0.1202","a":"0.1318","d":"0.98911","t":"-0.16961","g":"0.00004","v":"2.66584","vo":"0.10001","mp":"2003.5102","hl":"2023.511","ll":"1983.511","eep":"0"}"#).unwrap();
            let msg = json!({
                "stream": stream,
                "data": payload,
            });

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

            yield_now().await;

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

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

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

            let params = Ticker24HourByUnderlyingAssetAndExpirationDataParams::builder("ETH".to_string(),"220930".to_string(),).id(Some(id.clone())).build().unwrap();

            let Ticker24HourByUnderlyingAssetAndExpirationDataParams {
                underlying_asset,expiration_date,id,
            } = params.clone();

            let pairs: &[(&str, Option<String>)] = &[
                ("underlyingAsset",
                        Some(underlying_asset.clone())
                ),
                ("expirationDate",
                        Some(expiration_date.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("/<underlyingAsset>@ticker@<expirationDate>", &vars);
            let ws_stream = api.ticker24_hour_by_underlying_asset_and_expiration_data(params).await.expect("ticker24_hour_by_underlying_asset_and_expiration_data should return a WebsocketStream");

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

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

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

            let params = Ticker24HourByUnderlyingAssetAndExpirationDataParams::builder("ETH".to_string(),"220930".to_string(),).id(Some(id.clone())).build().unwrap();

            let Ticker24HourByUnderlyingAssetAndExpirationDataParams {
                underlying_asset,expiration_date,id,
            } = params.clone();

            let pairs: &[(&str, Option<String>)] = &[
                ("underlyingAsset",
                        Some(underlying_asset.clone())
                ),
                ("expirationDate",
                        Some(expiration_date.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("/<underlyingAsset>@ticker@<expirationDate>", &vars);

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

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

            let payload: Value = serde_json::from_str(r#"[{"e":"24hrTicker","E":1657706425200,"T":1657706425220,"s":"ETH-220930-1600-C","o":"2000","h":"2020","l":"2000","c":"2020","V":"1.42","A":"2841","P":"0.01","p":"20","Q":"0.01","F":"27","L":"48","n":22,"bo":"2012","ao":"2020","bq":"4.9","aq":"0.03","b":"0.1202","a":"0.1318","d":"0.98911","t":"-0.16961","g":"0.00004","v":"2.66584","vo":"0.10001","mp":"2003.5102","hl":"2023.511","ll":"1983.511","eep":"0"},{"e":"24hrTicker","E":1663685112123,"s":"ETH-220930-1500-C","o":"34.9","h":"44.6","l":"26.8","c":"26.8","V":"11.84","A":"444.37","P":"-0.232","p":"-8.1","Q":"0","F":"91","L":"129","n":39,"bo":"26.8","ao":"33.9","bq":"0.65","aq":"0.01","b":"0.88790536","a":"0.98729014","d":"0.2621153","t":"-3.44806807","g":"0.00158298","v":"0.7148147","vo":"0.93759775","mp":"30.3","hl":"228.7","ll":"0.1","eep":"0"}]"#).unwrap();
            let msg = json!({
                "stream": stream,
                "data": payload,
            });

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

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

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

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

            let params = Ticker24HourByUnderlyingAssetAndExpirationDataParams::builder("ETH".to_string(),"220930".to_string(),).id(Some(id.clone())).build().unwrap();

            let Ticker24HourByUnderlyingAssetAndExpirationDataParams {
                underlying_asset,expiration_date,id,
            } = params.clone();

            let pairs: &[(&str, Option<String>)] = &[
                ("underlyingAsset",
                        Some(underlying_asset.clone())
                ),
                ("expirationDate",
                        Some(expiration_date.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("/<underlyingAsset>@ticker@<expirationDate>", &vars);

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

            let called = Arc::new(AtomicBool::new(false));
            let called_clone = called.clone();
            ws_stream.on_message(move |_payload: Vec<models::Ticker24HourByUnderlyingAssetAndExpirationDataResponseInner>| {
                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":1657706425200,"T":1657706425220,"s":"ETH-220930-1600-C","o":"2000","h":"2020","l":"2000","c":"2020","V":"1.42","A":"2841","P":"0.01","p":"20","Q":"0.01","F":"27","L":"48","n":22,"bo":"2012","ao":"2020","bq":"4.9","aq":"0.03","b":"0.1202","a":"0.1318","d":"0.98911","t":"-0.16961","g":"0.00004","v":"2.66584","vo":"0.10001","mp":"2003.5102","hl":"2023.511","ll":"1983.511","eep":"0"},{"e":"24hrTicker","E":1663685112123,"s":"ETH-220930-1500-C","o":"34.9","h":"44.6","l":"26.8","c":"26.8","V":"11.84","A":"444.37","P":"-0.232","p":"-8.1","Q":"0","F":"91","L":"129","n":39,"bo":"26.8","ao":"33.9","bq":"0.65","aq":"0.01","b":"0.88790536","a":"0.98729014","d":"0.2621153","t":"-3.44806807","g":"0.00158298","v":"0.7148147","vo":"0.93759775","mp":"30.3","hl":"228.7","ll":"0.1","eep":"0"}]"#).unwrap();
            let msg = json!({
                "stream": stream,
                "data": payload,
            });

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

            yield_now().await;

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

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

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

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

            let TradeStreamsParams { 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>@trade", &vars);
            let ws_stream = api
                .trade_streams(params)
                .await
                .expect("trade_streams should return a WebsocketStream");

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

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

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

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

            let TradeStreamsParams {
                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>@trade", &vars);

            let ws_stream = api.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::TradeStreamsResponse| {
                called_with_message.store(true, Ordering::SeqCst);
            });

            let payload: Value = serde_json::from_str(r#"{"e":"trade","E":1591677941092,"s":"BTC-200630-9000-P","t":"1","p":"1000","q":"-2","b":4611781675939004000,"a":4611781675939004000,"T":1591677567872,"S":"-1","X":"MARKET"}"#).unwrap();
            let msg = json!({
                "stream": stream,
                "data": payload,
            });

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

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

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

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

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

            let TradeStreamsParams {
                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>@trade", &vars);

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

            let called = Arc::new(AtomicBool::new(false));
            let called_clone = called.clone();
            ws_stream.on_message(move |_payload: models::TradeStreamsResponse| {
                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":"trade","E":1591677941092,"s":"BTC-200630-9000-P","t":"1","p":"1000","q":"-2","b":4611781675939004000,"a":4611781675939004000,"T":1591677567872,"S":"-1","X":"MARKET"}"#).unwrap();
            let msg = json!({
                "stream": stream,
                "data": payload,
            });

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

            yield_now().await;

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