binance-sdk 69.2.2

/*
 * Stocks Trading WebSocket Streams
 *
 * WebSocket stream definitions for Binance Stocks Trading. Base URL: wss://nbstream.binance.com/equity
 *
 * 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::models::StreamId;
use crate::stocks::websocket_streams::models;

#[async_trait]
pub trait MarketStreamsApi: Send + Sync {
    async fn calendar_stream(
        &self,
        params: CalendarStreamParams,
    ) -> anyhow::Result<Arc<WebsocketStream<models::CalendarStreamResponse>>>;
    async fn kline_stream(
        &self,
        params: KlineStreamParams,
    ) -> anyhow::Result<Arc<WebsocketStream<models::KlineStreamResponse>>>;
    async fn price_stream(
        &self,
        params: PriceStreamParams,
    ) -> anyhow::Result<Arc<WebsocketStream<models::PriceStreamResponse>>>;
    async fn quote_stream(
        &self,
        params: QuoteStreamParams,
    ) -> anyhow::Result<Arc<WebsocketStream<models::QuoteStreamResponse>>>;
    async fn tradability_stream(
        &self,
        params: TradabilityStreamParams,
    ) -> anyhow::Result<Arc<WebsocketStream<models::TradabilityStreamResponse>>>;
    async fn trading_status_stream(
        &self,
        params: TradingStatusStreamParams,
    ) -> anyhow::Result<Arc<WebsocketStream<models::TradingStatusStreamResponse>>>;
}

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

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

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

impl CalendarStreamParams {
    /// Create a builder for [`calendar_stream`].
    ///
    #[must_use]
    pub fn builder() -> CalendarStreamParamsBuilder {
        CalendarStreamParamsBuilder::default()
    }
}
/// Request parameters for the [`kline_stream`] operation.
///
/// This struct holds all of the inputs you can pass when calling
/// [`kline_stream`](#method.kline_stream).
#[derive(Clone, Debug, Builder, Deserialize)]
#[builder(pattern = "owned", build_fn(error = "ParamBuildError"))]
pub struct KlineStreamParams {
    /// US-equity ticker (UPPERCASE), e.g. `AAPL`.
    ///
    /// This field is **required.
    #[builder(setter(into))]
    #[serde(rename = "symbol")]
    pub symbol: String,
    /// Kline interval — `5m` / `1h` / `1d` / `1w` / `1M`.
    ///
    /// This field is **required.
    #[builder(setter(into))]
    #[serde(rename = "interval")]
    pub interval: String,
    /// Unique WebSocket request ID.
    ///
    /// This field is **optional.
    #[builder(setter(into), default)]
    #[serde(rename = "id", default)]
    pub id: Option<String>,
}

impl KlineStreamParams {
    /// Create a builder for [`kline_stream`].
    ///
    /// Required parameters:
    ///
    /// * `symbol` — US-equity ticker (UPPERCASE), e.g. `AAPL`.
    /// * `interval` — Kline interval — `5m` / `1h` / `1d` / `1w` / `1M`.
    ///
    #[must_use]
    pub fn builder(symbol: String, interval: String) -> KlineStreamParamsBuilder {
        KlineStreamParamsBuilder::default()
            .symbol(symbol)
            .interval(interval)
    }
}
/// Request parameters for the [`price_stream`] operation.
///
/// This struct holds all of the inputs you can pass when calling
/// [`price_stream`](#method.price_stream).
#[derive(Clone, Debug, Builder, Deserialize, Default)]
#[builder(pattern = "owned", build_fn(error = "ParamBuildError"))]
pub struct PriceStreamParams {
    /// Unique WebSocket request ID.
    ///
    /// This field is **optional.
    #[builder(setter(into), default)]
    #[serde(rename = "id", default)]
    pub id: Option<String>,
}

impl PriceStreamParams {
    /// Create a builder for [`price_stream`].
    ///
    #[must_use]
    pub fn builder() -> PriceStreamParamsBuilder {
        PriceStreamParamsBuilder::default()
    }
}
/// Request parameters for the [`quote_stream`] operation.
///
/// This struct holds all of the inputs you can pass when calling
/// [`quote_stream`](#method.quote_stream).
#[derive(Clone, Debug, Builder, Deserialize)]
#[builder(pattern = "owned", build_fn(error = "ParamBuildError"))]
pub struct QuoteStreamParams {
    /// US-equity ticker (UPPERCASE), e.g. `AAPL`.
    ///
    /// This field is **required.
    #[builder(setter(into))]
    #[serde(rename = "symbol")]
    pub symbol: String,
    /// Unique WebSocket request ID.
    ///
    /// This field is **optional.
    #[builder(setter(into), default)]
    #[serde(rename = "id", default)]
    pub id: Option<String>,
}

impl QuoteStreamParams {
    /// Create a builder for [`quote_stream`].
    ///
    /// Required parameters:
    ///
    /// * `symbol` — US-equity ticker (UPPERCASE), e.g. `AAPL`.
    ///
    #[must_use]
    pub fn builder(symbol: String) -> QuoteStreamParamsBuilder {
        QuoteStreamParamsBuilder::default().symbol(symbol)
    }
}
/// Request parameters for the [`tradability_stream`] operation.
///
/// This struct holds all of the inputs you can pass when calling
/// [`tradability_stream`](#method.tradability_stream).
#[derive(Clone, Debug, Builder, Deserialize)]
#[builder(pattern = "owned", build_fn(error = "ParamBuildError"))]
pub struct TradabilityStreamParams {
    /// US-equity ticker (UPPERCASE), e.g. `AAPL`.
    ///
    /// This field is **required.
    #[builder(setter(into))]
    #[serde(rename = "symbol")]
    pub symbol: String,
    /// Unique WebSocket request ID.
    ///
    /// This field is **optional.
    #[builder(setter(into), default)]
    #[serde(rename = "id", default)]
    pub id: Option<String>,
}

impl TradabilityStreamParams {
    /// Create a builder for [`tradability_stream`].
    ///
    /// Required parameters:
    ///
    /// * `symbol` — US-equity ticker (UPPERCASE), e.g. `AAPL`.
    ///
    #[must_use]
    pub fn builder(symbol: String) -> TradabilityStreamParamsBuilder {
        TradabilityStreamParamsBuilder::default().symbol(symbol)
    }
}
/// Request parameters for the [`trading_status_stream`] operation.
///
/// This struct holds all of the inputs you can pass when calling
/// [`trading_status_stream`](#method.trading_status_stream).
#[derive(Clone, Debug, Builder, Deserialize)]
#[builder(pattern = "owned", build_fn(error = "ParamBuildError"))]
pub struct TradingStatusStreamParams {
    /// US-equity ticker (UPPERCASE), e.g. `AAPL`.
    ///
    /// This field is **required.
    #[builder(setter(into))]
    #[serde(rename = "symbol")]
    pub symbol: String,
    /// Unique WebSocket request ID.
    ///
    /// This field is **optional.
    #[builder(setter(into), default)]
    #[serde(rename = "id", default)]
    pub id: Option<String>,
}

impl TradingStatusStreamParams {
    /// Create a builder for [`trading_status_stream`].
    ///
    /// Required parameters:
    ///
    /// * `symbol` — US-equity ticker (UPPERCASE), e.g. `AAPL`.
    ///
    #[must_use]
    pub fn builder(symbol: String) -> TradingStatusStreamParamsBuilder {
        TradingStatusStreamParamsBuilder::default().symbol(symbol)
    }
}

#[async_trait]
impl MarketStreamsApi for MarketStreamsApiClient {
    async fn calendar_stream(
        &self,
        params: CalendarStreamParams,
    ) -> anyhow::Result<Arc<WebsocketStream<models::CalendarStreamResponse>>> {
        let CalendarStreamParams { 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("/calendar", &vars);

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

    async fn kline_stream(
        &self,
        params: KlineStreamParams,
    ) -> anyhow::Result<Arc<WebsocketStream<models::KlineStreamResponse>>> {
        let KlineStreamParams {
            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::KlineStreamResponse>(
            WebsocketBase::WebsocketStreams(Arc::clone(&self.websocket_streams_base)),
            stream,
            id_opt.map(|s| {
                if !s.is_empty() && s.bytes().all(|b| b.is_ascii_digit()) {
                    if let Ok(n) = s.parse::<u32>() {
                        return StreamId::Number(n);
                    }
                }
                StreamId::Str(s)
            }),
            None,
        )
        .await)
    }

    async fn price_stream(
        &self,
        params: PriceStreamParams,
    ) -> anyhow::Result<Arc<WebsocketStream<models::PriceStreamResponse>>> {
        let PriceStreamParams { 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("/price", &vars);

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

    async fn quote_stream(
        &self,
        params: QuoteStreamParams,
    ) -> anyhow::Result<Arc<WebsocketStream<models::QuoteStreamResponse>>> {
        let QuoteStreamParams { 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>@quote", &vars);

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

    async fn tradability_stream(
        &self,
        params: TradabilityStreamParams,
    ) -> anyhow::Result<Arc<WebsocketStream<models::TradabilityStreamResponse>>> {
        let TradabilityStreamParams { 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>@tradability", &vars);

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

    async fn trading_status_stream(
        &self,
        params: TradingStatusStreamParams,
    ) -> anyhow::Result<Arc<WebsocketStream<models::TradingStatusStreamResponse>>> {
        let TradingStatusStreamParams { 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>@tradingStatus", &vars);

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

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

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

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

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

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

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

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

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

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

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

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

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

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

            let payload: Value = serde_json::from_str(
                r#"{"e":"calendar","from":"PRE_MARKET","to":"MARKET_OPEN","ts":1710334200000}"#,
            )
            .unwrap_or_else(|_| serde_json::json!({}));
            let msg = json!({
                "stream": stream,
                "data": payload,
            });

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

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

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

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

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

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

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

            let called = Arc::new(AtomicBool::new(false));
            let called_clone = called.clone();
            ws_stream.on_message(move |_payload: models::CalendarStreamResponse| {
                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":"calendar","from":"PRE_MARKET","to":"MARKET_OPEN","ts":1710334200000}"#,
            )
            .unwrap_or_else(|_| serde_json::json!({}));
            let msg = json!({
                "stream": stream,
                "data": payload,
            });

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

            yield_now().await;

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

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

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

            let params = KlineStreamParams::builder("AAPL".to_string(), "5m".to_string())
                .id(Some(id.clone()))
                .build()
                .unwrap();

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

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

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

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

            let params = KlineStreamParams::builder("AAPL".to_string(),"5m".to_string(),).id(Some(id.clone())).build().unwrap();

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

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

            let payload: Value = serde_json::from_str(r#"{"e":"kline","E":1710320400000,"s":"AAPL","k":{"t":1710316800000,"ct":1710317099999,"s":"AAPL","i":"5m","o":"182.50","c":"183.20","h":"183.50","l":"182.30","v":"15000","n":120,"x":false,"vw":"182.95"}}"#).unwrap_or_else(|_| serde_json::json!({}));
            let msg = json!({
                "stream": stream,
                "data": payload,
            });

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

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

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

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

            let params = KlineStreamParams::builder("AAPL".to_string(),"5m".to_string(),).id(Some(id.clone())).build().unwrap();

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

            let called = Arc::new(AtomicBool::new(false));
            let called_clone = called.clone();
            ws_stream.on_message(move |_payload: models::KlineStreamResponse| {
                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":1710320400000,"s":"AAPL","k":{"t":1710316800000,"ct":1710317099999,"s":"AAPL","i":"5m","o":"182.50","c":"183.20","h":"183.50","l":"182.30","v":"15000","n":120,"x":false,"vw":"182.95"}}"#).unwrap_or_else(|_| serde_json::json!({}));
            let msg = json!({
                "stream": stream,
                "data": payload,
            });

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

            yield_now().await;

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

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

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

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

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

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

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

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

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

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

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

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

            let payload: Value = serde_json::from_str(r#"{"e":"price","rates":[{"s":"AAPL","ac":"EQ_AAPL","p":"182.5","t":1710320400000,"pc":"181.2000","tc":"182.0000","mp":"O"}]}"#).unwrap_or_else(|_| serde_json::json!({}));
            let msg = json!({
                "stream": stream,
                "data": payload,
            });

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

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

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

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

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

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

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

            let called = Arc::new(AtomicBool::new(false));
            let called_clone = called.clone();
            ws_stream.on_message(move |_payload: models::PriceStreamResponse| {
                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":"price","rates":[{"s":"AAPL","ac":"EQ_AAPL","p":"182.5","t":1710320400000,"pc":"181.2000","tc":"182.0000","mp":"O"}]}"#).unwrap_or_else(|_| serde_json::json!({}));
            let msg = json!({
                "stream": stream,
                "data": payload,
            });

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

            yield_now().await;

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

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

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

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

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

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

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

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

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

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

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

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

            let payload: Value = serde_json::from_str(r#"{"e":"quote","E":1710320400000,"s":"AAPL","bp":"182.50","ap":"182.55","bs":100,"as":200,"T":1710320399500}"#).unwrap_or_else(|_| serde_json::json!({}));
            let msg = json!({
                "stream": stream,
                "data": payload,
            });

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

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

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

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

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

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

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

            let called = Arc::new(AtomicBool::new(false));
            let called_clone = called.clone();
            ws_stream.on_message(move |_payload: models::QuoteStreamResponse| {
                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":"quote","E":1710320400000,"s":"AAPL","bp":"182.50","ap":"182.55","bs":100,"as":200,"T":1710320399500}"#).unwrap_or_else(|_| serde_json::json!({}));
            let msg = json!({
                "stream": stream,
                "data": payload,
            });

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

            yield_now().await;

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

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

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

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

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

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

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

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

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

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

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

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

            let payload: Value = serde_json::from_str(r#"{"e":"tradability","symbol":"AAPL","assetCode":"EQ_AAPL","tradability":"BUY_SELL","t":1712345678000}"#).unwrap_or_else(|_| serde_json::json!({}));
            let msg = json!({
                "stream": stream,
                "data": payload,
            });

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

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

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

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

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

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

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

            let called = Arc::new(AtomicBool::new(false));
            let called_clone = called.clone();
            ws_stream.on_message(move |_payload: models::TradabilityStreamResponse| {
                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":"tradability","symbol":"AAPL","assetCode":"EQ_AAPL","tradability":"BUY_SELL","t":1712345678000}"#).unwrap_or_else(|_| serde_json::json!({}));
            let msg = json!({
                "stream": stream,
                "data": payload,
            });

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

            yield_now().await;

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

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

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

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

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

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

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

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

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

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

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

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

            let payload: Value = serde_json::from_str(r#"{"e":"tradingStatus","symbol":"AAPL","assetCode":"EQ_AAPL","status":"TRADING_HALTED","msg":"MATERIAL_NEWS","time":1614021300000,"z":"C","tradability":"NONE"}"#).unwrap_or_else(|_| serde_json::json!({}));
            let msg = json!({
                "stream": stream,
                "data": payload,
            });

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

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

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

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

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

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

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

            let called = Arc::new(AtomicBool::new(false));
            let called_clone = called.clone();
            ws_stream.on_message(move |_payload: models::TradingStatusStreamResponse| {
                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":"tradingStatus","symbol":"AAPL","assetCode":"EQ_AAPL","status":"TRADING_HALTED","msg":"MATERIAL_NEWS","time":1614021300000,"z":"C","tradability":"NONE"}"#).unwrap_or_else(|_| serde_json::json!({}));
            let msg = json!({
                "stream": stream,
                "data": payload,
            });

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

            yield_now().await;

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