binance_sdk/spot/websocket_streams/

mod.rs

/*
 * Binance Spot WebSocket Streams
 *
 * OpenAPI Specifications for the Binance Spot WebSocket Streams
 *
 * API documents:
 * - [Github web-socket-streams documentation file](https://github.com/binance/binance-spot-api-docs/blob/master/web-socket-streams.md)
 * - [General API information for web-socket-streams on website](https://developers.binance.com/docs/binance-spot-api-docs/web-socket-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 serde_json::Value;
use std::sync::{Arc, atomic::Ordering};
use tokio::spawn;

use crate::common::config::ConfigurationWebsocketStreams;
use crate::common::websocket::{
    Subscription, WebsocketBase, WebsocketStream, WebsocketStreams as WebsocketStreamsBase,
    create_stream_handler,
};
use crate::models::{StreamId, WebsocketEvent, WebsocketMode};

mod apis;
mod handle;
mod models;

pub use apis::*;
pub use handle::*;
pub use models::*;

const HAS_TIME_UNIT: bool = true;

pub struct WebsocketStreams {
    websocket_streams_base: Arc<WebsocketStreamsBase>,
    web_socket_streams_api_client: WebSocketStreamsApiClient,
}

impl WebsocketStreams {
    pub(crate) async fn connect(
        config: ConfigurationWebsocketStreams,
        streams: Vec<String>,
        mode: Option<WebsocketMode>,
    ) -> anyhow::Result<Self> {
        let mut cfg = config;
        if let Some(m) = mode {
            cfg.mode = m;
        }

        if !HAS_TIME_UNIT {
            cfg.time_unit = None;
        }

        let websocket_streams_base = WebsocketStreamsBase::new(cfg, vec![], vec![]);

        websocket_streams_base.clone().connect(streams).await?;

        Ok(Self {
            websocket_streams_base: websocket_streams_base.clone(),
            web_socket_streams_api_client: WebSocketStreamsApiClient::new(
                websocket_streams_base.clone(),
            ),
        })
    }

    /// Subscribes to WebSocket events with a provided callback function.
    ///
    /// # Arguments
    ///
    /// * `callback` - A mutable function that takes a `WebsocketEvent` and is `Send` and `'static`.
    ///
    /// # Returns
    ///
    /// A `Subscription` that can be used to manage the event subscription.
    ///
    /// # Examples
    ///
    ///
    /// let subscription = `websocket_streams.subscribe_on_ws_events(|event`| {
    ///     // Handle WebSocket event
    /// });
    ///
    pub fn subscribe_on_ws_events<F>(&self, callback: F) -> Subscription
    where
        F: FnMut(WebsocketEvent) + Send + 'static,
    {
        let base = Arc::clone(&self.websocket_streams_base);
        base.common.events.subscribe(callback)
    }

    /// Unsubscribes from WebSocket events for a given `Subscription`.
    ///
    /// # Arguments
    ///
    /// * `subscription` - The `Subscription` to unsubscribe from WebSocket events.
    ///
    /// # Examples
    ///
    ///
    /// let subscription = `websocket_streams.subscribe_on_ws_events(|event`| {
    ///     // Handle WebSocket event
    /// });
    /// `websocket_streams.unsubscribe_from_ws_events(subscription)`;
    ///
    pub fn unsubscribe_from_ws_events(&self, subscription: Subscription) {
        subscription.unsubscribe();
    }

    /// Disconnects the WebSocket connection.
    ///
    /// # Returns
    ///
    /// A `Result` indicating whether the disconnection was successful.
    /// Returns an error if the disconnection fails.
    ///
    /// # Errors
    ///
    /// Returns an [`anyhow::Error`] if the connection fails.
    ///
    /// # Examples
    ///
    ///
    /// let `websocket_streams` = `WebSocketStreams::new`(...);
    /// `websocket_streams.disconnect().await`?;
    ///
    pub async fn disconnect(&self) -> anyhow::Result<()> {
        self.websocket_streams_base
            .disconnect()
            .await
            .map_err(anyhow::Error::msg)
    }

    /// Checks if the WebSocket connection is currently active.
    ///
    /// # Returns
    ///
    /// A `bool` indicating whether the WebSocket connection is established and connected.
    ///
    /// # Examples
    ///
    ///
    /// let `is_active` = `websocket_streams.is_connected().await`;
    /// if `is_active` {
    ///     // WebSocket connection is active
    /// }
    ///
    pub async fn is_connected(&self) -> bool {
        self.websocket_streams_base.is_connected().await
    }

    /// Sends a ping to the WebSocket server to maintain the connection.
    ///
    /// # Examples
    ///
    ///
    /// `websocket_streams.ping_server().await`;
    ///
    ///
    /// This method sends a ping request to the WebSocket server to keep the connection alive
    /// and check the server's responsiveness.
    pub async fn ping_server(&self) {
        self.websocket_streams_base.ping_server().await;
    }

    /// Subscribes to specified WebSocket streams.
    ///
    /// # Arguments
    ///
    /// * `streams` - A vector of stream names to subscribe to
    /// * `id` - An optional identifier for the subscription request
    ///
    /// # Examples
    ///
    ///
    /// `websocket_streams.subscribe(vec`!["`btcusdt@trade".to_string()`], None).await;
    ///
    ///
    /// This method initiates an asynchronous subscription to the specified WebSocket streams.
    /// The subscription is performed in a separate task using `spawn`.
    pub fn subscribe(&self, streams: Vec<String>, id: Option<String>) {
        let base = Arc::clone(&self.websocket_streams_base);
        spawn(async move { base.subscribe(streams, id.map(StreamId::from), None).await });
    }

    /// Unsubscribes from specified WebSocket streams.
    ///
    /// # Arguments
    ///
    /// * `streams` - A vector of stream names to unsubscribe from
    /// * `id` - An optional identifier for the unsubscription request
    ///
    /// # Examples
    ///
    ///
    /// `websocket_streams.unsubscribe(vec`!["`btcusdt@trade".to_string()`], None).await;
    ///
    ///
    /// This method initiates an asynchronous unsubscription from the specified WebSocket streams.
    /// The unsubscription is performed in a separate task using `spawn`.
    pub fn unsubscribe(&self, streams: Vec<String>, id: Option<String>) {
        let base = Arc::clone(&self.websocket_streams_base);
        spawn(async move {
            base.unsubscribe(streams, id.map(StreamId::from), None)
                .await;
        });
    }

    /// Checks if the current WebSocket stream is subscribed to a specific stream.
    ///
    /// # Arguments
    ///
    /// * `stream` - The name of the stream to check for subscription
    ///
    /// # Returns
    ///
    /// A boolean indicating whether the stream is currently subscribed
    ///
    /// # Examples
    ///
    ///
    /// let `is_subscribed` = `websocket_streams.is_subscribed("btcusdt@trade").await`;
    ///
    ///
    /// This method checks the subscription status of a specific WebSocket stream.
    pub async fn is_subscribed(&self, stream: &str) -> bool {
        self.websocket_streams_base.is_subscribed(stream).await
    }

    /// WebSocket Aggregate Trade Streams
    ///
    /// The Aggregate Trade Streams push trade information that is aggregated for a single taker order.
    ///
    /// # Arguments
    ///
    /// - `params`: [`AggTradeParams`]
    ///   The parameters for this operation.
    ///
    /// # Returns
    ///
    /// [`Arc<WebsocketStream<models::AggTradeResponse>>`] on success.
    ///
    /// # Errors
    ///
    /// Returns an [`anyhow::Error`] if the stream request fails, if parameters are invalid, or if parsing the response fails.
    ///
    ///
    /// For full API details, see the [Binance API Documentation](https://developers.binance.com/docs/binance-spot-api-docs/web-socket-streams#aggregate-trade-streams).
    ///
    pub async fn agg_trade(
        &self,
        params: AggTradeParams,
    ) -> anyhow::Result<Arc<WebsocketStream<models::AggTradeResponse>>> {
        self.web_socket_streams_api_client.agg_trade(params).await
    }

    /// WebSocket All Market Rolling Window Statistics Streams
    ///
    /// Rolling window ticker statistics for all market symbols, computed over multiple windows.
    /// Note that only tickers that have changed will be present in the array.
    ///
    /// # Arguments
    ///
    /// - `params`: [`AllMarketRollingWindowTickerParams`]
    ///   The parameters for this operation.
    ///
    /// # Returns
    ///
    /// [`Arc<WebsocketStream<Vec<models::AllMarketRollingWindowTickerResponseInner>>>`] on success.
    ///
    /// # Errors
    ///
    /// Returns an [`anyhow::Error`] if the stream request fails, if parameters are invalid, or if parsing the response fails.
    ///
    ///
    /// For full API details, see the [Binance API Documentation](https://developers.binance.com/docs/binance-spot-api-docs/web-socket-streams#all-market-rolling-window-statistics-streams).
    ///
    pub async fn all_market_rolling_window_ticker(
        &self,
        params: AllMarketRollingWindowTickerParams,
    ) -> anyhow::Result<Arc<WebsocketStream<Vec<models::AllMarketRollingWindowTickerResponseInner>>>>
    {
        self.web_socket_streams_api_client
            .all_market_rolling_window_ticker(params)
            .await
    }

    /// WebSocket All Market Mini Tickers Stream
    ///
    /// 24hr rolling window mini-ticker statistics for all symbols that changed in an array. These are NOT the statistics of the UTC day, but a 24hr rolling window for the previous 24hrs. Note that only tickers that have changed will be present in the array.
    ///
    /// # Arguments
    ///
    /// - `params`: [`AllMiniTickerParams`]
    ///   The parameters for this operation.
    ///
    /// # Returns
    ///
    /// [`Arc<WebsocketStream<Vec<models::AllMiniTickerResponseInner>>>`] on success.
    ///
    /// # Errors
    ///
    /// Returns an [`anyhow::Error`] if the stream request fails, if parameters are invalid, or if parsing the response fails.
    ///
    ///
    /// For full API details, see the [Binance API Documentation](https://developers.binance.com/docs/binance-spot-api-docs/web-socket-streams#all-market-mini-tickers-stream).
    ///
    pub async fn all_mini_ticker(
        &self,
        params: AllMiniTickerParams,
    ) -> anyhow::Result<Arc<WebsocketStream<Vec<models::AllMiniTickerResponseInner>>>> {
        self.web_socket_streams_api_client
            .all_mini_ticker(params)
            .await
    }

    /// WebSocket Average Price
    ///
    /// Average price streams push changes in the average price over a fixed time interval.
    ///
    /// # Arguments
    ///
    /// - `params`: [`AvgPriceParams`]
    ///   The parameters for this operation.
    ///
    /// # Returns
    ///
    /// [`Arc<WebsocketStream<models::AvgPriceResponse>>`] on success.
    ///
    /// # Errors
    ///
    /// Returns an [`anyhow::Error`] if the stream request fails, if parameters are invalid, or if parsing the response fails.
    ///
    ///
    /// For full API details, see the [Binance API Documentation](https://developers.binance.com/docs/binance-spot-api-docs/web-socket-streams#average-price).
    ///
    pub async fn avg_price(
        &self,
        params: AvgPriceParams,
    ) -> anyhow::Result<Arc<WebsocketStream<models::AvgPriceResponse>>> {
        self.web_socket_streams_api_client.avg_price(params).await
    }

    /// WebSocket Block Trade Streams
    ///
    ///
    ///
    /// # Arguments
    ///
    /// - `params`: [`BlockTradeParams`]
    ///   The parameters for this operation.
    ///
    /// # Returns
    ///
    /// [`Arc<WebsocketStream<models::BlockTradeResponse>>`] on success.
    ///
    /// # Errors
    ///
    /// Returns an [`anyhow::Error`] if the stream request fails, if parameters are invalid, or if parsing the response fails.
    ///
    ///
    /// For full API details, see the [Binance API Documentation](https://developers.binance.com/docs/binance-spot-api-docs/web-socket-streams#block-trade-streams).
    ///
    pub async fn block_trade(
        &self,
        params: BlockTradeParams,
    ) -> anyhow::Result<Arc<WebsocketStream<models::BlockTradeResponse>>> {
        self.web_socket_streams_api_client.block_trade(params).await
    }

    /// WebSocket Individual Symbol Book Ticker Streams
    ///
    /// Pushes any update to the best bid or ask's price or quantity in real-time for a specified symbol.
    /// Multiple `<symbol>@bookTicker` streams can be subscribed to over one connection.
    ///
    /// # Arguments
    ///
    /// - `params`: [`BookTickerParams`]
    ///   The parameters for this operation.
    ///
    /// # Returns
    ///
    /// [`Arc<WebsocketStream<models::BookTickerResponse>>`] on success.
    ///
    /// # Errors
    ///
    /// Returns an [`anyhow::Error`] if the stream request fails, if parameters are invalid, or if parsing the response fails.
    ///
    ///
    /// For full API details, see the [Binance API Documentation](https://developers.binance.com/docs/binance-spot-api-docs/web-socket-streams#individual-symbol-book-ticker-streams).
    ///
    pub async fn book_ticker(
        &self,
        params: BookTickerParams,
    ) -> anyhow::Result<Arc<WebsocketStream<models::BookTickerResponse>>> {
        self.web_socket_streams_api_client.book_ticker(params).await
    }

    /// WebSocket Diff. Depth Stream
    ///
    /// Order book price and quantity depth updates used to locally manage an order book.
    ///
    /// # Arguments
    ///
    /// - `params`: [`DiffBookDepthParams`]
    ///   The parameters for this operation.
    ///
    /// # Returns
    ///
    /// [`Arc<WebsocketStream<models::DiffBookDepthResponse>>`] on success.
    ///
    /// # Errors
    ///
    /// Returns an [`anyhow::Error`] if the stream request fails, if parameters are invalid, or if parsing the response fails.
    ///
    ///
    /// For full API details, see the [Binance API Documentation](https://developers.binance.com/docs/binance-spot-api-docs/web-socket-streams#diff-depth-stream).
    ///
    pub async fn diff_book_depth(
        &self,
        params: DiffBookDepthParams,
    ) -> anyhow::Result<Arc<WebsocketStream<models::DiffBookDepthResponse>>> {
        self.web_socket_streams_api_client
            .diff_book_depth(params)
            .await
    }

    /// WebSocket Kline/Candlestick Streams for UTC
    ///
    /// The Kline/Candlestick Stream push updates to the current klines/candlestick every second in `UTC+0` timezone
    ///
    /// <a id="kline-intervals"></a>
    ///
    /// # Arguments
    ///
    /// - `params`: [`KlineParams`]
    ///   The parameters for this operation.
    ///
    /// # Returns
    ///
    /// [`Arc<WebsocketStream<models::KlineResponse>>`] on success.
    ///
    /// # Errors
    ///
    /// Returns an [`anyhow::Error`] if the stream request fails, if parameters are invalid, or if parsing the response fails.
    ///
    ///
    /// For full API details, see the [Binance API Documentation](https://developers.binance.com/docs/binance-spot-api-docs/web-socket-streams#klinecandlestick-streams-for-utc).
    ///
    pub async fn kline(
        &self,
        params: KlineParams,
    ) -> anyhow::Result<Arc<WebsocketStream<models::KlineResponse>>> {
        self.web_socket_streams_api_client.kline(params).await
    }

    /// WebSocket Kline/Candlestick Streams with timezone offset
    ///
    /// The Kline/Candlestick Stream push updates to the current klines/candlestick every second in `UTC+8` timezone
    ///
    /// # Arguments
    ///
    /// - `params`: [`KlineOffsetParams`]
    ///   The parameters for this operation.
    ///
    /// # Returns
    ///
    /// [`Arc<WebsocketStream<models::KlineOffsetResponse>>`] on success.
    ///
    /// # Errors
    ///
    /// Returns an [`anyhow::Error`] if the stream request fails, if parameters are invalid, or if parsing the response fails.
    ///
    ///
    /// For full API details, see the [Binance API Documentation](https://developers.binance.com/docs/binance-spot-api-docs/web-socket-streams#klinecandlestick-streams-with-timezone-offset).
    ///
    pub async fn kline_offset(
        &self,
        params: KlineOffsetParams,
    ) -> anyhow::Result<Arc<WebsocketStream<models::KlineOffsetResponse>>> {
        self.web_socket_streams_api_client
            .kline_offset(params)
            .await
    }

    /// WebSocket Individual Symbol Mini Ticker Stream
    ///
    /// 24hr rolling window mini-ticker statistics. These are NOT the statistics of the UTC day, but a 24hr rolling window for the previous 24hrs.
    ///
    /// # Arguments
    ///
    /// - `params`: [`MiniTickerParams`]
    ///   The parameters for this operation.
    ///
    /// # Returns
    ///
    /// [`Arc<WebsocketStream<models::MiniTickerResponse>>`] on success.
    ///
    /// # Errors
    ///
    /// Returns an [`anyhow::Error`] if the stream request fails, if parameters are invalid, or if parsing the response fails.
    ///
    ///
    /// For full API details, see the [Binance API Documentation](https://developers.binance.com/docs/binance-spot-api-docs/web-socket-streams#individual-symbol-mini-ticker-stream).
    ///
    pub async fn mini_ticker(
        &self,
        params: MiniTickerParams,
    ) -> anyhow::Result<Arc<WebsocketStream<models::MiniTickerResponse>>> {
        self.web_socket_streams_api_client.mini_ticker(params).await
    }

    /// WebSocket Partial Book Depth Streams
    ///
    /// Top **\<levels\>** bids and asks, pushed every second. Valid **\<levels\>** are 5, 10, or 20.
    ///
    /// # Arguments
    ///
    /// - `params`: [`PartialBookDepthParams`]
    ///   The parameters for this operation.
    ///
    /// # Returns
    ///
    /// [`Arc<WebsocketStream<models::PartialBookDepthResponse>>`] on success.
    ///
    /// # Errors
    ///
    /// Returns an [`anyhow::Error`] if the stream request fails, if parameters are invalid, or if parsing the response fails.
    ///
    ///
    /// For full API details, see the [Binance API Documentation](https://developers.binance.com/docs/binance-spot-api-docs/web-socket-streams#partial-book-depth-streams).
    ///
    pub async fn partial_book_depth(
        &self,
        params: PartialBookDepthParams,
    ) -> anyhow::Result<Arc<WebsocketStream<models::PartialBookDepthResponse>>> {
        self.web_socket_streams_api_client
            .partial_book_depth(params)
            .await
    }

    /// WebSocket Reference Price Streams
    ///
    ///
    ///
    /// # Arguments
    ///
    /// - `params`: [`ReferencePriceParams`]
    ///   The parameters for this operation.
    ///
    /// # Returns
    ///
    /// [`Arc<WebsocketStream<models::ReferencePriceResponse>>`] on success.
    ///
    /// # Errors
    ///
    /// Returns an [`anyhow::Error`] if the stream request fails, if parameters are invalid, or if parsing the response fails.
    ///
    ///
    /// For full API details, see the [Binance API Documentation](https://developers.binance.com/docs/binance-spot-api-docs/web-socket-streams#reference-price-streams).
    ///
    pub async fn reference_price(
        &self,
        params: ReferencePriceParams,
    ) -> anyhow::Result<Arc<WebsocketStream<models::ReferencePriceResponse>>> {
        self.web_socket_streams_api_client
            .reference_price(params)
            .await
    }

    /// WebSocket Individual Symbol Rolling Window Statistics Streams
    ///
    /// Rolling window ticker statistics for a single symbol, computed over multiple windows.
    ///
    /// # Arguments
    ///
    /// - `params`: [`RollingWindowTickerParams`]
    ///   The parameters for this operation.
    ///
    /// # Returns
    ///
    /// [`Arc<WebsocketStream<models::RollingWindowTickerResponse>>`] on success.
    ///
    /// # Errors
    ///
    /// Returns an [`anyhow::Error`] if the stream request fails, if parameters are invalid, or if parsing the response fails.
    ///
    ///
    /// For full API details, see the [Binance API Documentation](https://developers.binance.com/docs/binance-spot-api-docs/web-socket-streams#individual-symbol-rolling-window-statistics-streams).
    ///
    pub async fn rolling_window_ticker(
        &self,
        params: RollingWindowTickerParams,
    ) -> anyhow::Result<Arc<WebsocketStream<models::RollingWindowTickerResponse>>> {
        self.web_socket_streams_api_client
            .rolling_window_ticker(params)
            .await
    }

    /// WebSocket Individual Symbol Ticker Streams
    ///
    /// 24hr rolling window ticker statistics for a single symbol. These are NOT the statistics of the UTC day, but a 24hr rolling window for the previous 24hrs.
    ///
    /// # Arguments
    ///
    /// - `params`: [`TickerParams`]
    ///   The parameters for this operation.
    ///
    /// # Returns
    ///
    /// [`Arc<WebsocketStream<models::TickerResponse>>`] on success.
    ///
    /// # Errors
    ///
    /// Returns an [`anyhow::Error`] if the stream request fails, if parameters are invalid, or if parsing the response fails.
    ///
    ///
    /// For full API details, see the [Binance API Documentation](https://developers.binance.com/docs/binance-spot-api-docs/web-socket-streams#individual-symbol-ticker-streams).
    ///
    pub async fn ticker(
        &self,
        params: TickerParams,
    ) -> anyhow::Result<Arc<WebsocketStream<models::TickerResponse>>> {
        self.web_socket_streams_api_client.ticker(params).await
    }

    /// WebSocket Trade Streams
    ///
    /// The Trade Streams push raw trade information; each trade has a unique buyer and seller.
    ///
    /// # Arguments
    ///
    /// - `params`: [`TradeParams`]
    ///   The parameters for this operation.
    ///
    /// # Returns
    ///
    /// [`Arc<WebsocketStream<models::TradeResponse>>`] on success.
    ///
    /// # Errors
    ///
    /// Returns an [`anyhow::Error`] if the stream request fails, if parameters are invalid, or if parsing the response fails.
    ///
    ///
    /// For full API details, see the [Binance API Documentation](https://developers.binance.com/docs/binance-spot-api-docs/web-socket-streams#trade-streams).
    ///
    pub async fn trade(
        &self,
        params: TradeParams,
    ) -> anyhow::Result<Arc<WebsocketStream<models::TradeResponse>>> {
        self.web_socket_streams_api_client.trade(params).await
    }
}