/*
 * 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;
use crate::models::StreamId;

#[async_trait]
pub trait MarketApi: Send + Sync {
    async fn index_price_streams(
        &self,
        params: IndexPriceStreamsParams,
    ) -> anyhow::Result<Arc<WebsocketStream<Vec<models::IndexPriceStreamsResponseInner>>>>;
    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>>>>;
}

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

impl MarketApiClient {
    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, Default)]
#[builder(pattern = "owned", build_fn(error = "ParamBuildError"))]
pub struct IndexPriceStreamsParams {
    /// Unique WebSocket request ID.
    ///
    /// This field is **optional.
    #[builder(setter(into), default)]
    pub id: Option<u32>,
}

impl IndexPriceStreamsParams {
    /// Create a builder for [`index_price_streams`].
    ///
    #[must_use]
    pub fn builder() -> IndexPriceStreamsParamsBuilder {
        IndexPriceStreamsParamsBuilder::default()
    }
}
/// 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<u32>,
}

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 underlying parameter
    ///
    /// This field is **required.
    #[builder(setter(into))]
    pub underlying: String,
    /// Unique WebSocket request ID.
    ///
    /// This field is **optional.
    #[builder(setter(into), default)]
    pub id: Option<u32>,
}

impl MarkPriceParams {
    /// Create a builder for [`mark_price`].
    ///
    /// Required parameters:
    ///
    /// * `underlying` — The underlying parameter
    ///
    #[must_use]
    pub fn builder(underlying: String) -> MarkPriceParamsBuilder {
        MarkPriceParamsBuilder::default().underlying(underlying)
    }
}
/// 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<u32>,
}

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 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<u32>,
}

impl OpenInterestParams {
    /// Create a builder for [`open_interest`].
    ///
    /// Required parameters:
    ///
    /// * `expiration_date` — The expirationDate parameter
    ///
    #[must_use]
    pub fn builder(expiration_date: String) -> OpenInterestParamsBuilder {
        OpenInterestParamsBuilder::default().expiration_date(expiration_date)
    }
}

#[async_trait]
impl MarketApi for MarketApiClient {
    async fn index_price_streams(
        &self,
        params: IndexPriceStreamsParams,
    ) -> anyhow::Result<Arc<WebsocketStream<Vec<models::IndexPriceStreamsResponseInner>>>> {
        let IndexPriceStreamsParams { id } = params;

        let pairs: &[(&str, Option<String>)] = &[("id", id.map(|v| v.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").map(std::string::ToString::to_string);

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

        Ok(
            create_stream_handler::<Vec<models::IndexPriceStreamsResponseInner>>(
                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)
                }),
                Some("market".to_string()),
            )
            .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.map(|v| v.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").map(std::string::ToString::to_string);

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

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

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

        let pairs: &[(&str, Option<String>)] = &[
            ("underlying", Some(underlying.clone())),
            ("id", id.map(|v| v.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").map(std::string::ToString::to_string);

        let stream = replace_websocket_streams_placeholders("/<underlying>@optionMarkPrice", &vars);

        Ok(
            create_stream_handler::<Vec<models::MarkPriceResponseInner>>(
                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)
                }),
                Some("market".to_string()),
            )
            .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.map(|v| v.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").map(std::string::ToString::to_string);

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

        Ok(create_stream_handler::<models::NewSymbolInfoResponse>(
            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)
            }),
            Some("market".to_string()),
        )
        .await)
    }

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

        let pairs: &[(&str, Option<String>)] = &[
            ("expirationDate", Some(expiration_date.clone())),
            ("id", id.map(|v| v.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").map(std::string::ToString::to_string);

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

        Ok(
            create_stream_handler::<Vec<models::OpenInterestResponseInner>>(
                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)
                }),
                Some("market".to_string()),
            )
            .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()], vec!["market".to_string()]);
        {
            let mut st = conn.state.lock().await;
            st.url_path = Some("market".to_string());
        }
        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 = MarketApiClient::new(streams_base.clone());

            let id = 123456u32;

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

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

            let pairs: &[(&str, Option<String>)] = &[("id", id.map(|v| v.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("/!index@arr", &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, Some(StreamId::Number(123456u32)));
        });
    }

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

            let id = 123456u32;

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

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

            let pairs: &[(&str, Option<String>)] = &[
                ("id",
                        id.map(|v| v.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("/!index@arr", &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: Vec<models::IndexPriceStreamsResponseInner>| {
                called_with_message.store(true, Ordering::SeqCst);
            });

            let payload: Value = serde_json::from_str(r#"[{"e":"indexPrice","E":1763092572229,"s":"ETHUSDT","p":"3224.51976744"},{"e":"indexPrice","E":1763092572229,"s":"BTCUSDT","p":"99102.32326087"}]"#).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 = MarketApiClient::new(streams_base.clone());

            let id = 123456u32;

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

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

            let pairs: &[(&str, Option<String>)] = &[
                ("id",
                        id.map(|v| v.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("/!index@arr", &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: Vec<models::IndexPriceStreamsResponseInner>| {
                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","E":1763092572229,"s":"ETHUSDT","p":"3224.51976744"},{"e":"indexPrice","E":1763092572229,"s":"BTCUSDT","p":"99102.32326087"}]"#).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 = MarketApiClient::new(streams_base.clone());

            let id = 123456u32;

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

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

            let pairs: &[(&str, Option<String>)] = &[
                ("symbol", Some(symbol.clone())),
                ("interval", Some(interval.clone())),
                ("id", id.map(|v| v.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>@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::Number(123456u32)));
        });
    }

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

            let id = 123456u32;

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

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

            let pairs: &[(&str, Option<String>)] = &[
                ("symbol",
                        Some(symbol.clone())
                ),
                ("interval",
                        Some(interval.clone())
                ),
                ("id",
                        id.map(|v| v.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>@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 = MarketApiClient::new(streams_base.clone());

            let id = 123456u32;

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

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

            let pairs: &[(&str, Option<String>)] = &[
                ("symbol",
                        Some(symbol.clone())
                ),
                ("interval",
                        Some(interval.clone())
                ),
                ("id",
                        id.map(|v| v.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>@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 = MarketApiClient::new(streams_base.clone());

            let id = 123456u32;

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

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

            let pairs: &[(&str, Option<String>)] = &[
                ("underlying", Some(underlying.clone())),
                ("id", id.map(|v| v.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("/<underlying>@optionMarkPrice", &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, Some(StreamId::Number(123456u32)));
        });
    }

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

            let id = 123456u32;

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

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

            let pairs: &[(&str, Option<String>)] = &[
                ("underlying",
                        Some(underlying.clone())
                ),
                ("id",
                        id.map(|v| v.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("/<underlying>@optionMarkPrice", &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#"[{"s":"BTC-251120-126000-C","mp":"770.543","E":1762867543321,"e":"markPrice","i":"104334.60217391","P":"0.000","bo":"0.000","ao":"900.000","bq":"0.0000","aq":"0.2000","b":"-1.0","a":"0.98161161","hl":"924.652","ll":"616.435","vo":"0.9408058","rf":"0.0","d":"0.11111964","t":"-164.26702615","g":"0.00001245","v":"30.63855919"},{"s":"BTC-251123-126000-C","mp":"1249.61","E":1762867543321,"e":"markPrice","i":"104334.60217391","P":"0.000","bo":"1200.000","ao":"1300.000","bq":"0.3000","aq":"0.6000","b":"0.92159033","a":"0.94461441","hl":"1499.533","ll":"999.688","vo":"0.93310237","rf":"0.0","d":"0.14869196","t":"-172.12148811","g":"0.00001326","v":"43.43627792"}]"#).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 = MarketApiClient::new(streams_base.clone());

            let id = 123456u32;

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

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

            let pairs: &[(&str, Option<String>)] = &[
                ("underlying",
                        Some(underlying.clone())
                ),
                ("id",
                        id.map(|v| v.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("/<underlying>@optionMarkPrice", &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#"[{"s":"BTC-251120-126000-C","mp":"770.543","E":1762867543321,"e":"markPrice","i":"104334.60217391","P":"0.000","bo":"0.000","ao":"900.000","bq":"0.0000","aq":"0.2000","b":"-1.0","a":"0.98161161","hl":"924.652","ll":"616.435","vo":"0.9408058","rf":"0.0","d":"0.11111964","t":"-164.26702615","g":"0.00001245","v":"30.63855919"},{"s":"BTC-251123-126000-C","mp":"1249.61","E":1762867543321,"e":"markPrice","i":"104334.60217391","P":"0.000","bo":"1200.000","ao":"1300.000","bq":"0.3000","aq":"0.6000","b":"0.92159033","a":"0.94461441","hl":"1499.533","ll":"999.688","vo":"0.93310237","rf":"0.0","d":"0.14869196","t":"-172.12148811","g":"0.00001326","v":"43.43627792"}]"#).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 = MarketApiClient::new(streams_base.clone());

            let id = 123456u32;

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

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

            let pairs: &[(&str, Option<String>)] = &[("id", id.map(|v| v.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("/!optionSymbol", &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, Some(StreamId::Number(123456u32)));
        });
    }

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

            let id = 123456u32;

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

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

            let pairs: &[(&str, Option<String>)] = &[
                ("id",
                        id.map(|v| v.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("/!optionSymbol", &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":"optionSymbol","E":1669356423908,"s":"BTC-250926-140000-C","ps":"BTCUSDT","qa":"USDT","d":"CALL","sp":"21000","dt":4133404800000,"u":1,"ot":1569398400000,"cs":"TRADING"}"#).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 = MarketApiClient::new(streams_base.clone());

            let id = 123456u32;

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

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

            let pairs: &[(&str, Option<String>)] = &[
                ("id",
                        id.map(|v| v.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("/!optionSymbol", &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":"optionSymbol","E":1669356423908,"s":"BTC-250926-140000-C","ps":"BTCUSDT","qa":"USDT","d":"CALL","sp":"21000","dt":4133404800000,"u":1,"ot":1569398400000,"cs":"TRADING"}"#).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 = MarketApiClient::new(streams_base.clone());

            let id = 123456u32;

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

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

            let pairs: &[(&str, Option<String>)] = &[
                ("expirationDate", Some(expiration_date.clone())),
                ("id", id.map(|v| v.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(
                "/underlying@optionOpenInterest@<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, Some(StreamId::Number(123456u32)));
        });
    }

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

            let id = 123456u32;

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

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

            let pairs: &[(&str, Option<String>)] = &[
                ("expirationDate",
                        Some(expiration_date.clone())
                ),
                ("id",
                        id.map(|v| v.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("/underlying@optionOpenInterest@<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 = MarketApiClient::new(streams_base.clone());

            let id = 123456u32;

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

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

            let pairs: &[(&str, Option<String>)] = &[
                ("expirationDate",
                        Some(expiration_date.clone())
                ),
                ("id",
                        id.map(|v| v.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("/underlying@optionOpenInterest@<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");
        });
    }
}
